Handle e-stops in Enclosure and Overwatcher

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Next version
 
+### 🚀 New
+
+* Monitor and handle e-stops in the Overwatcher.
+
 ### ✨ Improved
 
 * Require two consecutive ping failures before restarting an actor.

src/gort/devices/enclosure.py

@@ -17,6 +17,7 @@
 from gort.devices.core import GortDevice, GortDeviceSet
 from gort.enums import Event
 from gort.exceptions import ErrorCode, GortEnclosureError, GortTelescopeError
+from gort.gort import Gort
 
 
 if TYPE_CHECKING:
@@ -104,6 +105,45 @@ async def dome_all_off(self):
         def __getattr__(self, light: str) -> Light: ...
 
 
+class E_Stops:
+    """Reports status of the emergency stops buttons and allows to reset the relays."""
+
+    def __init__(self, enclosure: Enclosure):
+        self.enclosure = enclosure
+        self.ecp = enclosure.actor
+        self.gort = enclosure.gort
+
+    @Retrier(max_attempts=3, delay=0.5)
+    async def status(self):
+        """Returns :obj:`True` if the emergency stops are pressed."""
+
+        status = await self.enclosure.status()
+        labels = status.get("safety_status_labels", None)
+
+        if labels is None:
+            raise GortEnclosureError("Cannot determine the status of the e-stops.")
+
+        return "E_STOP" in labels
+
+    @Retrier(max_attempts=3, delay=0.5)
+    async def trigger(self):
+        """Triggers the e-stop relays."""
+
+        try:
+            await self.ecp.commands.emergency_stop()
+        except Exception as err:
+            raise GortEnclosureError(f"Failed to trigger the e-stops: {err}")
+
+    @Retrier(max_attempts=3, delay=0.5)
+    async def reset(self):
+        """Resets the e-stop relays after the e-stops have been released."""
+
+        try:
+            await self.ecp.commands.engineering_mode.commands.reset_e_stops()
+        except Exception as err:
+            raise GortEnclosureError(f"Failed to reset the e-stops: {err}")
+
+
 class Enclosure(GortDevice):
     """Class representing the LVM enclosure."""
 
@@ -113,6 +153,7 @@ def __init__(self, gort: Gort, name: str, actor: str, **kwargs):
         super().__init__(gort, name, actor)
 
         self.lights = Lights(self)
+        self.e_stops = E_Stops(self)
 
     async def restart(self):
         """Restarts the ``lvmecp`` deployment."""

src/gort/overwatcher/alerts.py

@@ -38,6 +38,7 @@ class AlertsSummary(BaseModel):
     door_alert: bool | None = None
     camera_temperature_alert: bool | None = None
     camera_alerts: dict[str, bool] | None = None
+    e_stops: bool | None = None
     o2_alert: bool | None = None
     o2_room_alerts: dict[str, bool] | None = None
     heater_alert: bool | None = None
@@ -62,6 +63,7 @@ class ActiveAlert(enum.Flag):
     DOOR = enum.auto()
     CAMERA_TEMPERATURE = enum.auto()
     O2 = enum.auto()
+    E_STOPS = enum.auto()
     LOCKED = enum.auto()
     UNAVAILABLE = enum.auto()
     DISCONNECTED = enum.auto()
@@ -178,6 +180,11 @@ def is_safe(self) -> tuple[bool, ActiveAlert]:
             active_alerts |= ActiveAlert.WIND
             is_safe = False
 
+        if self.state.e_stops:
+            self.log.warning("E-stops triggered.")
+            active_alerts |= ActiveAlert.E_STOPS
+            is_safe = False
+
         if self.connectivity.internet.is_set():
             self.log.warning("Internet connectivity lost.")
             active_alerts |= ActiveAlert.DISCONNECTED
@@ -219,6 +226,12 @@ async def update_status(self) -> AlertsSummary:
         alerts_data = await get_lvmapi_route("/alerts/summary")
         summary = AlertsSummary(**alerts_data)
 
+        try:
+            summary.e_stops = await self.gort.enclosure.e_stops.status()
+        except Exception:
+            self.log.warning("Failed to retrieve e-stop status.")
+            pass
+
         # For connectivity we want to avoid one single failure to trigger an alert
         # which closes the dome. The connectivity status is a set of triggers that
         # need several settings to be activated.

src/gort/overwatcher/overwatcher.py

@@ -141,7 +141,9 @@ async def handle_unsafe(self):
         calibrating = ow.state.calibrating
 
         _, alerts_status = ow.alerts.is_safe()
+
         is_raining = bool(alerts_status & ActiveAlert.RAIN)
+        e_stops_in = bool(alerts_status & ActiveAlert.E_STOPS)
 
         if not closed or observing or calibrating:
             try:
@@ -159,10 +161,6 @@ async def handle_unsafe(self):
                                 f"Error stopping observing: {decap(err)}",
                                 level="error",
                             )
-                            await ow.notify(
-                                "I will close the dome anyway.",
-                                level="warning",
-                            )
 
                     if calibrating:
                         await ow.calibrations.cancel()
@@ -180,7 +178,7 @@ async def handle_unsafe(self):
                 )
 
             finally:
-                if not closed:
+                if not closed and not e_stops_in:
                     await ow.notify("Closing the dome due to unsafe conditions.")
                     await ow.dome.shutdown(retry=True, park=True)
 
@@ -194,6 +192,12 @@ async def handle_unsafe(self):
                         level="warning",
                     )
 
+                elif e_stops_in:
+                    await ow.notify(
+                        "E-stops triggered. The dome will not be closed.",
+                        level="warning",
+                    )
+
         if ow.state.enabled:
             if is_raining:
                 await ow.notify(
@@ -202,6 +206,13 @@ async def handle_unsafe(self):
                     level="warning",
                 )
                 ow.state.enabled = False
+            elif e_stops_in:
+                await ow.notify(
+                    "Disabling the Overwatcher due to e-stops. "
+                    "Manually re-enable it when safe.",
+                    level="warning",
+                )
+                ow.state.enabled = False
 
     async def handle_daytime(self):
         """Handles daytime."""

src/gort/overwatcher/safety.py

@@ -15,6 +15,7 @@
 
 from lvmopstools.retrier import Retrier
 
+from gort.overwatcher.alerts import ActiveAlert
 from gort.overwatcher.core import OverwatcherModule, OverwatcherModuleTask
 from gort.tools import decap
 
@@ -122,9 +123,13 @@ async def task(self):
     async def get_data(self):
         """Returns the safety status and whether the dome is open."""
 
-        is_safe, _ = self.overwatcher.alerts.is_safe()
+        is_safe, status = self.overwatcher.alerts.is_safe()
         dome_open = await self.overwatcher.gort.enclosure.is_open()
 
+        # If is_safe=False but the e-stops are in, we do not want to close the dome.
+        if not is_safe and status & ActiveAlert.E_STOPS:
+            is_safe = True
+
         return is_safe, dome_open