Add anti-flap mechanism

python/lvmecp/actor/commands/dome.py

@@ -42,7 +42,7 @@ async def open(command: ECPCommand, force=False):
     try:
         await command.actor.plc.dome.open(force=force)
     except DomeError as err:
-        return command.fail(err)
+        return command.fail(f"Dome failed to open with error: {err}")
 
     status = command.actor.plc.dome.status
     if status and status & DomeStatus.OPEN:
@@ -61,7 +61,7 @@ async def close(command: ECPCommand, force=False):
     try:
         await command.actor.plc.dome.close(force=force)
     except DomeError as err:
-        return command.fail(err)
+        return command.fail(f"Dome failed to close with error: {err}")
 
     status = command.actor.plc.dome.status
     if status and status & DomeStatus.CLOSED:

python/lvmecp/dome.py

@@ -31,6 +31,9 @@ class DomeController(PLCModule[DomeStatus]):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
+        # Timestamps when we have opened the dome. For the anti-flap mechanism.
+        self._open_attempt_times: list[float] = []
+
     async def _update_internal(self, use_cache: bool = True, **kwargs):
         dome_registers = await self.plc.modbus.read_group("dome", use_cache=use_cache)
 
@@ -148,8 +151,10 @@ async def _move(self, open: bool, force: bool = False):
     async def open(self, force: bool = False):
         """Open the dome."""
 
+        self._open_attempt_times.append(time())
+
         if not self.is_allowed():
-            raise DomeError("Dome cannot be opened during daytime.")
+            raise DomeError("Dome is not allowed to open.")
 
         await self._move(True, force=force)
 
@@ -180,32 +185,34 @@ async def reset(self):
         await asyncio.sleep(1)
 
     def is_allowed(self):
-        """Returns whether the dome is allowed to move.
-
-        Currently the only check performed is to confirm that it is not daytime,
-        but this method could be expanded in the future.
-
-        """
-
-        is_daytime: bool | None
-        if not config["dome.daytime_allowed"]:
-            is_daytime = self.is_daytime()
-        else:
-            is_daytime = None
-
-        if not is_daytime:
-            return True
+        """Returns whether the dome is allowed to move."""
 
         if self.plc._actor and self.plc._actor._engineering_mode:
             self.plc._actor.write(
                 "w",
-                text="Daytime detected but engineering mode is active. "
-                "Allowing to open the dome.",
+                text="Skipping dome tests due to engineering mode.",
             )
-
             return True
 
-        return False
+        if not config["dome.daytime_allowed"] and self.is_daytime():
+            raise DomeError("Dome is not allowed to open during daytime.")
+
+        anti_flap_n, anti_flap_interval = config["dome.anti_flap_tolerance"] or [3, 600]
+
+        attempts_in_interval: list[float] = []
+        for tt in self._open_attempt_times[::-1]:
+            if time() - tt < anti_flap_interval:
+                attempts_in_interval.append(tt)
+            else:
+                break
+
+        if len(attempts_in_interval) >= anti_flap_n:
+            raise DomeError(
+                "Too many open attempts in a short interval. "
+                f"Wait {anti_flap_interval} seconds before trying again."
+            )
+
+        return True
 
     def is_daytime(self):  # pragma: no cover
         """Returns whether it is daytime."""

python/lvmecp/etc/lvmecp.yml

@@ -175,6 +175,7 @@ safety:
 dome:
   daytime_allowed: false
   daytime_tolerance: 600
+  anti_flap_tolerance: [3, 600]
 
 hvac:
   host: 10.8.38.49

tests/test_command_dome.py

@@ -12,8 +12,11 @@
 
 from typing import TYPE_CHECKING
 
+import pytest
+
 import lvmecp.actor.actor
 import lvmecp.dome
+from lvmecp.exceptions import DomeError
 from lvmecp.maskbits import DomeStatus
 
 
@@ -57,11 +60,16 @@ async def test_command_dome_daytime(actor: ECPActor, mocker: MockerFixture):
 
 
 async def test_command_dome_daytime_allowed(actor: ECPActor, mocker: MockerFixture):
-    mocker.patch.object(
-        lvmecp.dome,
-        "config",
-        return_value={"dome": {"daytime_allowed": True}},
+    mocker.patch.dict(
+        "lvmecp.dome.config",
+        {
+            "dome": {
+                "daytime_allowed": True,
+                "anti_flap_tolerance": [3, 600],
+            }
+        },
     )
+
     mocker.patch.object(actor.plc.dome, "is_daytime", return_value=True)
     mocker.patch.object(actor.plc.dome, "_move", return_value=True)
 
@@ -113,3 +121,20 @@ async def test_actor_daytime_task_eng_mode(actor: ECPActor, mocker: MockerFixtur
     dome_close_mock.assert_not_called()
 
     task.cancel()
+
+
+async def test_dome_anti_flap(actor: ECPActor, mocker: MockerFixture):
+    mocker.patch.dict("lvmecp.dome.config", {"dome": {"anti_flap_tolerance": [3, 1]}})
+
+    mocker.patch.object(actor.plc.dome, "is_daytime", return_value=False)
+    mocker.patch.object(actor.plc.dome, "_move", return_value=True)
+
+    await actor.plc.dome.open()
+    await asyncio.sleep(0.1)
+
+    await actor.plc.dome.open()
+    await asyncio.sleep(0.1)
+
+    with pytest.raises(DomeError):
+        await actor.plc.dome.open()
+        await asyncio.sleep(0.1)