diff --git a/core/physical/domains/hardware_engine.py b/core/physical/domains/hardware_engine.py new file mode 100644 index 0000000..7bc41b5 --- /dev/null +++ b/core/physical/domains/hardware_engine.py @@ -0,0 +1,314 @@ +""" +core/physical/domains/hardware_engine.py +========================================= +Hardware Repair Domain Guidance Engine. + +Analyses a single perception frame (object detections, hand landmarks, OCR) +and returns a structured GuidanceInstruction with safety-critical warnings +and step guidance appropriate to the current hardware repair task. + +Safety priority (highest to lowest): + 1. CRITICAL_WARNING — must be resolved before any work continues + 2. WARNING — proceed with care + 3. OK — all safety checks passed + +Multiple hazards in a single frame are ALL reported (comma-separated message), +so no warning is silently dropped by elif short-circuit logic. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + + +# --------------------------------------------------------------------------- +# Data types +# --------------------------------------------------------------------------- + +@dataclass +class GuidanceInstruction: + status: str # "OK" | "WARNING" | "CRITICAL_WARNING" + message: str # Human-readable, actionable guidance + critical_warning: bool # True if any CRITICAL rule triggered + detected_components: List[str] # Mapped component names (no duplicates) + active_template: Optional[str] = None # Which step template is in use + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _bbox_center(bbox: List[float]) -> Tuple[float, float]: + """Return (cx, cy) for a bounding box [x1, y1, x2, y2].""" + x1, y1, x2, y2 = bbox + return (x1 + x2) / 2.0, (y1 + y2) / 2.0 + + +def _bbox_distance(bbox_a: List[float], bbox_b: List[float]) -> float: + """Euclidean distance between the centres of two bounding boxes.""" + cx_a, cy_a = _bbox_center(bbox_a) + cx_b, cy_b = _bbox_center(bbox_b) + return math.hypot(cx_a - cx_b, cy_a - cy_b) + + +# --------------------------------------------------------------------------- +# HardwareEngine +# --------------------------------------------------------------------------- + +class HardwareEngine: + """ + Guidance engine for hardware repair tasks. + + Step templates + -------------- + The engine picks the most relevant built-in step template from + ``self.templates`` by matching detected component classes against + known template keywords. The chosen template is surfaced in the + returned ``GuidanceInstruction.active_template`` field so that the + caller (e.g. IntelligenceCore) can display contextual step instructions. + + Safety rules + ------------ + Rules are evaluated INDEPENDENTLY (not elif) so that every hazard + present in a frame is reported in the same message. + + Rule 1: circuit_board without esd_strap → CRITICAL_WARNING + Rule 2: soldering_iron within proximity_threshold + pixels of any non-target component → WARNING + Rule 3: power_tool without safety_glasses → CRITICAL_WARNING + """ + + # Proximity threshold in pixels. Soldering iron within this distance of + # a non-target component triggers a warning. + SOLDERING_PROXIMITY_THRESHOLD: float = 150.0 + + # Component dictionary: detection class → human-readable label + component_dictionary: Dict[str, str] = { + "screwdriver": "Tool: Screwdriver", + "m3_screw": "Fastener: M3 Screw", + "heat_sink": "Component: Heat Sink", + "circuit_board": "Component: Circuit Board", + "esd_strap": "Safety: ESD Strap", + "soldering_iron": "Tool: Soldering Iron", + "power_tool": "Tool: Power Tool", + "safety_glasses": "Safety: Safety Glasses", + } + + # Step templates: name → list of component keywords that indicate this task + templates: Dict[str, List[str]] = { + "PC assembly": ["circuit_board", "heat_sink", "screwdriver", "m3_screw"], + "circuit board repair": ["circuit_board", "soldering_iron", "esd_strap"], + "appliance disassembly": ["screwdriver", "power_tool", "safety_glasses"], + "cable management": ["screwdriver"], + } + + def __init__(self) -> None: + # Re-expose as instance attributes so callers can introspect/override + self.component_dictionary = dict(HardwareEngine.component_dictionary) + self.templates = dict(HardwareEngine.templates) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def analyze( + self, + detections: List[Dict[str, Any]], + hand_results: Any, + ocr_text: str, + ) -> GuidanceInstruction: + """ + Analyse the current perception frame and return a GuidanceInstruction. + + Parameters + ---------- + detections: + List of detected objects. Each dict must have a ``"class"`` key + (str) and may optionally have a ``"bbox"`` key ([x1, y1, x2, y2]) + for proximity calculations. + hand_results: + MediaPipe / hand-landmark output. Reserved for future gesture-based + rules (e.g. "hand approaching soldering iron tip"). + ocr_text: + Raw OCR text extracted from the frame. Used to surface part numbers + or repair step labels mentioned in the scene. + """ + # Normalise: extract class names as lowercase strings + detected_classes: List[str] = [ + det.get("class", "").lower() for det in detections + ] + + # ------------------------------------------------------------------ + # Collect all triggered messages INDEPENDENTLY (no elif) + # ------------------------------------------------------------------ + critical_messages: List[str] = [] + warning_messages: List[str] = [] + + # Rule 1: circuit board without ESD strap → CRITICAL + if ( + "circuit_board" in detected_classes + and "esd_strap" not in detected_classes + ): + critical_messages.append( + "Circuit board detected without an ESD strap — " + "wear an ESD strap immediately to prevent static damage." + ) + + # Rule 3: power tool without safety glasses → CRITICAL + if ( + "power_tool" in detected_classes + and "safety_glasses" not in detected_classes + ): + critical_messages.append( + "Power tool detected without safety glasses — " + "put on safety glasses before proceeding." + ) + + # Rule 2: soldering iron proximity to non-target components → WARNING + if "soldering_iron" in detected_classes: + proximity_breach = self._check_soldering_proximity(detections) + if proximity_breach: + warning_messages.append( + f"Soldering iron is within {self.SOLDERING_PROXIMITY_THRESHOLD:.0f}px " + f"of '{proximity_breach}' — move iron away from non-target components." + ) + else: + warning_messages.append( + "Soldering iron active — keep it away from non-target components and wires." + ) + + # ------------------------------------------------------------------ + # OCR integration: surface part numbers / step labels + # ------------------------------------------------------------------ + if ocr_text.strip(): + ocr_note = self._parse_ocr(ocr_text) + if ocr_note: + warning_messages.append(ocr_note) + + # ------------------------------------------------------------------ + # Build unified status and message + # ------------------------------------------------------------------ + if critical_messages: + status = "CRITICAL_WARNING" + critical_warning = True + all_parts = ["CRITICAL: " + m for m in critical_messages] + if warning_messages: + all_parts += ["WARNING: " + m for m in warning_messages] + message = " | ".join(all_parts) + elif warning_messages: + status = "WARNING" + critical_warning = False + message = " | ".join("WARNING: " + m for m in warning_messages) + else: + status = "OK" + critical_warning = False + message = "Step validation passed. Safe to proceed." + + # ------------------------------------------------------------------ + # Component identification + # ------------------------------------------------------------------ + matched_components = list({ + self.component_dictionary.get(cls, f"Unknown: {cls}") + for cls in detected_classes + if cls # skip empty strings from missing "class" keys + }) + + # ------------------------------------------------------------------ + # Step template selection + # ------------------------------------------------------------------ + active_template = self._select_template(detected_classes) + + return GuidanceInstruction( + status=status, + message=message, + critical_warning=critical_warning, + detected_components=matched_components, + active_template=active_template, + ) + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _check_soldering_proximity( + self, detections: List[Dict[str, Any]] + ) -> Optional[str]: + """ + Return the class name of the closest non-target component to the + soldering iron if it is within SOLDERING_PROXIMITY_THRESHOLD pixels, + or None if no proximity breach. + + Falls back gracefully when bounding boxes are absent (returns None). + """ + iron_det = next( + (d for d in detections if d.get("class", "").lower() == "soldering_iron"), + None, + ) + if iron_det is None or "bbox" not in iron_det: + return None # no bbox data available; skip proximity check + + iron_bbox = iron_det["bbox"] + + # Components considered "non-target" — not the iron itself and not + # intentional soldering targets like the circuit board during repair. + non_target_classes = { + "esd_strap", "screwdriver", "m3_screw", + "heat_sink", "power_tool", "safety_glasses", + } + + closest_name: Optional[str] = None + closest_dist = float("inf") + + for det in detections: + cls = det.get("class", "").lower() + if cls not in non_target_classes: + continue + if "bbox" not in det: + continue + dist = _bbox_distance(iron_bbox, det["bbox"]) + if dist < closest_dist: + closest_dist = dist + closest_name = cls + + if closest_dist < self.SOLDERING_PROXIMITY_THRESHOLD: + return closest_name + return None + + def _select_template(self, detected_classes: List[str]) -> Optional[str]: + """ + Return the template name with the most keyword matches against + the detected classes. Returns None if no class matches any template. + """ + best_template: Optional[str] = None + best_score = 0 + + for template_name, keywords in self.templates.items(): + score = sum(1 for kw in keywords if kw in detected_classes) + if score > best_score: + best_score = score + best_template = template_name + + return best_template if best_score > 0 else None + + def _parse_ocr(self, ocr_text: str) -> Optional[str]: + """ + Extract actionable information from OCR text. + + Currently surfaces part numbers (strings matching typical P/N formats) + and step headings. Extend this for richer repair-manual parsing. + """ + import re + + # Look for step headings like "Step 3:" or "STEP 3 –" + step_match = re.search(r"step\s*(\d+)", ocr_text, re.IGNORECASE) + if step_match: + return f"OCR detected repair step {step_match.group(1)} in frame." + + # Look for part numbers like PN-1234 or P/N: AB-5678 + pn_match = re.search(r"p/?n[:\s-]*([A-Z0-9\-]+)", ocr_text, re.IGNORECASE) + if pn_match: + return f"OCR detected part number: {pn_match.group(1)}." + + return None \ No newline at end of file diff --git a/docs/physical_domain.md b/docs/physical_domain.md new file mode 100644 index 0000000..ed7a3b6 --- /dev/null +++ b/docs/physical_domain.md @@ -0,0 +1,107 @@ +# Hardware Repair Domain + +## Overview + +The `HardwareEngine` analyses a single perception frame (CV detections, +hand landmarks, OCR text) and returns a `GuidanceInstruction` with +safety-critical warnings and contextual step guidance for hardware repair tasks. + +--- + +## Component Dictionary + +Raw CV detection class names are mapped to human-readable component labels: + +| Detection Class | Component Mapping | Description | +|:-----------------|:---------------------------|:-------------------------------------| +| `screwdriver` | Tool: Screwdriver | General hand tool for fasteners. | +| `m3_screw` | Fastener: M3 Screw | Common M3 hardware fastener. | +| `heat_sink` | Component: Heat Sink | Cooling element for CPUs/chips. | +| `circuit_board` | Component: Circuit Board | Motherboards or logic boards. | +| `esd_strap` | Safety: ESD Strap | Anti-static wrist strap. | +| `soldering_iron` | Tool: Soldering Iron | High-heat soldering tool. | +| `power_tool` | Tool: Power Tool | Drills, electric screwdrivers, etc. | +| `safety_glasses` | Safety: Safety Glasses | Protective eyewear. | + +Unknown detection classes are returned as `"Unknown: "`. + +--- + +## Step Templates + +The engine automatically selects the most relevant repair template by counting +keyword matches between detected component classes and each template's keyword list. +The winning template is returned in `GuidanceInstruction.active_template`. + +| Template Name | Trigger Keywords | +|:-----------------------|:----------------------------------------------------------| +| PC assembly | `circuit_board`, `heat_sink`, `screwdriver`, `m3_screw` | +| circuit board repair | `circuit_board`, `soldering_iron`, `esd_strap` | +| appliance disassembly | `screwdriver`, `power_tool`, `safety_glasses` | +| cable management | `screwdriver` | + +--- + +## Safety Rules + +Rules are evaluated **independently** — all hazards present in a frame are +reported in a single combined message, separated by ` | `. + +### Rule 1 — ESD Protection (CRITICAL) + +**Trigger:** `circuit_board` detected and `esd_strap` not in frame. + +**Message:** `CRITICAL: Circuit board detected without an ESD strap — wear an ESD strap immediately to prevent static damage.` + +### Rule 2 — Soldering Iron Proximity (WARNING) + +**Trigger:** `soldering_iron` detected. + +- If the soldering iron bounding box centre is within **150 px** of any + non-target component (ESD strap, screwdriver, M3 screw, heat sink, + power tool, safety glasses), a proximity-specific warning is emitted. +- If no bounding box data is available, a generic caution message is emitted. + +**Message (proximity breach):** `WARNING: Soldering iron is within 150px of '' — move iron away from non-target components.` + +**Message (no bbox / no breach):** `WARNING: Soldering iron active — keep it away from non-target components and wires.` + +### Rule 3 — Power Tool Safety (CRITICAL) + +**Trigger:** `power_tool` detected and `safety_glasses` not in frame. + +**Message:** `CRITICAL: Power tool detected without safety glasses — put on safety glasses before proceeding.` + +--- + +## Status Codes + +| Status | Meaning | +|:------------------|:-------------------------------------------------| +| `OK` | All safety checks passed; safe to proceed. | +| `WARNING` | Proceed with caution; hazard present but not critical. | +| `CRITICAL_WARNING`| Stop immediately; safety equipment is missing. | + +--- + +## OCR Integration + +If `ocr_text` is non-empty, the engine attempts to extract: + +- **Step numbers** — e.g. `"Step 3:"` → `"OCR detected repair step 3 in frame."` +- **Part numbers** — e.g. `"P/N: AB-5678"` → `"OCR detected part number: AB-5678."` + +Extracted notes are appended as `WARNING` level messages so they appear +alongside (but never override) safety-critical guidance. + +--- + +## GuidanceInstruction fields + +| Field | Type | Description | +|:---------------------|:---------------|:-------------------------------------------------| +| `status` | `str` | `"OK"`, `"WARNING"`, or `"CRITICAL_WARNING"` | +| `message` | `str` | Full human-readable guidance; hazards joined by ` | ` | +| `critical_warning` | `bool` | `True` if any CRITICAL rule triggered | +| `detected_components`| `List[str]` | Deduplicated mapped component names | +| `active_template` | `Optional[str]`| Best-matching step template, or `None` | \ No newline at end of file diff --git a/tests/test_hardware_engine.py b/tests/test_hardware_engine.py new file mode 100644 index 0000000..bfeb240 --- /dev/null +++ b/tests/test_hardware_engine.py @@ -0,0 +1,249 @@ +""" +tests/test_hardware_engine.py +============================== +Unit tests for HardwareEngine — every safety rule, edge case, +template selection, OCR integration, and proximity detection. + +Run with: + pytest tests/test_hardware_engine.py -v +""" + +import pytest +from core.physical.domains.hardware_engine import HardwareEngine + + +@pytest.fixture +def engine(): + return HardwareEngine() + + +# --------------------------------------------------------------------------- +# Rule 1: circuit_board without ESD strap +# --------------------------------------------------------------------------- + +class TestCircuitBoardRule: + + def test_circuit_board_missing_esd_strap(self, engine): + """Original test — must still pass.""" + detections = [{"class": "circuit_board"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is True + assert result.status == "CRITICAL_WARNING" + assert "ESD strap" in result.message + + def test_circuit_board_with_esd_strap_ok(self, engine): + """When ESD strap is present, no critical warning for this rule.""" + detections = [{"class": "circuit_board"}, {"class": "esd_strap"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is False + assert "ESD" not in result.message or result.status != "CRITICAL_WARNING" + + +# --------------------------------------------------------------------------- +# Rule 3: power_tool without safety_glasses +# --------------------------------------------------------------------------- + +class TestPowerToolRule: + + def test_power_tool_missing_safety_glasses(self, engine): + """Original test — must still pass.""" + detections = [{"class": "power_tool"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is True + assert result.status == "CRITICAL_WARNING" + assert "safety glasses" in result.message.lower() + + def test_power_tool_with_safety_glasses_ok(self, engine): + detections = [{"class": "power_tool"}, {"class": "safety_glasses"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is False + + +# --------------------------------------------------------------------------- +# Rule 2: soldering iron +# --------------------------------------------------------------------------- + +class TestSolderingIronRule: + + def test_soldering_iron_warning_no_bbox(self, engine): + """Original test — must still pass (no bbox data → generic warning).""" + detections = [{"class": "soldering_iron"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is False + assert result.status == "WARNING" + assert "Soldering iron" in result.message + + def test_soldering_iron_proximity_breach(self, engine): + """Soldering iron bbox within threshold of a non-target component.""" + detections = [ + {"class": "soldering_iron", "bbox": [100, 100, 150, 150]}, + {"class": "m3_screw", "bbox": [160, 100, 200, 140]}, # ~10px away + ] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.status == "WARNING" + assert "m3_screw" in result.message or "proximity" in result.message.lower() + + def test_soldering_iron_no_proximity_breach(self, engine): + """Soldering iron far from all non-target components → generic warning only.""" + detections = [ + {"class": "soldering_iron", "bbox": [0, 0, 50, 50]}, + {"class": "m3_screw", "bbox": [1000, 1000, 1100, 1100]}, + ] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.status == "WARNING" + assert "m3_screw" not in result.message + + +# --------------------------------------------------------------------------- +# Safe operations (original test) +# --------------------------------------------------------------------------- + +class TestSafeOperations: + + def test_all_safety_gear_present(self, engine): + """Original test — must still pass.""" + detections = [ + {"class": "power_tool"}, + {"class": "safety_glasses"}, + {"class": "circuit_board"}, + {"class": "esd_strap"}, + ] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is False + assert result.status == "OK" + + +# --------------------------------------------------------------------------- +# FIX VERIFICATION: multiple simultaneous hazards (was broken by elif) +# --------------------------------------------------------------------------- + +class TestMultipleSimultaneousHazards: + + def test_both_critical_rules_fire_simultaneously(self, engine): + """ + power_tool (no glasses) + circuit_board (no ESD) in the same frame. + BOTH critical warnings must appear — the old elif code silently dropped + one of them. + """ + detections = [ + {"class": "power_tool"}, + {"class": "circuit_board"}, + ] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is True + assert result.status == "CRITICAL_WARNING" + assert "ESD" in result.message or "static" in result.message.lower() + assert "safety glasses" in result.message.lower() or "glasses" in result.message.lower() + + def test_critical_and_soldering_warning_together(self, engine): + """ + circuit_board (no ESD) + soldering iron → CRITICAL takes priority + but soldering warning is still surfaced in the same message. + """ + detections = [ + {"class": "circuit_board"}, + {"class": "soldering_iron"}, + ] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.critical_warning is True + assert "Soldering iron" in result.message or "soldering" in result.message.lower() + + +# --------------------------------------------------------------------------- +# Template selection +# --------------------------------------------------------------------------- + +class TestTemplateSelection: + + def test_pc_assembly_template_selected(self, engine): + detections = [{"class": "circuit_board"}, {"class": "esd_strap"}, + {"class": "screwdriver"}, {"class": "heat_sink"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.active_template == "PC assembly" + + def test_circuit_board_repair_template(self, engine): + detections = [{"class": "soldering_iron"}, {"class": "esd_strap"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.active_template == "circuit board repair" + + def test_no_template_when_no_match(self, engine): + detections = [{"class": "unknown_widget"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.active_template is None + + def test_appliance_disassembly_template(self, engine): + detections = [{"class": "power_tool"}, {"class": "safety_glasses"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.active_template == "appliance disassembly" + + +# --------------------------------------------------------------------------- +# OCR integration +# --------------------------------------------------------------------------- + +class TestOCRIntegration: + + def test_ocr_step_number_surfaced(self, engine): + result = engine.analyze([], hand_results=None, ocr_text="Step 3: Remove heat sink") + assert "step 3" in result.message.lower() or "step" in result.message.lower() + + def test_ocr_part_number_surfaced(self, engine): + result = engine.analyze([], hand_results=None, ocr_text="P/N: AB-5678") + assert "AB-5678" in result.message + + def test_empty_ocr_no_extra_message(self, engine): + result = engine.analyze([], hand_results=None, ocr_text="") + assert result.status == "OK" + assert result.message == "Step validation passed. Safe to proceed." + + +# --------------------------------------------------------------------------- +# Component identification +# --------------------------------------------------------------------------- + +class TestComponentIdentification: + + def test_known_components_mapped_correctly(self, engine): + detections = [{"class": "screwdriver"}, {"class": "m3_screw"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert "Tool: Screwdriver" in result.detected_components + assert "Fastener: M3 Screw" in result.detected_components + + def test_unknown_component_reported(self, engine): + detections = [{"class": "unknown_widget"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert any("Unknown" in c for c in result.detected_components) + + def test_no_duplicate_components(self, engine): + detections = [{"class": "screwdriver"}, {"class": "screwdriver"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.detected_components.count("Tool: Screwdriver") == 1 + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + +class TestEdgeCases: + + def test_empty_detections(self, engine): + result = engine.analyze([], hand_results=None, ocr_text="") + assert result.status == "OK" + assert result.critical_warning is False + assert result.detected_components == [] + + def test_uppercase_class_normalised(self, engine): + """CV models may return mixed-case class names — should still trigger rules.""" + detections = [{"class": "Circuit_Board"}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.status == "CRITICAL_WARNING" + + def test_missing_class_key_does_not_crash(self, engine): + """Detections missing the 'class' key should be handled gracefully.""" + detections = [{"bbox": [0, 0, 100, 100]}] + result = engine.analyze(detections, hand_results=None, ocr_text="") + assert result.status == "OK" + + def test_hand_results_none_does_not_crash(self, engine): + result = engine.analyze([], hand_results=None, ocr_text="") + assert result is not None \ No newline at end of file