Fix System Failure Tracer Logic (#214)

* Implemented a stringifier. (#210)

* Separated callback methods into system callback and device callback. (#210)

* Separated callback methods into system callback and device callback. (#210)

* Included error in `on_device_fail()`. (#210)

Author: ATATC

leads/dt/device.py

@@ -15,6 +15,10 @@ def __init__(self, *pins: int | str) -> None:
         self._parent_tags: tuple[str, ...] = ()
         self._pins: tuple[int | str, ...] = pins
 
+    @_override
+    def __str__(self) -> str:
+        return f"{len(self._parent_tags)}.{self._tag}"
+
     def level(self) -> int:
         """
         Get the level of the device in the device tree.

leads/sft.py

@@ -19,25 +19,39 @@ def read_marked_system(device: Device) -> list[str] | None:
 class SystemFailureTracer(object):
     def __init__(self) -> None:
         super().__init__()
-        self.on_fail: _Callable[[Device, SuspensionEvent], None] = lambda _, __: None
-        self.on_recover: _Callable[[Device, SuspensionEvent], None] = lambda _, __: None
+        self.on_fail: _Callable[[SuspensionEvent], None] = lambda _: None
+        self.on_recover: _Callable[[SuspensionEvent], None] = lambda _: None
+        self.on_device_fail: _Callable[[Device, str | Exception], None] = lambda _, __: None
+        self.on_device_recover: _Callable[[Device], None] = lambda _: None
+        self._system_failure: dict[str, int] = {}
 
     def fail(self, device: Device, error: str | Exception) -> None:
         if isinstance(error, Exception):
             error = repr(error)
         if not (systems := read_marked_system(device)):
             raise RuntimeWarning(f"No system marked for device {device}")
+        self.on_device_fail(device, error)
+        L.error(f"{device} error: {error}")
         for system in systems:
-            self.on_fail(device, e := SuspensionEvent(context := require_context(), system, error))
+            if system not in self._system_failure:
+                self._system_failure[system] = 0
+            self._system_failure[system] += 1
+            self.on_fail(e := SuspensionEvent(context := require_context(), system, error))
             context.suspend(e)
-            L.error(f"{system} error: {error}")
 
     def recover(self, device: Device) -> None:
         if not (systems := read_marked_system(device)):
             raise RuntimeWarning(f"System not marked for device {device}")
+        self.on_device_recover(device)
+        L.debug(f"{device} recovered")
         for system in systems:
-            self.on_recover(device, SuspensionEvent(require_context(), system, "Recovered"))
-            L.info(f"{system} recovered")
+            if system not in self._system_failure:
+                continue
+            self._system_failure[system] -= 1
+            if self._system_failure[system] > 0:
+                continue
+            self._system_failure.pop(system)
+            self.on_recover(SuspensionEvent(require_context(), system, "Recovered"))
 
 
 SFT: SystemFailureTracer = SystemFailureTracer()

leads_vec/cli.py

@@ -250,7 +250,7 @@ def hazard(self, e: Event, state: bool) -> None:
     uim["motor_fault"] = _Label(root, text="")
     uim["wsc_fault"] = _Label(root, text="")
 
-    def on_fail(_, e: SuspensionEvent) -> None:
+    def on_fail(e: SuspensionEvent) -> None:
         match e.system:
             case "BATT":
                 uim["battery_fault"].configure(image=Battery(color=Color.RED))
@@ -267,7 +267,7 @@ def on_fail(_, e: SuspensionEvent) -> None:
 
     SFT.on_fail = on_fail
 
-    def on_recover(_, e: SuspensionEvent) -> None:
+    def on_recover(e: SuspensionEvent) -> None:
         match e.system:
             case "BATT":
                 uim["battery_fault"].configure(image=None)