Merge pull request #172 from lsst-ts/tickets/DM-46309

tickets/DM-46309: Implement open and close mirror covers in MTCS

doc/news/DM-46309.feature.rst

@@ -0,0 +1 @@
+Implement open and close mirror covers methods in MTCS.

python/lsst/ts/observatory/control/maintel/mtcs.py

@@ -32,7 +32,7 @@
 from astropy.coordinates import Angle
 from lsst.ts import salobj, utils
 from lsst.ts.utils import angle_diff
-from lsst.ts.xml.enums import MTM1M3, MTM2, MTDome, MTPtg, MTRotator
+from lsst.ts.xml.enums import MTM1M3, MTM2, MTDome, MTMount, MTPtg, MTRotator
 
 try:
     from lsst.ts.xml.tables.m1m3 import FATable
@@ -134,6 +134,7 @@ def __init__(
         self.tel_flat_el = 39.0
         self.tel_flat_az = 205.7
         self.tel_settle_time = 3.0
+        self.tel_operate_mirror_covers_el = 70.0
 
         self.dome_park_az = 285.0
         self.dome_park_el = 80.0
@@ -165,6 +166,9 @@ def __init__(
             [(fa.actuator_id, fa.s_index) for fa in FATable if fa.s_index is not None]
         )
 
+        # Mirror covers operation timeout, in seconds.
+        self.mirror_covers_timeout = 120.0
+
         try:
             self._create_asyncio_events()
         except RuntimeError:
@@ -773,8 +777,69 @@ async def close_dome(self) -> None:
         raise NotImplementedError("# TODO: Implement (DM-21336).")
 
     async def close_m1_cover(self) -> None:
-        # TODO: Implement (DM-21336).
-        raise NotImplementedError("# TODO: Implement (DM-21336).")
+        """Method to close mirror covers.
+
+        Warnings
+        --------
+        The mirror covers should be closed when the telescope is pointing to
+        the zenith. The method will check if the telescope is in an operational
+        range and, if not, will move the telescope to an operational elevation,
+        maintaining the same azimuth before closing the mirror cover. The
+        telescope will be left in that same position in the end.
+
+        Raises
+        ------
+        RuntimeError
+            If mirror covers state is neither DEPLOYED nor RETRACTED.
+            If mirror system state is FAULT.
+        """
+        self.rem.mtmount.evt_mirrorCoversMotionState.flush()
+        cover_state = await self.rem.mtmount.evt_mirrorCoversMotionState.aget(
+            timeout=self.fast_timeout
+        )
+        self.log.debug(
+            f"Cover state: {MTMount.DeployableMotionState(cover_state.state)!r}"
+        )
+
+        if cover_state.state == MTMount.DeployableMotionState.DEPLOYED:
+            self.log.info("Mirror covers already closed.")
+        elif cover_state.state == MTMount.DeployableMotionState.RETRACTED:
+            self.log.info("Closing mirror covers.")
+
+            # Mirror covers shall close at zenith pointing.
+            if not await self.in_m1_cover_operational_range():
+                await self.slew_to_m1_cover_operational_range()
+
+            await self.rem.mtmount.cmd_closeMirrorCovers.start(
+                timeout=self.long_long_timeout
+            )
+
+            while cover_state.state != MTMount.DeployableMotionState.DEPLOYED:
+                cover_state = await self.rem.mtmount.evt_mirrorCoversMotionState.next(
+                    flush=False, timeout=self.mirror_covers_timeout
+                )
+                self.log.debug(
+                    f"Cover state: {MTMount.DeployableMotionState(cover_state.state)!r}"
+                )
+                cover_system_state = (
+                    await self.rem.mtmount.evt_mirrorCoversSystemState.aget(
+                        timeout=self.fast_timeout
+                    )
+                )
+                if cover_system_state.state == MTMount.PowerState.FAULT:
+                    raise RuntimeError(
+                        "Close cover failed. Cover system state: "
+                        f"{MTMount.PowerState(cover_system_state.state)!r}"
+                    )
+            self.log.info(
+                f"Cover state: {MTMount.DeployableMotionState(cover_state.state)!r}"
+            )
+        else:
+            raise RuntimeError(
+                f"Mirror covers in {MTMount.DeployableMotionState(cover_state.state)!r} "
+                f"state. Expected {MTMount.DeployableMotionState.RETRACTED!r} or "
+                f"{MTMount.DeployableMotionState.DEPLOYED!r}"
+            )
 
     async def home_dome(self) -> None:
         # TODO: Implement (DM-21336).
@@ -785,8 +850,107 @@ async def open_dome_shutter(self) -> None:
         raise NotImplementedError("# TODO: Implement (DM-21336).")
 
     async def open_m1_cover(self) -> None:
-        # TODO: Implement (DM-21336).
-        raise NotImplementedError("# TODO: Implement (DM-21336).")
+        """Method to open mirror covers.
+
+        Warnings
+        --------
+        The mirror covers should be opened when the telescope is pointing to
+        the zenith. The method will check if the telescope is in an operational
+        range and, if not, will move the telescope to an operational elevation,
+        maintaining the same azimuth before opening the mirror cover. The
+        telescope will be left in that same position in the end.
+
+        Raises
+        ------
+        RuntimeError
+            If mirror covers state is neither DEPLOYED nor RETRACTED.
+            If mirror system state is FAULT.
+        """
+        self.rem.mtmount.evt_mirrorCoversMotionState.flush()
+        cover_state = await self.rem.mtmount.evt_mirrorCoversMotionState.aget(
+            timeout=self.fast_timeout
+        )
+        self.log.debug(
+            f"Cover state: {MTMount.DeployableMotionState(cover_state.state)!r}"
+        )
+
+        if cover_state.state == MTMount.DeployableMotionState.RETRACTED:
+            self.log.info("Mirror covers already opened.")
+        elif cover_state.state == MTMount.DeployableMotionState.DEPLOYED:
+            self.log.info("Opening mirror covers.")
+
+            # Mirror covers shall open at zenith pointing.
+            if not await self.in_m1_cover_operational_range():
+                await self.slew_to_m1_cover_operational_range()
+
+            await self.rem.mtmount.cmd_openMirrorCovers.set_start(
+                leaf=MTMount.MirrorCover.ALL, timeout=self.long_long_timeout
+            )
+
+            while cover_state.state != MTMount.DeployableMotionState.RETRACTED:
+                cover_state = await self.rem.mtmount.evt_mirrorCoversMotionState.next(
+                    flush=False, timeout=self.mirror_covers_timeout
+                )
+                self.log.debug(
+                    f"Cover state: {MTMount.DeployableMotionState(cover_state.state)!r}"
+                )
+                cover_system_state = (
+                    await self.rem.mtmount.evt_mirrorCoversSystemState.aget(
+                        timeout=self.fast_timeout
+                    )
+                )
+                if cover_system_state.state == MTMount.PowerState.FAULT:
+                    raise RuntimeError(
+                        "Open cover failed. Cover system state: "
+                        f"{MTMount.PowerState(cover_system_state.state)!r}"
+                    )
+            self.log.info(
+                f"Cover state: {MTMount.DeployableMotionState(cover_state.state)!r}"
+            )
+        else:
+            raise RuntimeError(
+                f"Mirror covers in {MTMount.DeployableMotionState(cover_state.state)!r} "
+                f"state. Expected {MTMount.DeployableMotionState.RETRACTED!r} or "
+                f"{MTMount.DeployableMotionState.DEPLOYED!r}"
+            )
+
+    async def slew_to_m1_cover_operational_range(self) -> None:
+        """Slew the telescope to safe range for mirror covers operation.
+
+        This method will slew the telescope to a safe elevation to perform
+        mirror covers operations. It should be used in combination with the
+        in_m1_covers_operational_range method.
+
+        """
+        self.log.debug(
+            "Slewing telescope to operational range for mirror covers operation."
+        )
+        azimuth = await self.rem.mtmount.tel_azimuth.aget(timeout=self.fast_timeout)
+        rotation_data = await self.rem.mtrotator.tel_rotation.aget(
+            timeout=self.fast_timeout
+        )
+
+        await self.point_azel(
+            target_name="Mirror covers operation",
+            az=azimuth.actualPosition,
+            el=self.tel_operate_mirror_covers_el,
+            rot_tel=rotation_data.actualPosition,
+            wait_dome=False,
+        )
+
+    async def in_m1_cover_operational_range(self) -> bool:
+        """Check if MTMount is in safe range for mirror covers operation.
+
+        Returns
+        -------
+        elevation_in_range: `bool`
+            Returns `True` when telescope elevation is in safe range for
+            mirror covers operation.
+
+        """
+        elevation = await self.rem.mtmount.tel_elevation.aget(timeout=self.fast_timeout)
+
+        return elevation.actualPosition >= self.tel_operate_mirror_covers_el
 
     async def prepare_for_flatfield(self, check: typing.Any = None) -> None:
         # TODO: Implement (DM-21336).

python/lsst/ts/observatory/control/mock/mtcs_async_mock.py

@@ -98,6 +98,13 @@ async def setup_types(self) -> None:
             inPosition=True,
         )
 
+        self._mtmount_evt_mirror_covers_motion_state = types.SimpleNamespace(
+            state=xml.enums.MTMount.DeployableMotionState.DEPLOYED
+        )
+        self._mtmount_evt_mirror_covers_system_state = types.SimpleNamespace(
+            state=xml.enums.MTMount.PowerState.ON
+        )
+
         # MTRotator data
         self._mtrotator_tel_rotation = types.SimpleNamespace(
             demandPosition=0.0,
@@ -201,6 +208,7 @@ async def setup_mtmount(self) -> None:
             "evt_target.next.side_effect": self.mtmount_evt_target_next,
             "evt_target.aget.side_effect": self.mtmount_evt_target_next,
             "tel_azimuth.next.side_effect": self.mtmount_tel_azimuth_next,
+            "tel_azimuth.aget.side_effect": self.mtmount_tel_azimuth_next,
             "tel_azimuth.DataType.return_value": self.get_sample(
                 "MTMount", "tel_azimuth"
             ),
@@ -211,6 +219,11 @@ async def setup_mtmount(self) -> None:
             "evt_cameraCableWrapFollowing.aget.side_effect": self.mtmount_evt_cameraCableWrapFollowing,
             "cmd_enableCameraCableWrapFollowing.start.side_effect": self.mtmount_cmd_enable_ccw_following,
             "cmd_disableCameraCableWrapFollowing.start.side_effect": self.mtmount_cmd_disable_ccw_following,
+            "evt_mirrorCoversMotionState.aget.side_effect": self.mtmount_evt_mirror_covers_motion_state,
+            "evt_mirrorCoversMotionState.next.side_effect": self.mtmount_evt_mirror_covers_motion_state,
+            "evt_mirrorCoversSystemState.aget.side_effect": self.mtmount_evt_mirror_covers_system_state,
+            "cmd_closeMirrorCovers.start.side_effect": self.mtmount_cmd_close_mirror_covers,
+            "cmd_openMirrorCovers.set_start.side_effect": self.mtmount_cmd_open_mirror_covers,
         }
 
         self.mtcs.rem.mtmount.configure_mock(**mtmount_mocks)
@@ -385,6 +398,7 @@ async def mtmount_tel_azimuth_next(
     async def mtmount_tel_elevation_next(
         self, *args: typing.Any, **kwargs: typing.Any
     ) -> types.SimpleNamespace:
+        await asyncio.sleep(self.heartbeat_time)
         return self._mtmount_tel_elevation
 
     async def mtmount_evt_elevation_in_position_next(
@@ -415,6 +429,46 @@ async def mtmount_cmd_disable_ccw_following(
     async def mtrotator_cmd_move(self, *args: typing.Any, **kwargs: typing.Any) -> None:
         asyncio.create_task(self._mtrotator_move(position=kwargs.get("position", 0.0)))
 
+    async def mtmount_evt_mirror_covers_motion_state(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> types.SimpleNamespace:
+        await asyncio.sleep(self.heartbeat_time)
+        return self._mtmount_evt_mirror_covers_motion_state
+
+    async def mtmount_evt_mirror_covers_system_state(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> types.SimpleNamespace:
+        await asyncio.sleep(self.heartbeat_time)
+        return self._mtmount_evt_mirror_covers_system_state
+
+    async def mtmount_cmd_close_mirror_covers(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> None:
+        asyncio.create_task(self._mtmount_cmd_close_mirror_covers())
+
+    async def _mtmount_cmd_close_mirror_covers(self) -> None:
+        self._mtmount_evt_mirror_covers_motion_state.state = (
+            xml.enums.MTMount.DeployableMotionState.DEPLOYING
+        )
+        await asyncio.sleep(self.heartbeat_time)
+        self._mtmount_evt_mirror_covers_motion_state.state = (
+            xml.enums.MTMount.DeployableMotionState.DEPLOYED
+        )
+
+    async def mtmount_cmd_open_mirror_covers(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> None:
+        asyncio.create_task(self._mtmount_cmd_open_mirror_covers())
+
+    async def _mtmount_cmd_open_mirror_covers(self) -> None:
+        self._mtmount_evt_mirror_covers_motion_state.state = (
+            xml.enums.MTMount.DeployableMotionState.RETRACTING
+        )
+        await asyncio.sleep(self.heartbeat_time)
+        self._mtmount_evt_mirror_covers_motion_state.state = (
+            xml.enums.MTMount.DeployableMotionState.RETRACTED
+        )
+
     async def _mtrotator_move(self, position: float) -> None:
         self._mtrotator_evt_in_position.inPosition = False