Update robotics_diagnostician.py

robotics_diagnostician.py

@@ -1,32 +1,65 @@
+"""
+Root cause analysis for robotic system failures.
+Works with IoTEvent objects that contain temperature, vibration, motor_current, position_error.
+"""
+
 import logging
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 from agentic_reliability_framework.runtime.agents.base import BaseAgent, AgentSpecialization
-from agentic_reliability_framework.core.models.event import ReliabilityEvent
+from iot_event import IoTEvent  # Use the specific event type (though base ReliabilityEvent also works)
 
 logger = logging.getLogger(__name__)
 
+
 class RoboticsDiagnostician(BaseAgent):
-    """Root cause analysis for robotic system failures."""
-    def __init__(self):
+    """
+    Diagnoses robotic system failures by comparing sensor readings against thresholds.
+    Intended to be used with IoTEvent instances.
+    """
+
+    def __init__(self, thresholds: Optional[Dict[str, float]] = None):
+        """
+        Args:
+            thresholds: Optional dict overriding default sensor thresholds.
+                       Keys: temperature, vibration, motor_current, position_error.
+        """
         super().__init__(AgentSpecialization.DIAGNOSTICIAN)
-        # Define thresholds for each sensor
-        self.thresholds = {
+        self.thresholds = thresholds or {
             'temperature': 40.0,
             'vibration': 0.3,
             'motor_current': 3.5,
             'position_error': 0.05
         }
 
-    async def analyze(self, event: ReliabilityEvent) -> Dict[str, Any]:
+    async def analyze(self, event: IoTEvent) -> Dict[str, Any]:
+        """
+        Analyze sensor readings and return diagnosis.
+
+        Args:
+            event: An IoTEvent containing sensor fields.
+
+        Returns:
+            Dictionary with keys:
+                - specialization: 'robotics'
+                - confidence: float (0‑1)
+                - findings: dict with 'exceeded_thresholds' and 'likely_root_causes'
+                - recommendations: list of strings
+        """
         try:
-            # Extract metrics from event (assuming they are stored in custom fields)
+            # Extract metrics (safe even if event is base ReliabilityEvent without these fields)
             temp = getattr(event, 'temperature', None)
             vib = getattr(event, 'vibration', None)
             curr = getattr(event, 'motor_current', None)
             pos_err = getattr(event, 'position_error', None)
 
             if temp is None:
-                return {'specialization': 'robotics', 'findings': {}, 'recommendations': []}
+                # Event does not contain IoT fields – cannot diagnose
+                return {
+                    'specialization': 'robotics',
+                    'confidence': 0.0,
+                    'findings': {},
+                    'recommendations': []
+                }
 
             # Detect which thresholds are exceeded
             flags = []
@@ -65,4 +98,9 @@ class RoboticsDiagnostician(BaseAgent):
             }
         except Exception as e:
             logger.error(f"RoboticsDiagnostician error: {e}", exc_info=True)
-            return {'specialization': 'robotics', 'findings': {}, 'recommendations': []}
+            return {
+                'specialization': 'robotics',
+                'confidence': 0.0,
+                'findings': {},
+                'recommendations': []
+            }