Add mode to close dome using overcurrent only

.github/workflows/docker.yml

@@ -8,6 +8,10 @@ on:
       - '*'
     paths-ignore:
       - 'docs/**'
+  pull_request:
+    branches: [main]
+    paths-ignore:
+      - 'docs/**'
 
 jobs:
   docker:
@@ -30,6 +34,10 @@ jobs:
           then
             BRANCH=$(echo ${GITHUB_REF#refs/heads/} | sed -r 's/[\/]/_/g')
             echo TAGS=$USER/$APP:$BRANCH >> $GITHUB_OUTPUT
+          elif [[ $GITHUB_REF == refs/pull/* ]]
+          then
+            BRANCH=${{ github.head_ref || github.ref_name }}
+            echo TAGS=$USER/$APP:$BRANCH >> $GITHUB_OUTPUT
           else
             echo TAGS=$USER/$APP:${GITHUB_REF#refs/tags/} >> $GITHUB_OUTPUT
           fi

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Next version
+
+### 🚀 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.
+
 ## 1.1.0 - January 1, 2025
 
 ### ✨ Improved

python/lvmecp/actor/commands/dome.py

@@ -34,7 +34,7 @@ def dome():
 
 @dome.command()
 @click.option("--force", is_flag=True, help="Force dome opening.")
-async def open(command: ECPCommand, force=False):
+async def open(command: ECPCommand, force: bool = False):
     """Opens the dome."""
 
     command.info("Opening dome.")
@@ -53,13 +53,21 @@ async def open(command: ECPCommand, force=False):
 
 @dome.command()
 @click.option("--force", is_flag=True, help="Force dome closing.")
-async def close(command: ECPCommand, force=False):
+@click.option(
+    "--overcurrent",
+    is_flag=True,
+    help="Close the dome using the overcurrent mode.",
+)
+async def close(command: ECPCommand, force: bool = False, overcurrent: bool = False):
     """Closes the dome."""
 
     command.info("Closing dome.")
 
     try:
-        await command.actor.plc.dome.close(force=force)
+        await command.actor.plc.dome.close(
+            force=force,
+            mode="overcurrent" if overcurrent else "normal",
+        )
     except DomeError as err:
         return command.fail(f"Dome failed to close with error: {err}")
 

python/lvmecp/dome.py

@@ -13,6 +13,8 @@
 from time import time
 from types import SimpleNamespace
 
+from typing import Literal
+
 import numpy
 from astropy.time import Time
 from lvmopstools.ephemeris import get_ephemeris_summary
@@ -25,6 +27,8 @@
 
 MOVE_CHECK_INTERVAL: float = 0.5
 
+DRIVE_MODE_TYPE = Literal["normal", "overcurrent"]
+
 
 class DomeController(PLCModule[DomeStatus]):
     """Controller for the rolling dome."""
@@ -90,12 +94,20 @@
         await self.modbus["drive_direction"].write(open)
         await self.update(use_cache=False)
 
-    async def _move(self, open: bool, force: bool = False):
+    async def _move(
+        self,
+        open: bool,
+        force: bool = False,
+        mode: DRIVE_MODE_TYPE = "normal",
+    ):
         """Moves the dome to open/close position."""
 
         if not (await self.plc.safety.is_remote()):
             raise DomeError("Cannot move dome while in local mode.")
 
+        if mode == "overcurrent" and open:
+            raise DomeError("Cannot open dome in overcurrent mode.")
+
         await self.update(use_cache=False)
 
         assert self.status is not None and self.flag is not None
@@ -125,6 +137,13 @@
             else:
                 warnings.warn("Dome already at position, but forcing.", ECPWarning)
 
+        if mode == "normal":
+            log.debug("Setting drive mode to normal.")
+            await self.modbus["drive_mode"].write(0)
+        elif mode == "overcurrent":
+            log.debug("Setting drive mode to overcurrent.")
+            await self.modbus["drive_mode"].write(1)
+
         log.debug("Setting motor_direction.")
         await self.modbus["motor_direction"].write(open)
 
@@ -157,6 +176,9 @@
             if not drive_enabled and (time() - last_enabled) > 5:
                 raise DomeError("Dome drive has been disabled.")
 
+        # Reset drive_mode.
+        await self.modbus["drive_mode"].write(0)
+
         await self.update(use_cache=False)
 
     async def open(self, force: bool = False):
@@ -169,10 +191,10 @@
 
         await self._move(True, force=force)
 
-    async def close(self, force: bool = False):
+    async def close(self, force: bool = False, mode: DRIVE_MODE_TYPE = "normal"):
         """Close the dome."""
 
-        await self._move(False, force=force)
+        await self._move(False, force=force, mode=mode)
 
     async def stop(self):
         """Stops the dome."""

python/lvmecp/etc/lvmecp.yml

@@ -93,6 +93,11 @@ modbus:
       group: lights
       mode: coil
       readonly: true
+    drive_mode:
+      address: 98
+      group: dome
+      mode: coil
+      readonly: false
     drive_enabled:
       address: 99
       group: dome

tests/test_command_dome.py

@@ -147,6 +147,41 @@ async def test_command_dome_close(actor: ECPActor, mocker: MockerFixture):
     assert cmd.status.did_succeed
 
 
+async def test_command_dome_close_overcurrent(
+    actor: ECPActor,
+    context: ModbusSlaveContext,
+    mocker: MockerFixture,
+):
+    async def close_with_delay():
+        await asyncio.sleep(0.3)
+
+        # drive_mode should be 1 (overcurrent) after we call _move but before the
+        # move completes.
+        assert context.getValues(1, actor.plc.modbus["drive_mode"].address)[0] == 1
+
+        context.setValues(1, actor.plc.modbus["dome_open"].address, [0])
+        context.setValues(1, actor.plc.modbus["dome_closed"].address, [1])
+        context.setValues(1, actor.plc.modbus["drive_enabled"].address, [0])
+
+    mocker.patch.object(lvmecp.dome, "MOVE_CHECK_INTERVAL", 0.1)
+
+    # Simulate open dome.
+    context.setValues(1, actor.plc.modbus["dome_open"].address, [1])
+    context.setValues(1, actor.plc.modbus["dome_closed"].address, [0])
+
+    cmd = await actor.invoke_mock_command("dome close --overcurrent")
+
+    await asyncio.sleep(0.1)
+    asyncio.create_task(close_with_delay())
+
+    await cmd
+
+    assert cmd.status.did_succeed
+
+    # drive_mode should have been reset to 0 after the closure.
+    assert context.getValues(1, actor.plc.modbus["drive_mode"].address)[0] == 0
+
+
 async def test_command_dome_daytime(actor: ECPActor, mocker: MockerFixture):
     mocker.patch.object(actor.plc.dome, "is_daytime", return_value=True)
     mocker.patch.object(actor.plc.dome, "_move", return_value=True)