make borzoi use start/stop time instead of runstate

Author: rerpha

README.md

@@ -1,3 +1,3 @@
 # Borzoi
 
-Borzoi like running. This listens to ISISDAE's run state and pushes start/stop Flatbuffer blobs to Kafka accordingly.
+Borzoi like running. This listens to ISISDAE's start/stop times and pushes start/stop Flatbuffer blobs to Kafka accordingly.

main.py

@@ -4,7 +4,7 @@
 import os
 import uuid
 
-from aioca import caget, camonitor
+from aioca import camonitor
 from aiokafka import AIOKafkaProducer
 from epicscorelibs.ca.dbr import DBR_CHAR_BYTES, ca_bytes
 from ibex_non_ca_helpers.compress_hex import dehex_decompress_and_dejson
@@ -27,64 +27,68 @@ def __init__(
         self.instrument_name = instrument_name
         self.topic = topic
         self.current_run_number = None
-        self.current_start_time_ms = None
+        self.current_start_time = None
+        self.current_stop_time = None
 
     async def set_up_monitors(self) -> None:
         logger.info("Setting up monitors")
         camonitor(
             f"{self.prefix}CS:BLOCKSERVER:BLOCKNAMES",
             callback=self._update_blocks,
+            all_updates=True,
             datatype=DBR_CHAR_BYTES,
         )
         camonitor(
-            f"{self.prefix}DAE:RUNSTATE",
-            callback=self._react_to_runstate_change,
+            f"{self.prefix}DAE:RUNNUMBER",
+            callback=self._update_run_number,
             all_updates=True,
             datatype=str,
         )
         camonitor(
-            f"{self.prefix}DAE:RUNNUMBER",
-            callback=self._update_run_number,
+            f"{self.prefix}DAE:START_TIME",
+            callback=self.construct_and_send_runstart,
             all_updates=True,
-            datatype=int,
+            datatype=float,
         )
         camonitor(
-            f"{self.prefix}DAE:START_TIME",
-            callback=self._update_start_time_ms,
+            f"{self.prefix}DAE:STOP_TIME",
+            self.construct_and_send_runstop,
             all_updates=True,
-            datatype=int,
+            datatype=float,
         )
 
     def _update_run_number(self, value: int) -> None:
         # Cache this as we want the run start message construction and production to be as fast as
         # possible so we don't miss events
-        logger.debug(f"Run number updated to {value}")
+        logger.info(f"Run number updated to {value}")
         self.current_run_number = value
 
     def _update_start_time_ms(self, value: int) -> None:
         # Cache this as we want the run start message construction and production to be as fast as
         # possible so we don't miss events
-        logger.debug(f"Run start time updated to {value} so changing it to ms ({value * 1000})")
-        self.current_start_time_ms = value * 1000
+        logger.info(f"Run start time updated to {value} so changing it to ms ({value * 1000})")
+        self.current_start_time_ms = int(value) * 1000
 
     def _update_blocks(self, value: ca_bytes) -> None:
         logger.debug(f"blocks_hexed: {value}")
         blocks_unhexed = dehex_decompress_and_dejson(bytes(value))
         logger.debug(f"blocks_unhexed: {blocks_unhexed}")
         self.blocks = [f"{self.prefix}CS:SB:{x}" for x in blocks_unhexed]
 
-    async def _react_to_runstate_change(self, value: str) -> None:
-        logger.info(f"Runstate changed to {value}")
+    async def construct_and_send_runstart(self, value: float) -> None:
+        if self.current_start_time is None:
+            logger.info("Initial update for start time - not sending run start")
+            self.current_start_time = value
+            return
 
-        if value == "BEGINNING":
-            self.current_job_id = str(uuid.uuid4())
-            await self.construct_and_send_runstart(self.current_job_id)
-        elif value == "ENDING":
-            await self.construct_and_send_runstop(self.current_job_id)
+        if value == self.current_start_time or value is None:
+            logger.error("run start time is the same as cached or invalid. ignoring update")
+            return
 
-    async def construct_and_send_runstart(self, job_id: str) -> None:
-        logger.info(f"Sending run start with job_id: {job_id}")
-        start_time_ms = self.current_start_time_ms
+        self.current_start_time = value
+        self.current_job_id = str(uuid.uuid4())
+        logger.info(f"Sending run start with job_id: {self.current_job_id}")
+        start_time_ms = int(self.current_start_time) * 1000
         logger.info(f"Start time: {start_time_ms}")
 
         runnum = self.current_run_number
@@ -133,31 +137,40 @@ async def construct_and_send_runstart(self, job_id: str) -> None:
         filename = f"{self.instrument_name}{runnum}.nxs"
 
         blob = serialise_pl72(
-            job_id,
+            self.current_job_id,
             filename=filename,
             start_time=start_time_ms,
             nexus_structure=json.dumps(nexus_structure),
             run_name=runnum,
+            instrument_name=self.instrument_name,
         )
         await self.producer.send(self.topic, blob)
+        logger.info(f"Sent {blob} blob")
+
+    async def construct_and_send_runstop(self, value: float) -> None:
+        if self.current_stop_time is None:
+            self.current_stop_time = value
+            logger.info("Initial update for stop time - not sending run stop")
+            return
+
+        if value == self.current_stop_time or value is None:
+            logger.error("run stop time is the same as cached or invalid")
+            return
 
-    async def construct_and_send_runstop(self, job_id: str) -> None:
-        logger.info(f"Sending run stop with job_id: {job_id}")
-        # stop_time only gets set to a non-zero value when the runstate goes back to SETUP.
-        # This is dirty, but poll it every 0.5 seconds until it does.
-        while (
-            current_runstate := await caget(f"{self.prefix}DAE:RUNSTATE", datatype=str) != "SETUP"
-        ):
-            logger.debug(f"Waiting for run state to go back to SETUP. Currently {current_runstate}")
-            await asyncio.sleep(0.5)
-
-        stop_time_s = await caget(f"{self.prefix}DAE:STOP_TIME")
-        if stop_time_s is None:
-            logger.error(f"Failed to get stop time from {job_id}")
+        if value == 0:
+            # This happens when a new run starts
+            logger.debug("stop time set to 0")
+            self.current_stop_time = value
             return
-        stop_time_ms = int(stop_time_s * 1000)
+
+        self.current_stop_time = value
+        logger.info(f"Sending run stop with job_id: {self.current_job_id}")
+
+        stop_time_ms = int(value * 1000)
         logger.info(f"stop time: {stop_time_ms}")
-        blob = serialise_6s4t(job_id, stop_time=stop_time_ms, command_id=self.current_job_id)
+        blob = serialise_6s4t(
+            self.current_job_id, stop_time=stop_time_ms, command_id=self.current_job_id
+        )
         await self.producer.send(self.topic, blob)