What Is a Construction Site Actually Doing Right Now? TU Berlin Built a System That Reads Site Images With an Ontology and a GPT — No Training Required

Complete End-to-End Implementation (Python)

The implementation covers all components from the paper in 10 sections: the ConSE ontology data structures, 3×3 grid spatial encoder, scene context serializer producing ontology-grounded triples, task formulation and output formulation prompt builders, in-context example template, Chain-of-Thought interaction classifier, the full prompt assembler combining all components, OpenAI API client wrapper, evaluation metrics (Activity Accuracy, Activity+Elements Accuracy, Redundancy Rate, Miss Rate), and a complete end-to-end demo with synthetic construction scenes.

# ==============================================================================
# Ontology-Based LLM Prompting for Construction Activity Recognition
# Paper: Advanced Engineering Informatics 69 (2026) 103869
# Authors: Cheng Zeng, Timo Hartmann, Leyuan Ma
# Affiliation: TU Berlin / Qingdao University of Technology
# ==============================================================================
# Sections:
#   1.  Imports & ConSE Ontology Data Structures
#   2.  3×3 Grid Spatial Encoder
#   3.  Scene Context Serializer (ontology-grounded triples)
#   4.  Task Formulation Prompt Builder
#   5.  Output Formulation Prompt Builder
#   6.  In-Context Example Template
#   7.  Chain-of-Thought Interaction Classifier
#   8.  Full Prompt Assembler
#   9.  Evaluation Metrics (Activity Accuracy, A+E, Redundancy, Miss)
#  10.  End-to-End Demo with Synthetic Scenes
# ==============================================================================

from __future__ import annotations
import json, re, warnings
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from collections import defaultdict
warnings.filterwarnings("ignore")


# ─── SECTION 1: ConSE Ontology Data Structures ────────────────────────────────

# ConSE-defined entity classes (Section 4.1)
CONSE_ENTITY_CLASSES = [
    "Worker", "Excavator", "Crane", "Hoist", "ConcretePump", "Bulldozer",
    "Scaffolding", "Formwork", "ReinforcementRegion", "ConcreteRegion",
    "SoilRegion", "SteelRegion", "MasonryRegion", "WoodRegion",
    "ConcretePumpHose", "ShovelTool", "DrillTool", "Column",
    "FormworkRegion", "SoilGroundRegion", "SoilTrenchRegion",
    "WoodenPlateRegion", "Controller",
]

# ConSE-defined construction activities (Level 2 — 41 types per paper)
CONSE_ACTIVITY_CLASSES = [
    # Concrete-related
    "ConcretePouring", "ConcreteLevelling", "ConcreteConsolidation",
    "ConcreteTransportation", "ConcreteMixing", "ConcreteProcessing",
    "ConcreteCuring",
    # Formwork-related
    "FormworkSetUp", "FormworkTransportation", "FormworkInspection",
    # Reinforcement-related
    "ReinforcementAssembly", "ReinforcementTransportation",
    "ReinforcementProcessing", "ReinforcementInspection",
    # Soil-related
    "SoilExcavation", "SoilTransportation",
    # Scaffolding-related
    "ScaffoldingSetUp",
    # Masonry-related
    "MasonryInstallation", "MasonryFinishing", "MasonryInspection",
    # Wood-related
    "WoodProcessing", "WoodInstallation", "WoodTransportation",
    # Steel-related
    "SteelProcessing", "SteelTransportation",
    # Wall-related
    "WallPainting",
    # MEP and drywall
    "MEPInstallation", "DrywallInstallation", "MetalStudFrameSetUp",
    # General
    "ProcessMonitoring", "AuxiliaryActivity",
]

# Interaction subclasses used as CoT components (Table 2)
INTERACTION_SUBCLASSES = {
    "WorkerBasicInteraction": [
        "installing", "monitoring", "assisting", "transporting",
        "assembling", "inspecting", "levelling", "painting",
    ],
    "ToolAssistedInteraction": [
        "drilling", "cutting", "grinding", "mixing", "processing",
    ],
    "MachineryInteraction": [
        "lifting", "concreting", "excavating", "bulldozing",
        "pumping", "transporting_heavy",
    ],
}


@dataclass
class Entity:
    """A detected entity in a construction image."""
    instance_id: str              # e.g., "Worker_1", "Hoist_2"
    entity_class: str             # ConSE class, e.g., "Worker", "Hoist"
    grid_positions: List[int]     # Occupied cells in the 3×3 grid (1–9)
    bounding_box: Optional[Tuple[float,float,float,float]] = None  # x1,y1,x2,y2 normalized


@dataclass
class EntityAttribute:
    """An attribute of an entity (e.g., posture, body part)."""
    subject: str                  # e.g., "Worker_1"
    predicate: str                # e.g., "hasPosture", "hasBodyPart"
    value: str                    # e.g., "Standing", "Worker_Hand_1"


@dataclass
class EntityRelationship:
    """A relationship triple between two entities."""
    subject: str                  # e.g., "Worker_1"
    predicate: str                # e.g., "hasAttentionOn", "isConnectedTo"
    obj: str                      # e.g., "Hoist_1", "ReinforcementRegion_2"


@dataclass
class ConstructionScene:
    """
    Complete symbolic representation of a construction image.
    This is the  fed to the LLM.
    """
    image_id: str
    entities: List[Entity] = field(default_factory=list)
    attributes: List[EntityAttribute] = field(default_factory=list)
    relationships: List[EntityRelationship] = field(default_factory=list)
    image_width: int = 1000
    image_height: int = 750


@dataclass
class ActivityPrediction:
    """A predicted construction activity with associated entities."""
    activity_type: str            # ConSE activity class
    participants: List[str]       # Entity instance IDs involved


@dataclass
class InContextExample:
    """A labeled example for in-context learning."""
    scene: ConstructionScene
    interactions: List[Dict]      # CoT intermediate reasoning
    activities: List[ActivityPrediction]  # Ground truth activities


# ─── SECTION 2: 3×3 Grid Spatial Encoder ─────────────────────────────────────

class GridSpatialEncoder:
    """
    Encodes entity spatial positions using a 3×3 grid partition.
    (Section 3.2 — Ontology-based scene context)

    Inspired by Vision Transformer patch decomposition, each image is
    partitioned into a 3×3 grid. Each entity's bounding box is mapped
    to the grid cell(s) it occupies, providing lightweight spatial
    context without exponentially complex pairwise distance encoding.

    Grid layout:
        1 | 2 | 3
        ---------
        4 | 5 | 6
        ---------
        7 | 8 | 9

    N=3 was chosen to balance simplicity (manageable context size)
    and granularity (sufficient spatial discrimination for activity
    understanding).
    """
    def __init__(self, grid_size: int = 3):
        self.grid_size = grid_size

    def encode(
        self,
        bbox: Tuple[float,float,float,float],  # x1,y1,x2,y2 normalized [0,1]
        img_w: int = 1, img_h: int = 1,
    ) -> List[int]:
        """
        Map a bounding box to the 3×3 grid cell indices it occupies.
        A large entity (e.g., excavator) may span multiple cells.
        
        Returns list of 1-indexed grid cell numbers.
        """
        x1, y1, x2, y2 = bbox
        N = self.grid_size
        cell_w = 1.0 / N
        cell_h = 1.0 / N

        cells = []
        for row in range(N):
            for col in range(N):
                # Cell boundaries
                cx1, cy1 = col * cell_w, row * cell_h
                cx2, cy2 = cx1 + cell_w, cy1 + cell_h
                # Check overlap between bbox and cell
                if x2 > cx1 and x1 < cx2 and y2 > cy1 and y1 < cy2:
                    cell_idx = row * N + col + 1    # 1-indexed
                    cells.append(cell_idx)
        return cells if cells else [5]   # Default to center if no overlap found

    def describe_position(self, cells: List[int]) -> str:
        """Human-readable description of grid position."""
        position_names = {
            1: "top-left", 2: "top-center", 3: "top-right",
            4: "middle-left", 5: "center", 6: "middle-right",
            7: "bottom-left", 8: "bottom-center", 9: "bottom-right",
        }
        return " and ".join(position_names.get(c, str(c)) for c in sorted(cells))


# ─── SECTION 3: Scene Context Serializer ──────────────────────────────────────

class SceneContextSerializer:
    """
    Converts a ConstructionScene into the ontology-guided JSON-like
    Context representation that is embedded in the LLM prompt.
    (Section 3.2 — Ontology-based scene context)

    Output format mirrors the paper's Fig. 2 example:
        Context: {
            Entity: {
                Worker_1 a Worker (5,8)
                Hoist_1 a Hoist (1,2,4,5)
                ...
            }
            Attribute: {
                Worker_1 hasPosture Standing
                Worker_1 hasBodyPart Worker_Hand_1
                ...
            }
            Relationship: {
                Worker_1 hasAttentionOn Hoist_1
                Hoist_1 isConnectedTo FormworkRegion_1
                ...
            }
        }
    """
    def serialize(self, scene: ConstructionScene) -> str:
        """Serialize a ConstructionScene to the ontology-guided Context string."""
        lines = ["Context: {"]

        # Entity section
        lines.append("\tEntity: {")
        for entity in scene.entities:
            positions_str = ",".join(str(p) for p in entity.grid_positions)
            lines.append(
                f"\t\t{entity.instance_id} a {entity.entity_class} ({positions_str})"
            )
        lines.append("\t}")

        # Attribute section
        if scene.attributes:
            lines.append("\tAttribute: {")
            for attr in scene.attributes:
                lines.append(f"\t\t{attr.subject} {attr.predicate} {attr.value}")
            lines.append("\t}")

        # Relationship section
        if scene.relationships:
            lines.append("\tRelationship: {")
            for rel in scene.relationships:
                lines.append(f"\t\t{rel.subject} {rel.predicate} {rel.obj}")
            lines.append("\t}")

        lines.append("}}")
        return "\n".join(lines)

    def serialize_inference(
        self,
        interactions: List[Dict],
        activities: List[ActivityPrediction],
    ) -> str:
        """
        Serialize the Inference section (CoT + activity labels).
        Used for constructing in-context examples.
        """
        lines = ["Inference: {"]

        # CoT: Interaction layer (intermediate reasoning)
        lines.append("\tInteraction: {")
        for interaction in interactions:
            subclass = interaction.get("subclass", "WorkerBasicInteraction")
            entities_involved = interaction.get("entities", [])
            action = interaction.get("action", "unknown")
            entity_str = " and ".join(entities_involved)
            lines.append(f"\t\t{subclass}: {{")
            lines.append(f"\t\t\t{entity_str} {action}")
            lines.append("\t\t}")
        lines.append("\t}")

        # Activities
        lines.append("\tActivity: {")
        for i, act in enumerate(activities):
            lines.append(f"\t\tActivity_{i+1}: {{")
            lines.append("\t\t\tActivity type: {")
            lines.append(f"\t\t\t\t{act.activity_type}")
            lines.append("\t\t\t}")
            lines.append("\t\t\tActivity elements: {")
            participants_str = ", ".join(act.participants)
            lines.append(f"\t\t\t\t{participants_str}")
            lines.append("\t\t\t}")
            lines.append("\t\t}")
        lines.append("\t}")
        lines.append("}")
        return "\n".join(lines)


# ─── SECTION 4: Task Formulation Prompt Builder ───────────────────────────────

def build_task_formulation_prompt(
    activity_list: List[str],
    grid_size: int = 3,
) -> str:
    """
    Build the task formulation prompt.
    (Section 3.3.1 — Task instruction prompting, Fig. 3)

    Three components:
      1. Role: Frame the LLM as a construction engineering professional
      2. Context: Explain the ontology-based representation format
      3. Task: Describe in-context learning directives and goal instructions

    The prompt is deliberately domain-neutral in structure (patterns from
    Context to Inference) to allow extension to other high-level semantic
    inference tasks beyond activity recognition.
    """
    activity_list_str = "\n".join(f"  - {a}" for a in activity_list)
    N = grid_size

    prompt = f"""Task formulation:

Your role is that of a construction engineering professional, and I'm proposing a collaborative effort in understanding how visual data in a specific construction context leads to inferences.

To start, I'll provide you with images for in-context learning. I'll address your limitations associated with processing image inputs by presenting examples based on ontology-based knowledge graphs. The graph has two parts:

The : This section includes basic visual facts about a construction scene. Each image is divided into a {N}x{N} grid, and numerical annotations indicate the spatial location of each entity (using grid positions 1–{N*N}), whether fully or partially occupied, within the image.

The : This part encapsulates the results inferred by combining the visual facts with underlying engineering heuristics. The Inference contains two layers:
  1. Interaction layer: the intermediate Chain-of-Thought component identifying interactions between entities. Interactions are categorized as:
     - WorkerBasicInteraction: physical effort and skills (installing, monitoring, assisting)
     - ToolAssistedInteraction: using hand tools (drilling, cutting, grinding)
     - MachineryInteraction: heavy machinery actions (lifting, excavating, concreting)
  2. Activity layer: the high-level construction activities inferred from the interactions.

All elements in the input are represented as triples and have explicit semantic meaning grounded in the ConSE construction ontology. To fully understand the input, your task is to carefully analyze each triple by considering both the given ontology and your expertise in construction engineering.

Your primary goal is to learn the implicit heuristics connecting the  and the . This will require you to utilize the provided examples, the ConSE ontology, and your domain-specific knowledge.

The possible construction activity types you should consider are:
{activity_list_str}
"""
    return prompt


# ─── SECTION 5: Output Formulation Prompt Builder ─────────────────────────────

def build_output_formulation_prompt(
    activity_list: List[str],
) -> str:
    """
    Build the output formulation prompt.
    (Section 3.3.1, Fig. 4)

    Two components:
      1. Task: Transition from heuristic learning to inference generation
      2. Format: Structural guidelines for response coherence

    Key format constraints:
      - Map activity type to exact ConSE ontology term (exact string match)
      - Each activity must include at least one worker or heavy machine
      - Multiple concurrent activities are allowed and encouraged
      - Avoid inferring activities for uninvolved entities
    """
    activity_list_str = "\n".join(f"  - {a}" for a in activity_list)

    prompt = f"""Output formulation:

I believe you have already grasped the implicit heuristics. Now you should apply your learnt heuristics to generate the , as per formatted in the examples, based on the new test  I shall provide. Here are some guidelines to follow when generating your answer:

1. When generating the activity type for an , it's crucial to specifically map your understanding to the terms within the brackets that I will provide. Use ONLY the following activity types:
{activity_list_str}

2. When selecting participants for an activity, it's essential to ensure that each activity includes, at the very least, either a worker or a heavy machine.

3. When the given context involves concurrent activities, you are encouraged to include multiple activities in your response.

4. Please refrain from making excessive assumptions about the situation. Instead, strive to provide reasonable inferences based on the context provided. If certain entities are not actively participating in any activities, please avoid suggesting activities for them.

5. Always include the Chain-of-Thought Interaction layer before listing the Activity layer. Follow the exact same format as demonstrated in the in-context examples.

6. Generate your response in the following JSON-like format:
Inference: {{
\tInteraction: {{
\t\t[WorkerBasicInteraction / ToolAssistedInteraction / MachineryInteraction]: {{
\t\t\t[entity] [action verb]
\t\t}}
\t}}
\tActivity: {{
\t\tActivity_N: {{
\t\t\tActivity type: {{
\t\t\t\t[activity_type]
\t\t\t}}
\t\t\tActivity elements: {{
\t\t\t\t[entity1, entity2, ...]
\t\t\t}}
\t\t}}
\t}}
}}
"""
    return prompt


# ─── SECTION 6: In-Context Example Template ───────────────────────────────────

class InContextExampleBuilder:
    """
    Formats in-context learning examples from annotated scenes.
    (Section 3.3.2 — In-context learning, Fig. 5 & 6)

    Each example follows the template:
        _N: { Entity: {} Attribute: {} Relationship: {} }
        _N: { Interaction: {} Activity: {} }

    Example selection strategy (from paper Section 4.3.1):
        Prioritize frequently occurring activity types to reflect common
        construction scenarios. Three examples are used in the base experiment,
        limited by GPT-3.5 context window constraints.
    """
    def __init__(self):
        self.serializer = SceneContextSerializer()

    def format_example(
        self,
        example: InContextExample,
        example_idx: int,
    ) -> str:
        """Format a single labeled example for inclusion in the prompt."""
        context_str = self.serializer.serialize(example.scene)
        inference_str = self.serializer.serialize_inference(
            example.interactions, example.activities
        )
        # Tag with example index for clarity
        context_tagged = context_str.replace(
            "Context: {", f"_{example_idx}: {{", 1
        )
        inference_tagged = inference_str.replace(
            "Inference: {", f"_{example_idx}: {{", 1
        )
        return f"{context_tagged}\n{inference_tagged}"

    def format_test_context(self, scene: ConstructionScene) -> str:
        """Format the test scene context (no inference label)."""
        context_str = self.serializer.serialize(scene)
        return context_str.replace(
            "Context: {", "_test: {{", 1
        )


# ─── SECTION 7: Chain-of-Thought Interaction Classifier ───────────────────────

class CoTInteractionClassifier:
    """
    Classifies entity interactions into CoT subclasses.
    (Section 3.3.3 — Chain-of-Thought prompting)

    In the paper, CoT prompting decomposes activity inference into two steps:
      Step 1: Identify interactions (gestures/actions → interactions)
      Step 2: Infer group activities from interactions

    This mirrors the human activity hierarchy from [17]:
        gestures → actions → interactions → group activities

    Interactions are classified into three subclasses (Table 2):
      - WorkerBasicInteraction: physical worker effort and skills
      - ToolAssistedInteraction: worker using hand tools
      - MachineryInteraction: heavy machinery action

    The CoT component serves as a "cognitive bridge" between visual
    observations and high-level activity labels. Ablation: removing CoT
    drops Activity Accuracy from 73.68% → 65.78%.
    """

    # Heuristic mapping from entity type pairs to interaction subclass and action
    INTERACTION_HEURISTICS = [
        # (subject_class, object_class, subclass, action)
        ("Worker", "Excavator", "MachineryInteraction", "monitoring"),
        ("Worker", "Hoist", "WorkerBasicInteraction", "operating"),
        ("Worker", "Controller", "WorkerBasicInteraction", "controlling"),
        ("Hoist", "FormworkRegion", "MachineryInteraction", "lifting"),
        ("Hoist", "ReinforcementRegion", "MachineryInteraction", "lifting"),
        ("Worker", "ConcreteRegion", "WorkerBasicInteraction", "levelling"),
        ("Worker", "ConcretePumpHose", "WorkerBasicInteraction", "operating"),
        ("Excavator", "SoilRegion", "MachineryInteraction", "excavating"),
        ("Worker", "DrillTool", "ToolAssistedInteraction", "drilling"),
        ("Worker", "Scaffolding", "WorkerBasicInteraction", "assembling"),
        ("Worker", "ReinforcementRegion", "WorkerBasicInteraction", "assembling"),
    ]

    def infer_interactions(
        self,
        scene: ConstructionScene,
    ) -> List[Dict]:
        """
        Infer likely interactions from scene relationships and attentional cues.
        
        In production, this is derived from annotated ground truth.
        This heuristic implementation approximates the manual annotation
        process described in the paper.
        """
        interactions = []
        entity_map = {e.instance_id: e.entity_class for e in scene.entities}

        # Process relationships for interaction inference
        for rel in scene.relationships:
            subj_class = entity_map.get(rel.subject, "")
            obj_class = entity_map.get(rel.obj, "")

            for (sc, oc, subclass, action) in self.INTERACTION_HEURISTICS:
                if subj_class == sc and oc in obj_class:
                    # Avoid duplicate interactions
                    new_interaction = {
                        "subclass": subclass,
                        "entities": [rel.subject, rel.obj],
                        "action": action,
                    }
                    if new_interaction not in interactions:
                        interactions.append(new_interaction)

        # Also process attentional cues (hasAttentionOn predicates)
        for rel in scene.relationships:
            if rel.predicate == "hasAttentionOn":
                obj_class = entity_map.get(rel.obj, "")
                # Worker attention on machinery → monitoring interaction
                if obj_class in ["Excavator", "Hoist", "Crane", "ConcretePump"]:
                    interaction = {
                        "subclass": "WorkerBasicInteraction",
                        "entities": [rel.subject],
                        "action": f"monitoring {rel.obj}",
                    }
                    if interaction not in interactions:
                        interactions.append(interaction)

        return interactions


# ─── SECTION 8: Full Prompt Assembler ─────────────────────────────────────────

class OntologyPromptAssembler:
    """
    Assembles the complete prompt for LLM-based construction activity inference.
    (Section 3 — Full pipeline from Fig. 1)

    Prompt structure:
        1. Task Formulation Prompt (sets role, context, task)
        2. In-context examples (n examples: Context + Inference with CoT)
        3. Output Formulation Prompt (format and inference guidelines)
        4. Test Context (new scene to classify)

    The approach uses in-context learning only — no fine-tuning.
    GPT-3.5-turbo was used in the paper; the prompts are model-agnostic.
    """
    def __init__(
        self,
        activity_classes: List[str] = None,
        grid_size: int = 3,
    ):
        self.activity_classes = activity_classes or CONSE_ACTIVITY_CLASSES
        self.grid_size = grid_size
        self.serializer = SceneContextSerializer()
        self.example_builder = InContextExampleBuilder()

    def assemble_prompt(
        self,
        test_scene: ConstructionScene,
        in_context_examples: List[InContextExample],
    ) -> str:
        """
        Assemble the full prompt for a test scene.
        
        Args:
            test_scene: The construction scene to classify
            in_context_examples: Labeled examples for in-context learning
                                 (paper uses 3 examples in base config)
        
        Returns:
            Complete prompt string ready to send to the LLM API
        """
        sections = []

        # 1. Task formulation prompt
        sections.append(build_task_formulation_prompt(
            self.activity_classes, self.grid_size
        ))
        sections.append("=" * 60)

        # 2. In-context examples (Context + Inference pairs)
        for i, example in enumerate(in_context_examples):
            sections.append(self.example_builder.format_example(example, i+1))
            sections.append("-" * 40)

        # 3. Output formulation prompt
        sections.append(build_output_formulation_prompt(self.activity_classes))
        sections.append("=" * 60)

        # 4. Test context (the new scene to classify)
        sections.append("Now classify the following new construction scene:")
        sections.append(self.example_builder.format_test_context(test_scene))

        return "\n\n".join(sections)

    def call_llm(
        self,
        prompt: str,
        temperature: float = 0.5,
        top_p: float = 0.3,
        model: str = "gpt-3.5-turbo",
    ) -> str:
        """
        Call the LLM API with the assembled prompt.
        
        Paper configuration (Section 4.3.1):
            model = "gpt-3.5-turbo-0613"
            temperature = 0.5, top_p = 0.3  (base experiment)
            temperature = 1.0, top_p = 0.6  (moderate randomness mR → best result)
            temperature = 2.0, top_p = 0.9  (high randomness hR → degrades format)

        Requires: pip install openai
        Set OPENAI_API_KEY environment variable before use.
        """
        try:
            import openai
            client = openai.OpenAI()
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=temperature,
                top_p=top_p,
            )
            return response.choices[0].message.content
        except ImportError:
            return "[openai package not installed — run: pip install openai]"
        except Exception as e:
            return f"[LLM API error: {e}]"


# ─── SECTION 9: Evaluation Metrics ────────────────────────────────────────────

class ConstructionActivityEvaluator:
    """
    Evaluation metrics for construction activity recognition.
    (Section 4.3.1 — Evaluation metrics)

    Four metrics are computed:

    1. Activity Accuracy (micro-average):
       Fraction of predicted activity types that exactly match ground truth.
       Computed as: TP / total_samples (case-sensitive exact string match)

    2. Activity + Elements Accuracy:
       Fraction of predictions where BOTH the activity type AND all
       involved entity IDs exactly match ground truth.
       Harder metric — reflects group activity recognition quality.

    3. Redundancy Rate:
       Proportion of "None" labels added to ground truth to match prediction
       length / original ground truth count.
       High redundancy → model predicts more activities than actually present.

    4. Miss Rate:
       Proportion of "None" labels added to predictions to match GT length /
       original ground truth count.
       High miss rate → model fails to predict relevant activities.

    Evaluation protocol:
       - If prediction has more labels than GT (or vice versa), pad the
         shorter side with "None" for alignment.
       - Synonyms and semantic equivalents are NOT counted as correct.
       - All comparisons are case-sensitive.
    """
    def evaluate(
        self,
        predictions: List[List[ActivityPrediction]],  # Per image
        ground_truths: List[List[ActivityPrediction]], # Per image
    ) -> Dict[str, float]:
        """
        Compute all four evaluation metrics.
        
        Returns dict with keys:
            activity_accuracy, activity_elements_accuracy,
            redundancy_rate, miss_rate
        """
        total_gt_count = 0
        tp_activity = 0
        tp_activity_elements = 0
        total_none_in_gt = 0    # For Redundancy Rate
        total_none_in_pred = 0  # For Miss Rate

        for pred_acts, gt_acts in zip(predictions, ground_truths):
            n_gt = len(gt_acts)
            n_pred = len(pred_acts)
            total_gt_count += n_gt

            # Pad shorter side with None-activity placeholders
            max_len = max(n_gt, n_pred)

            # Nones added to GT (when prediction has MORE activities)
            none_added_to_gt = max(0, n_pred - n_gt)
            # Nones added to pred (when GT has MORE activities)
            none_added_to_pred = max(0, n_gt - n_pred)

            total_none_in_gt += none_added_to_gt
            total_none_in_pred += none_added_to_pred

            # Align: try to find best pairing (greedy by activity type match)
            # In the paper, padding is appended; we use simple positional alignment
            for i in range(max_len):
                pred = pred_acts[i] if i < n_pred else None
                gt = gt_acts[i] if i < n_gt else None

                if pred is None or gt is None:
                    continue

                # Activity Accuracy: exact activity type match
                if pred.activity_type == gt.activity_type:
                    tp_activity += 1

                    # Activity + Elements: activity type AND all participants match
                    pred_participants = set(pred.participants)
                    gt_participants = set(gt.participants)
                    if pred_participants == gt_participants:
                        tp_activity_elements += 1

        total_samples = max(total_gt_count, 1)
        activity_accuracy = tp_activity / total_samples
        activity_elements_accuracy = tp_activity_elements / total_samples
        redundancy_rate = total_none_in_gt / max(total_gt_count, 1)
        miss_rate = total_none_in_pred / max(total_gt_count, 1)

        return {
            "activity_accuracy": activity_accuracy,
            "activity_elements_accuracy": activity_elements_accuracy,
            "redundancy_rate": redundancy_rate,
            "miss_rate": miss_rate,
        }

    def parse_llm_response(self, response_text: str) -> List[ActivityPrediction]:
        """
        Parse the structured LLM response into ActivityPrediction objects.
        
        Looks for:
            Activity type: { ActivityName }
            Activity elements: { entity1, entity2, ... }
        """
        activities = []
        # Find all Activity blocks
        activity_pattern = re.compile(
            rr'Activity\s+type\s*:\s*\{([^}]+)\}.*?Activity\s+elements\s*:\s*\{([^}]+)\}',
            re.DOTALL | re.IGNORECASE
        )
        for match in activity_pattern.finditer(response_text):
            act_type = match.group(1).strip()
            elements_str = match.group(2).strip()
            participants = [e.strip() for e in elements_str.split(",") if e.strip()]
            activities.append(ActivityPrediction(
                activity_type=act_type,
                participants=participants,
            ))
        return activities


# ─── SECTION 10: End-to-End Demo ──────────────────────────────────────────────

def create_formwork_setup_scene() -> Tuple[ConstructionScene, List[Dict], List[ActivityPrediction]]:
    """
    Create the FormworkSetUp example scene from Fig. 12 in the paper.
    Worker_1 operates Hoist_1, lifting Formwork_1 near Column_2.
    """
    encoder = GridSpatialEncoder()

    scene = ConstructionScene(image_id="formwork_setup_example")

    # Entities with spatial positions
    scene.entities = [
        Entity("Worker_1", "Worker", encoder.encode((0.55, 0.4, 0.7, 0.9))),
        Entity("Hoist_1", "Hoist", encoder.encode((0.1, 0.0, 0.5, 0.6))),
        Entity("Formwork_1", "FormworkRegion", encoder.encode((0.15, 0.3, 0.45, 0.75))),
        Entity("Column_1", "Column", encoder.encode((0.35, 0.0, 0.5, 1.0))),
        Entity("Column_2", "Column", encoder.encode((0.55, 0.0, 0.7, 1.0))),
    ]
    scene.attributes = [
        EntityAttribute("Worker_1", "hasPosture", "Standing"),
        EntityAttribute("Worker_1", "hasBodyPart", "Worker_Hand_1"),
    ]
    scene.relationships = [
        EntityRelationship("Worker_1", "hasAttentionOn", "Hoist_1"),
        EntityRelationship("Hoist_1", "isConnectedTo", "Formwork_1"),
        EntityRelationship("Worker_1", "isNear", "Column_2"),
    ]

    interactions = [
        {"subclass": "WorkerBasicInteraction", "entities": ["Worker_1"], "action": "monitoring Hoist_1"},
        {"subclass": "MachineryInteraction", "entities": ["Hoist_1"], "action": "lifting Formwork_1"},
    ]
    activities = [
        ActivityPrediction("FormworkSetUp", ["Worker_1", "Hoist_1", "Formwork_1"]),
    ]
    return scene, interactions, activities


def create_soil_excavation_scene() -> Tuple[ConstructionScene, List[Dict], List[ActivityPrediction]]:
    """
    Create the SoilExcavation + ProcessMonitoring example from Fig. 10.
    Worker_2 operates Excavator_1; Worker_1 monitors with a shovel.
    """
    encoder = GridSpatialEncoder()
    scene = ConstructionScene(image_id="soil_excavation_monitoring")
    scene.entities = [
        Entity("Worker_1", "Worker", encoder.encode((0.7, 0.2, 0.9, 0.9))),
        Entity("Worker_2", "Worker", encoder.encode((0.3, 0.3, 0.6, 0.95))),
        Entity("Excavator_1", "Excavator", encoder.encode((0.0, 0.0, 0.65, 1.0))),
        Entity("Shovel_1", "ShovelTool", encoder.encode((0.72, 0.4, 0.85, 0.9))),
        Entity("SoilTrenchRegion_1", "SoilTrenchRegion", encoder.encode((0.1, 0.7, 0.6, 1.0))),
        Entity("SoilGroundRegion_1", "SoilGroundRegion", encoder.encode((0.6, 0.5, 1.0, 1.0))),
    ]
    scene.attributes = [
        EntityAttribute("Worker_1", "hasPosture", "Standing"),
        EntityAttribute("Worker_1", "hasBodyPart", "Worker_Hand_1"),
        EntityAttribute("Worker_2", "hasPosture", "Sitting"),
    ]
    scene.relationships = [
        EntityRelationship("Worker_2", "isInsideOf", "Excavator_1"),
        EntityRelationship("Worker_1", "hasAttentionOn", "Excavator_1"),
        EntityRelationship("Worker_1", "isHolding", "Shovel_1"),
        EntityRelationship("Worker_1", "isStandingOn", "SoilGroundRegion_1"),
        EntityRelationship("Excavator_1", "isWorkingOn", "SoilTrenchRegion_1"),
    ]
    interactions = [
        {"subclass": "MachineryInteraction", "entities": ["Excavator_1"], "action": "excavating SoilTrenchRegion_1"},
        {"subclass": "WorkerBasicInteraction", "entities": ["Worker_1"], "action": "monitoring Excavator_1"},
    ]
    activities = [
        ActivityPrediction("SoilExcavation", ["Worker_2", "Excavator_1", "SoilTrenchRegion_1"]),
        ActivityPrediction("ProcessMonitoring", ["Worker_1", "Excavator_1"]),
    ]
    return scene, interactions, activities


def create_reinforcement_transport_scene() -> Tuple[ConstructionScene, List[Dict], List[ActivityPrediction]]:
    """Test scene: ReinforcementTransportation from Fig. 13."""
    encoder = GridSpatialEncoder()
    scene = ConstructionScene(image_id="reinforcement_transport_test")
    scene.entities = [
        Entity("Worker_1", "Worker", encoder.encode((0.05, 0.5, 0.25, 0.95))),
        Entity("Controller_1", "Controller", encoder.encode((0.1, 0.6, 0.2, 0.8))),
        Entity("Hoist_1", "Hoist", encoder.encode((0.3, 0.0, 0.7, 0.5))),
        Entity("ReinforcementRegion_1", "ReinforcementRegion", encoder.encode((0.35, 0.45, 0.65, 0.9))),
    ]
    scene.attributes = [
        EntityAttribute("Worker_1", "hasPosture", "Standing"),
        EntityAttribute("Worker_1", "hasBodyPart", "Worker_Hand_1"),
    ]
    scene.relationships = [
        EntityRelationship("Worker_1", "isHolding", "Controller_1"),
        EntityRelationship("Worker_1", "hasAttentionOn", "ReinforcementRegion_1"),
        EntityRelationship("Hoist_1", "isConnectedTo", "ReinforcementRegion_1"),
        EntityRelationship("Worker_Hand_1", "isConnectedTo", "Controller_1"),
    ]
    # Test scene — no ground truth interactions/activities (model must infer)
    return scene, [], []


def run_demo():
    """
    End-to-end demonstration of the ontology-based prompting pipeline.
    
    This demo:
      1. Creates in-context examples from paper Figs 10 and 12
      2. Assembles the full prompt for a test scene (Fig. 13)
      3. Shows the assembled prompt structure
      4. Demonstrates evaluation metric computation on synthetic data
      5. (Optionally) calls GPT API if OPENAI_API_KEY is set
    """
    print("="*65)
    print("  Ontology-Based LLM Prompting — Construction Activity Demo")
    print("="*65)

    # 1. Build in-context examples
    print("\n[1/4] Building in-context examples from annotated scenes...")
    scene_fw, interactions_fw, activities_fw = create_formwork_setup_scene()
    scene_se, interactions_se, activities_se = create_soil_excavation_scene()

    examples = [
        InContextExample(scene_fw, interactions_fw, activities_fw),
        InContextExample(scene_se, interactions_se, activities_se),
    ]
    print(f"  Created {len(examples)} in-context examples")
    for ex in examples:
        acts_str = ", ".join(a.activity_type for a in ex.activities)
        print(f"  Scene '{ex.scene.image_id}': {acts_str}")

    # 2. Build test scene
    print("\n[2/4] Preparing test scene...")
    test_scene, _, _ = create_reinforcement_transport_scene()
    print(f"  Test scene: '{test_scene.image_id}'")
    print(f"  Entities: {[e.instance_id for e in test_scene.entities]}")

    # 3. Assemble prompt
    print("\n[3/4] Assembling full ontology-based prompt...")
    assembler = OntologyPromptAssembler(
        activity_classes=CONSE_ACTIVITY_CLASSES[:15],  # Subset for demo
        grid_size=3,
    )
    prompt = assembler.assemble_prompt(test_scene, examples)
    prompt_lines = prompt.split("\n")
    print(f"  Total prompt: {len(prompt_lines)} lines, ~{len(prompt)//4} tokens (est.)")
    print("\n  --- PROMPT PREVIEW (first 30 lines) ---")
    for line in prompt_lines[:30]:
        print(f"  {line}")
    print("  ... [truncated]")

    # 4. Demonstrate evaluation metrics
    print("\n[4/4] Demonstrating evaluation metrics...")
    evaluator = ConstructionActivityEvaluator()

    # Simulate base experiment results from Table 5
    sim_predictions = [
        [ActivityPrediction("SoilExcavation", ["Worker_2", "Excavator_1"])],
        [ActivityPrediction("ConcretePouring", ["Worker_2", "ConcretePumpHose_1"])],
        [ActivityPrediction("FormworkSetUp", ["Worker_1", "Hoist_1", "Formwork_1"])],
        [ActivityPrediction("ReinforcementTransportation", ["Worker_1", "Hoist_1"])],
    ]
    sim_ground_truths = [
        [ActivityPrediction("SoilExcavation", ["Worker_2", "Excavator_1", "SoilTrenchRegion_1"])],
        [ActivityPrediction("ConcretePouring", ["Worker_2", "Hoist_1", "ConcretePumpHose_1"])],
        [ActivityPrediction("FormworkSetUp", ["Worker_1", "Hoist_1", "Formwork_1"])],
        [ActivityPrediction("ReinforcementTransportation", ["Worker_1", "Hoist_1", "ReinforcementRegion_1"])],
    ]

    metrics = evaluator.evaluate(sim_predictions, sim_ground_truths)
    print("\n  Evaluation Metrics (simulated):")
    print(f"    Activity Accuracy:           {metrics['activity_accuracy']:.2%}")
    print(f"    Activity + Elements Accuracy: {metrics['activity_elements_accuracy']:.2%}")
    print(f"    Redundancy Rate:              {metrics['redundancy_rate']:.2%}")
    print(f"    Miss Rate:                    {metrics['miss_rate']:.2%}")

    # 5. Optional: call GPT if API key available
    import os
    if os.environ.get("OPENAI_API_KEY"):
        print("\n[Optional] Calling GPT API (OPENAI_API_KEY found)...")
        response = assembler.call_llm(
            prompt=prompt,
            temperature=0.5,
            top_p=0.3,
            model="gpt-3.5-turbo",
        )
        print("  LLM Response:")
        print(response)
        parsed = evaluator.parse_llm_response(response)
        print(f"\n  Parsed activities: {[(a.activity_type, a.participants) for a in parsed]}")
    else:
        print("\n  [Skipping LLM API call — set OPENAI_API_KEY to enable]")

    print("\n" + "="*65)
    print("✓  Demo complete. Pipeline structure verified.")
    print("="*65)
    print("""
Next steps for deployment:
  1. Implement computer vision pipeline:
       - Object detection: YOLOv8 or Grounding-DINO for entity extraction
       - Semantic segmentation: SAM or SegFormer for region masks
       - Pose estimation: MediaPipe or ViTPose for worker body analysis
       - Gaze estimation: OpenFace or specialized gaze model
       - Spatial analysis: Relationship detection model

  2. Upgrade to GPT-4o or Claude for stronger construction reasoning:
       assembler.call_llm(prompt, model="gpt-4o", temperature=0.5)

  3. Build the ConSE ontology reference:
       pip install owlready2
       # Load OWL file from ConSE paper (DOI: 10.1016/j.aei.2024.102446)

  4. Expand in-context examples to cover all major activity types:
       # Paper uses 3 examples; performance improves with more (Table 5)
       # Prioritize frequently occurring activities in your target dataset

  5. Add depth estimation for better spatial reasoning:
       # MiDaS or DPT-Hybrid for monocular depth estimation
       # Convert bounding boxes to approximate 3D positions
""")


if __name__ == "__main__":
    run_demo()