Implement handling of e-stops

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### 🚀 New
 
 * [#33](https://github.com/sdss/lvmecp/pull/33) Add a mode to close the dome using only drive overcurrent to determine when the movement has completed.
+* [#34](https://github.com/sdss/lvmecp/pull/34) Implement emergency stop command, resetting e-stops, and prevent dome from opening if safety flags are active.
 
 ### ✨ Improved
 

python/lvmecp/actor/commands/e_stop.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2025-01-07
+# @Filename: e-stop.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+import asyncio
+
+from typing import TYPE_CHECKING
+
+from . import parser
+
+
+if TYPE_CHECKING:
+    from lvmecp.actor import ECPCommand
+
+
+@parser.command()
+async def emergency_stop(command: ECPCommand):
+    """Trigger and emergency stop."""
+
+    await command.actor.plc.safety.emergency_stop()
+    await asyncio.sleep(0.1)
+
+    return command.finish(text="Emergency stop triggered.")

python/lvmecp/actor/commands/engineering.py

@@ -106,3 +106,12 @@ async def status(command: ECPCommand):
     """Returns the status of the engineering mode."""
 
     return command.finish(engineering_mode=await get_eng_mode_status(command.actor))
+
+
+@engineering_mode.command()
+async def reset_e_stops(command: ECPCommand):
+    """Resets the e-stop relays."""
+
+    await command.actor.plc.safety.reset_e_stops()
+
+    return command.finish()

python/lvmecp/dome.py

@@ -20,7 +20,7 @@
 
 from lvmecp import config, log
 from lvmecp.exceptions import DomeError
-from lvmecp.maskbits import DomeStatus
+from lvmecp.maskbits import DomeStatus, SafetyStatus
 from lvmecp.module import PLCModule
 
 
@@ -102,6 +102,9 @@ async def _move(
         if not (await self.plc.safety.is_remote()):
             raise DomeError("Cannot move dome while in local mode.")
 
+        if not self.plc.safety.status or self.plc.safety.status & SafetyStatus.E_STOP:
+            raise DomeError("E-stops are pressed.")
+
         if mode == "overcurrent" and open:
             raise DomeError("Cannot open dome in overcurrent mode.")
 

python/lvmecp/etc/lvmecp.yml

@@ -459,6 +459,8 @@ simulator:
     door_locked: true
     door_closed: true
     dome_closed: true
+    oxygen_read_utilities_room: 200
+    oxygen_read_spectrograph_room: 200
   events:
     ur_new:
       on_value: 1

python/lvmecp/safety.py

@@ -93,3 +93,13 @@ async def is_remote(self):
         assert self.status is not None and self.flag is not None
 
         return not (self.status & self.flag.LOCAL)
+
+    async def emergency_stop(self):
+        """Triggers an emergency stop."""
+
+        await self.plc.modbus["e_stop"].write(True)
+
+    async def reset_e_stops(self):
+        """Resets the E-stop relays."""
+
+        await self.plc.modbus["e_relay_reset"].write(True)

tests/test_command_dome.py

@@ -356,3 +356,18 @@ async def test_dome_close_while_opening(
 
     stop_mock.assert_called()
     assert "Stopping the dome before moving to the commanded position" in caplog.text
+
+
+async def test_dome_open_with_safety_alerts(
+    actor: ECPActor,
+    context: ModbusSlaveContext,
+    mocker: MockerFixture,
+):
+    mocker.patch.object(actor.plc.dome, "is_daytime", return_value=False)
+    context.setValues(1, actor.plc.modbus["e_status"].address, [1])
+
+    cmd = await actor.invoke_mock_command("dome open")
+    await cmd
+
+    assert cmd.status.did_fail
+    assert "E-stops are pressed" in cmd.replies.get("error")

tests/test_command_engineering_mode.py

@@ -14,6 +14,7 @@
 
 
 if TYPE_CHECKING:
+    from pymodbus.datastore import ModbusSlaveContext
     from pytest_mock import MockerFixture
 
     from lvmecp.actor import ECPActor
@@ -60,3 +61,17 @@ async def test_command_engineering_mode_timeouts(actor: ECPActor):
     await asyncio.sleep(0.3)
 
     assert actor.is_engineering_mode_enabled() is False
+
+
+async def test_command_engineering_mode_reset_e_stops(
+    actor: ECPActor,
+    context: ModbusSlaveContext,
+):
+    context.setValues(1, actor.plc.modbus["e_status"].address, [1])
+
+    cmd = await actor.invoke_mock_command("engineering-mode reset-e-stops")
+    await cmd
+
+    await asyncio.sleep(0.1)
+
+    assert context.getValues(1, actor.plc.modbus["e_status"].address, 1)[0] == 0