Add engineering mode

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Next version
 
+### 🚀 New
+
+* [#29](https://vscode.dev/github/sdss/lvmecp/pull/29) Add a new engineering mode that can be used to bypass the heartbeat and to allow the dome to open during daytime.
+
 ### ✨ Improved
 
 * Add reporting of roll-off error state and allow resetting on error.

python/lvmecp/actor/actor.py

@@ -10,6 +10,7 @@
 
 import asyncio
 import logging
+import time
 
 from lvmopstools.actor import ErrorCodesBase, LVMActor
 
@@ -28,6 +29,9 @@
 class ECPActor(LVMActor):
     """Enclosure actor."""
 
+    _engineering_mode_hearbeat_interval: float = 5
+    _engineering_mode_timeout: float = 30
+
     parser = parser
 
     def __init__(
@@ -61,9 +65,8 @@ def __init__(
 
         self._emit_status_task: asyncio.Task | None = None
 
-
-    async def start(self, **kwargs):
-        """Starts the actor."""
+        self._engineering_mode: bool = False
+        self._engineering_mode_task: asyncio.Task | None = None
 
         self.running: bool = False
 
@@ -85,6 +88,8 @@ async def stop(self, **kwargs):
         """Stops the actor."""
 
         self._emit_status_task = await cancel_task(self._emit_status_task)
+        self._engineering_mode_task = await cancel_task(self._engineering_mode_task)
+
         await super().stop(**kwargs)
         self.running = False
 
@@ -97,6 +102,55 @@ async def emit_status(self, delay: float = 30.0):
             await self.send_command(self.name, "status", internal=True)
             await asyncio.sleep(delay)
 
+    async def engineering_mode(
+        self,
+        enable: bool,
+        timeout: float | None = None,
+    ):
+        """Sets or returns the engineering mode."""
+
+        # Kill current task if it exists.
+        self._engineering_mode_task = await cancel_task(self._engineering_mode_task)
+
+        if enable:
+            self._engineering_mode_task = asyncio.create_task(
+                self._run_eng_mode(timeout)
+            )
+
+        self._engineering_mode = enable
+
+    def is_engineering_mode_enabled(self):
+        """Returns whether engineering mode is enabled."""
+
+        return self._engineering_mode
+
+    async def _run_eng_mode(self, timeout: float | None = None):
+        """Runs the engineering mode.
+
+        Emits a heartbeat every N seconds even if we are not receiving heartbeat
+        commands from ``lvmbeat``. Monitors how long we have been in engineering
+        mode and disables it after a timeout.
+
+        """
+
+        started_at: float = time.time()
+        timeout = timeout or self._engineering_mode_timeout
+
+        while True:
+            await self.emit_heartbeat()
+
+            if time.time() - started_at > timeout:
+                self.write("w", text="Engineering mode timed out and was disabled.")
+                await self.engineering_mode(False)
+                return
+
+            await asyncio.sleep(self._engineering_mode_hearbeat_interval)
+
+    async def emit_heartbeat(self):
+        """Emits a heartbeat to the PLC."""
+
+        await self.plc.modbus["hb_set"].set(True)
+
     async def _check_internal(self):
         return await super()._check_internal()
 

python/lvmecp/actor/commands/engineering.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2024-12-24
+# @Filename: engineering.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+from . import parser
+
+
+if TYPE_CHECKING:
+    from lvmecp.actor import ECPCommand
+
+
+@parser.group(name="engineering-mode")
+def engineering_mode():
+    """Enable/disable the engineering mode."""
+
+    pass
+
+
+@engineering_mode.command()
+@click.option(
+    "--timeout",
+    "-t",
+    type=float,
+    help="Timeout for the engineering mode. "
+    "If not passed, the default timeout is used.",
+)
+async def enable(command: ECPCommand, timeout: float | None = None):
+    """Enables the engineering mode."""
+
+    await command.actor.engineering_mode(True, timeout=timeout)
+
+    return command.finish(engineering_mode=True)
+
+
+@engineering_mode.command()
+async def disable(command: ECPCommand):
+    """Disables the engineering mode."""
+
+    await command.actor.engineering_mode(False)
+
+    return command.finish(engineering_mode=False)
+
+
+@engineering_mode.command()
+async def status(command: ECPCommand):
+    """Returns the status of the engineering mode."""
+
+    return command.finish(engineering_mode=command.actor.is_engineering_mode_enabled())

python/lvmecp/actor/commands/heartbeat.py

@@ -22,7 +22,7 @@ async def heartbeat(command: ECPCommand):
     """Sets the heartbeat variable on the PLC."""
 
     try:
-        await command.actor.plc.modbus["hb_set"].set(True)
+        await command.actor.emit_heartbeat()
     except Exception:
         return command.fail("Failed to set heartbeat.")
     else:

python/lvmecp/etc/schema.json

@@ -22,7 +22,8 @@
       "type": "string"
     },
     "o2_percent_utilities": { "type": "number" },
-    "o2_percent_spectrograph": { "type": "number" }
+    "o2_percent_spectrograph": { "type": "number" },
+    "engineering_mode": { "type": "boolean" }
   },
   "additionalProperties": true
 }

tests/test_command_engineering_mode.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2024-12-24
+# @Filename: test_command_heartbeat.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+import asyncio
+
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
+    from lvmecp.actor import ECPActor
+
+
+async def test_command_engineering_mode_status(actor: ECPActor):
+    cmd = await actor.invoke_mock_command("engineering-mode status")
+    await cmd
+
+    assert cmd.status.did_succeed
+    assert cmd.replies.get("engineering_mode") is False
+
+
+async def test_command_engineering_mode_enable(actor: ECPActor, mocker: MockerFixture):
+    eng_mode_mock = mocker.patch.object(actor, "engineering_mode")
+
+    cmd = await actor.invoke_mock_command("engineering-mode enable --timeout 10")
+    await cmd
+
+    assert cmd.status.did_succeed
+    eng_mode_mock.assert_called_once_with(True, timeout=10)
+
+
+async def test_command_engineering_mode_no_mock(actor: ECPActor):
+    cmd = await actor.invoke_mock_command("engineering-mode enable --timeout 10")
+    await cmd
+
+    assert actor.is_engineering_mode_enabled() is True
+
+    cmd = await actor.invoke_mock_command("engineering-mode disable")
+    await cmd
+
+    assert actor.is_engineering_mode_enabled() is False
+
+
+async def test_command_engineering_mode_timeouts(actor: ECPActor):
+    actor._engineering_mode_hearbeat_interval = 0.1  # To speed up the test
+
+    cmd = await actor.invoke_mock_command("engineering-mode enable --timeout 0.2")
+    await cmd
+
+    assert actor.is_engineering_mode_enabled() is True
+
+    await asyncio.sleep(0.3)
+
+    assert actor.is_engineering_mode_enabled() is False