Add heartbeat command

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 * Add reporting of roll-off error state and allow resetting on error.
 
+### 🏷️ Changed
+
+* [#28](https://vscode.dev/github/sdss/lvmecp/pull/28) Removed the automatic setting of the heartbeat variable. Added a `heartbeat` command that will be triggered by a heartbeat middleware.
+
 ### 🔧 Fixed
 
 * Restore GS3 status registers and fix addresses for roll-off lockout and error.

pyproject.toml

@@ -73,6 +73,7 @@ sdss = ["sdsstools", "clu"]
 [tool.pytest.ini_options]
 addopts = "--cov lvmecp --cov-report xml --cov-report html --cov-report term"
 asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
 
 [tool.coverage.run]
 branch = true

python/lvmecp/actor/actor.py

@@ -64,7 +64,6 @@ async def start(self, **kwargs):
         await super().start(**kwargs)
 
         asyncio.create_task(self.emit_status())
-        asyncio.create_task(self.emit_heartbeat())
 
         return self
 
@@ -75,17 +74,6 @@ async def emit_status(self, delay: float = 30.0):
             await self.send_command(self.name, "status", internal=True)
             await asyncio.sleep(delay)
 
-    async def emit_heartbeat(self, delay: float = 5.0):
-        """Updates the heartbeat Modbus variable to indicate the system is alive."""
-
-        while True:
-            try:
-                await self.plc.modbus["hb_set"].set(True)
-            except Exception:
-                self.write("w", "Failed to set heartbeat variable.")
-            finally:
-                await asyncio.sleep(delay)
-
     async def _check_internal(self):
         return await super()._check_internal()
 

python/lvmecp/actor/commands/heartbeat.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego ([email protected])
+# @Date: 2024-12-20
+# @Filename: heartbeat.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from . import parser
+
+
+if TYPE_CHECKING:
+    from lvmecp.actor import ECPCommand
+
+
+@parser.command()
+async def heartbeat(command: ECPCommand):
+    """Sets the heartbeat variable on the PLC."""
+
+    try:
+        await command.actor.plc.modbus["hb_set"].set(True)
+    except Exception:
+        return command.fail("Failed to set heartbeat.")
+    finally:
+        return command.finish()

python/lvmecp/plc.py

@@ -44,12 +44,12 @@
             # Allow for 3 seconds for broadcast. This is needed because the PLC
             # starts before the actor and for the first message the exchange is
             # not yet available.
-            n_tries = 0
+            elapsed: float = 0
             while actor.connection.connection is None:
-                n_tries += 1
-                if n_tries >= 3:
-                    return None
-                await asyncio.sleep(1)
+                elapsed += 0.01
+                if elapsed > 3:
+                    return
+                await asyncio.sleep(0.01)
             actor.write(level, message)
         elif command is not None:
             command.write(level, message)

tests/conftest.py

@@ -51,6 +51,7 @@ async def actor(simulator: Simulator, mocker):
     mocker.patch.object(_actor.plc.hvac.modbus, "get_all", return_value={})
 
     _actor = await setup_test_actor(_actor)  # type: ignore
+    _actor.connection.connection = mocker.MagicMock(spec={"is_closed": False})
 
     yield _actor
 

tests/test_command_heartbeat.py

@@ -0,0 +1,37 @@
+#!/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
+
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
+    from lvmecp.actor import ECPActor
+
+
+async def test_command_heartbeat(actor: ECPActor, mocker: MockerFixture):
+    hb_set_mock = mocker.patch.object(actor.plc.modbus["hb_set"], "set")
+
+    cmd = await actor.invoke_mock_command("heartbeat")
+    await cmd
+
+    assert cmd.status.did_succeed
+
+    hb_set_mock.assert_called_once_with(True)
+
+
+async def test_command_heartbeat_fails(actor: ECPActor, mocker: MockerFixture):
+    mocker.patch.object(actor.plc.modbus["hb_set"], "set", side_effect=Exception)
+
+    cmd = await actor.invoke_mock_command("heartbeat")
+    await cmd
+
+    assert cmd.status.did_fail