DM-47667: Improve BaseCamera start/end take data operation to improve efficiency.

doc/news/DM-47552.feature.rst

@@ -0,0 +1 @@
+Refactor take image operation to allow returning from a take image sequence as soon as the exposure finished, instead of having to wait for the endReadout event.

python/lsst/ts/observatory/control/auxtel/latiss.py

@@ -326,7 +326,11 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                 components_attr=self.components_attr,
                 readonly=False,
                 generics=["summaryState"],
-                atcamera=["takeImages", "endReadout"],
+                atcamera=[
+                    "takeImages",
+                    "endReadout",
+                    "startIntegration",
+                ],
                 atspectrograph=[
                     "changeFilter",
                     "changeDisperser",
@@ -341,7 +345,11 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                 components_attr=["atcamera"],
                 readonly=False,
                 generics=["summaryState"],
-                atcamera=["takeImages", "endReadout"],
+                atcamera=[
+                    "takeImages",
+                    "endReadout",
+                    "startIntegration",
+                ],
             )
             usages[self.valid_use_cases.Setup] = UsagesResources(
                 components_attr=["atspectrograph"],
@@ -360,7 +368,11 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                 components_attr=["atcamera", "atspectrograph", "atheaderservice"],
                 readonly=False,
                 generics=["summaryState"],
-                atcamera=["takeImages", "endReadout"],
+                atcamera=[
+                    "takeImages",
+                    "endReadout",
+                    "startIntegration",
+                ],
                 atheaderservice=["largeFileObjectAvailable"],
                 atspectrograph=[
                     "changeFilter",

python/lsst/ts/observatory/control/base_camera.py

@@ -1213,7 +1213,7 @@ async def _handle_snaps(self, camera_exposure: CameraExposure) -> typing.List[in
             float(camera_exposure.exp_time) + self.read_out_time
         ) * camera_exposure.n + self.long_long_timeout
 
-        self.camera.evt_endReadout.flush()
+        await self.wait_for_camera_readiness()
         await self.camera.cmd_takeImages.start(timeout=take_images_timeout)
 
         exp_ids: typing.List[int] = []
@@ -1231,6 +1231,43 @@ async def _handle_snaps(self, camera_exposure: CameraExposure) -> typing.List[in
 
         return exp_ids
 
+    async def wait_for_camera_readiness(self) -> None:
+        """Wait until the camera is ready to take data."""
+        try:
+            self.camera.evt_endReadout.flush()
+            last_start_integration, last_end_readout = await asyncio.gather(
+                self.camera.evt_startIntegration.aget(timeout=self.long_timeout),
+                self.camera.evt_endReadout.aget(timeout=self.long_timeout),
+            )
+        except asyncio.TimeoutError:
+            self.log.info(
+                "Timeout getting last camera start integration and/or end readout events. "
+                "This usually means no data was taken with the camera yet, "
+                "or there is loss of historical data. Assuming camera is "
+                "ready to take data."
+            )
+            return
+
+        while last_end_readout.imageName != last_start_integration.imageName:
+            self.log.info(
+                f"Last integration started: {last_start_integration.imageName}, "
+                f"last integration completed: {last_end_readout.imageName}."
+            )
+            next_end_readout_timeout = (
+                self.long_timeout + last_start_integration.exposureTime
+            )
+            try:
+                last_end_readout = await self.camera.evt_endReadout.next(
+                    flush=False, timeout=next_end_readout_timeout
+                )
+            except asyncio.TimeoutError:
+                raise RuntimeError(
+                    f"No new end readout event in {next_end_readout_timeout}s. "
+                    f"Cannot determine if camera is ready to take data."
+                )
+
+        self.camera.evt_startIntegration.flush()
+
     async def _handle_take_stuttered(
         self, camera_exposure: CameraExposure
     ) -> typing.List[int]:
@@ -1328,12 +1365,12 @@ async def next_exposure_id(self) -> int:
         int
             Exposure id from next endReadout event.
         """
-        end_readout = await self.camera.evt_endReadout.next(
+        start_integration = await self.camera.evt_startIntegration.next(
             flush=False, timeout=self.long_long_timeout
         )
         # parse out visitID from filename
         # (Patrick comment) this is highly annoying
-        _, _, yyyymmdd, seq_num = end_readout.imageName.split("_")
+        _, _, yyyymmdd, seq_num = start_integration.imageName.split("_")
 
         return int((yyyymmdd + seq_num[1:]))
 

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

@@ -321,6 +321,7 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                     "takeImages",
                     "setFilter",
                     "endReadout",
+                    "startIntegration",
                     "endSetFilter",
                     "availableFilters",
                 ],
@@ -334,6 +335,7 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                     "takeImages",
                     "setFilter",
                     "endReadout",
+                    "startIntegration",
                     "endSetFilter",
                     "availableFilters",
                 ],
@@ -350,6 +352,7 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                     "setFilter",
                     "endReadout",
                     "endSetFilter",
+                    "startIntegration",
                     "availableFilters",
                 ],
                 ccheaderservice=["largeFileObjectAvailable"],

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

@@ -317,6 +317,7 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                     "setFilter",
                     "endReadout",
                     "endSetFilter",
+                    "startIntegration",
                     "availableFilters",
                 ],
                 mtheaderservice=["largeFileObjectAvailable"],
@@ -330,6 +331,7 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                     "setFilter",
                     "endReadout",
                     "endSetFilter",
+                    "startIntegration",
                     "availableFilters",
                 ],
             )
@@ -342,6 +344,7 @@ def usages(self) -> typing.Dict[int, UsagesResources]:
                     "setFilter",
                     "endReadout",
                     "endSetFilter",
+                    "startIntegration",
                     "availableFilters",
                 ],
                 mtheaderservice=["largeFileObjectAvailable"],

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

@@ -87,17 +87,132 @@ class children. It contains some additions to RemoteGroupAsyncMock
     """
 
     async def setup_mocks(self) -> None:
+        self._end_readout_event = asyncio.Event()
+        self._end_readout_event.clear()
+        self._start_integration_event = asyncio.Event()
+        self._start_integration_event.clear()
+
+        self.image_name: str | None = None
+
         self.remote_group.camera.evt_endReadout.next.configure_mock(
             side_effect=self.next_end_readout
         )
+        self.remote_group.camera.evt_endReadout.aget.configure_mock(
+            side_effect=self.aget_end_readout
+        )
+        self.remote_group.camera.evt_startIntegration.aget.configure_mock(
+            side_effect=self.aget_start_integration
+        )
+        self.remote_group.camera.evt_startIntegration.next.configure_mock(
+            side_effect=self.next_start_integration
+        )
+        self.remote_group.camera.cmd_takeImages.start.configure_mock(
+            side_effect=self.start_take_images
+        )
+        self.remote_group.camera.cmd_takeImages.set.configure_mock(
+            side_effect=self.set_take_images
+        )
+        try:
+            self.remote_group.camera.cmd_startImage.set.configure_mock(
+                side_effect=self.set_start_image
+            )
+            self.remote_group.camera.cmd_startImage.start.configure_mock(
+                side_effect=self.start_start_image
+            )
+            self.remote_group.camera.cmd_endImage.start.configure_mock(
+                side_effect=self.start_end_image
+            )
+        except AttributeError:
+            self.log.warning("Camera does not support stuttered images.")
+
+    async def aget_end_readout(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> types.SimpleNamespace:
+        if self.image_name is None:
+            raise asyncio.TimeoutError("No image taken.")
+        else:
+            return self.end_readout
 
     async def next_end_readout(
         self, *args: typing.Any, **kwargs: typing.Any
     ) -> types.SimpleNamespace:
-        date_id = astropy.time.Time.now().tai.isot.split("T")[0].replace("-", "")
-        self.end_readout.imageName = f"test_genericcamera_{date_id}_{next(index_gen)}"
+        self.log.info("Waiting for next end readout start.")
+        await asyncio.wait_for(
+            self._end_readout_event.wait(), timeout=kwargs["timeout"]
+        )
+        self.end_readout.imageName = self.image_name
+        self.log.info("Waiting for next end readout end.")
         return self.end_readout
 
+    async def aget_start_integration(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> types.SimpleNamespace:
+        if self.image_name is None:
+            raise asyncio.TimeoutError("No image taken.")
+        else:
+            return self.start_integration
+
+    async def next_start_integration(
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> types.SimpleNamespace:
+        self.log.debug("Next start integration start.")
+        await asyncio.wait_for(
+            self._start_integration_event.wait(), timeout=kwargs["timeout"]
+        )
+        self.log.debug("Next start integration end.")
+        return self.start_integration
+
+    async def start_take_images(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self.log.debug("Take images start.")
+        for snap in range(self.start_integration.numImages - 1):
+            self._end_readout_event.clear()
+            self.set_next_image_name()
+            self.start_integration.imageName = self.image_name
+            self._start_integration_event.set()
+            await self._mock_take_image()
+
+        self._end_readout_event.clear()
+        self.set_next_image_name()
+        self.start_integration.imageName = self.image_name
+        self._start_integration_event.set()
+        asyncio.create_task(self._mock_take_image())
+        self.log.debug("Take images end.")
+
+    def set_take_images(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self.log.debug("Set take images start.")
+        self.end_readout.exposureTime = kwargs["expTime"]
+        self.end_readout.numImages = kwargs["numImages"]
+        self.start_integration.exposureTime = kwargs["expTime"]
+        self.start_integration.numImages = kwargs["numImages"]
+        self.log.debug("Set take images end.")
+
+    def set_start_image(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self.log.debug("Set start image start.")
+        self.start_integration.exposureTime = kwargs["timeout"]
+        self.log.debug("Set start image end.")
+
+    async def start_start_image(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+
+        self._end_readout_event.clear()
+        self.set_next_image_name()
+        self.start_integration.imageName = self.image_name
+        self._start_integration_event.set()
+
+    async def start_end_image(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        self._end_readout_event.set()
+
+    def set_next_image_name(self) -> None:
+
+        date_id = astropy.time.Time.now().tai.isot.split("T")[0].replace("-", "")
+        self.image_name = f"test_genericcamera_{date_id}_{next(index_gen)}"
+
+    async def _mock_take_image(self) -> None:
+        self.log.debug("Mock take image start.")
+        await asyncio.sleep(0.5)
+        self._end_readout_event.set()
+        self._start_integration_event.clear()
+        self.log.debug("Mock take image end.")
+
     async def assert_take_bias(
         self,
         nbias: int,

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

@@ -116,8 +116,13 @@ async def cmd_take_images_callback(
             one_exp_time = data.expTime
             if data.shutter:
                 one_exp_time += self.shutter_time
+            date_id = astropy.time.Time.now().tai.isot.split("T")[0].replace("-", "")
+            image_name = f"test_latiss_{date_id}_{next(index_gen)}"
+            await self.atcam.evt_startIntegration.set_write(imageName=image_name)
             await asyncio.sleep(one_exp_time)
-            self.end_readout_task = asyncio.create_task(self.end_readout(data))
+            self.end_readout_task = asyncio.create_task(
+                self.end_readout(data, image_name=image_name)
+            )
             if i < data.numImages - 1:
                 await self.end_readout_task
 
@@ -136,7 +141,10 @@ async def cmd_start_image_callback(
         await asyncio.sleep(self.short_time)
         self.atcam_image_started = True
         self.atcam_start_image_time = utils.current_tai()
-        self.end_readout_coro = self.end_readout(data)
+        date_id = astropy.time.Time.now().tai.isot.split("T")[0].replace("-", "")
+        image_name = f"test_latiss_{date_id}_{next(index_gen)}"
+        await self.atcam.evt_startIntegration.set_write(imageName=image_name)
+        self.end_readout_coro = self.end_readout(data, image_name=image_name)
 
     async def cmd_end_image_callback(self, data: salobj.type_hints.BaseMsgType) -> None:
         await asyncio.sleep(self.short_time)
@@ -154,7 +162,9 @@ async def cmd_discard_rows_callback(
     ) -> None:
         await asyncio.sleep(self.short_time)
 
-    async def end_readout(self, data: salobj.type_hints.BaseMsgType) -> None:
+    async def end_readout(
+        self, data: salobj.type_hints.BaseMsgType, image_name: str
+    ) -> None:
         """Wait `self.readout_time` and send endReadout event."""
         self.log.debug(f"end_readout started: sleep {self.readout_time}")
         await asyncio.sleep(self.readout_time)
@@ -166,8 +176,6 @@ async def end_readout(self, data: salobj.type_hints.BaseMsgType) -> None:
         else:
             self.exptime_list.append(utils.current_tai() - self.atcam_image_started)
 
-        date_id = astropy.time.Time.now().tai.isot.split("T")[0].replace("-", "")
-        image_name = f"test_latiss_{date_id}_{next(index_gen)}"
         self.log.debug(f"sending endReadout: {image_name} :: {data}")
 
         additional_keys, additional_values = list(