fix ibex-non-ca-helper import

Author: rerpha

main.py

@@ -1,14 +1,13 @@
-import os
-import uuid
-import logging
 import asyncio
 import json
+import logging
+import os
+import uuid
 
 from aioca import caget, camonitor
 from aiokafka import AIOKafkaProducer
-from epicscorelibs.ca.dbr import DBR_CHAR_BYTES
-from ibex_non_ca_helpers.hex import dehex_decompress_and_dejson
-
+from epicscorelibs.ca.dbr import DBR_CHAR_BYTES, ca_bytes
+from ibex_non_ca_helpers.compress_hex import dehex_decompress_and_dejson
 from streaming_data_types.run_start_pl72 import serialise_pl72
 from streaming_data_types.run_stop_6s4t import serialise_6s4t
 
@@ -19,7 +18,7 @@
 class RunStarter:
     def __init__(
         self, prefix: str, instrument_name: str, producer: AIOKafkaProducer, topic: str
-    ):
+    ) -> None:
         self.producer = None
         self.prefix = prefix
         self.blocks = []
@@ -28,7 +27,7 @@ def __init__(
         self.instrument_name = instrument_name
         self.topic = topic
 
-    async def set_up_monitors(self):
+    async def set_up_monitors(self) -> None:
         logger.info("Setting up monitors")
         camonitor(
             f"{self.prefix}CS:BLOCKSERVER:BLOCKNAMES",
@@ -42,13 +41,13 @@ async def set_up_monitors(self):
             datatype=str,
         )
 
-    def _update_blocks(self, value):
+    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):
+    async def _react_to_runstate_change(self, value: str) -> None:
         logger.info(f"Runstate changed to {value}")
 
         if value == "BEGINNING":
@@ -57,7 +56,7 @@ async def _react_to_runstate_change(self, value):
         elif value == "ENDING":
             await self.construct_and_send_runstop(self.current_job_id)
 
-    async def construct_and_send_runstart(self, job_id: str):
+    async def construct_and_send_runstart(self, job_id: str) -> None:
         logger.info(f"Sending run start with job_id: {job_id}")
         start_time_s = await caget(f"{self.prefix}DAE:START_TIME")
         start_time_ms = int(start_time_s * 1000)
@@ -116,25 +115,20 @@ async def construct_and_send_runstart(self, job_id: str):
         )
         await self.producer.send(self.topic, blob)
 
-    async def construct_and_send_runstop(self, job_id: str):
+    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"
+            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}"
-            )
+            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")
         stop_time_ms = int(stop_time_s * 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(job_id, stop_time=stop_time_ms, command_id=self.current_job_id)
         await self.producer.send(self.topic, blob)
 
 
@@ -144,14 +138,12 @@ async def set_up_producer(broker: str) -> AIOKafkaProducer:
     return producer
 
 
-def main():
+def main() -> None:
     prefix = os.environ.get("MYPVPREFIX")
     instrument_name = os.environ.get("INSTRUMENT")
 
     if prefix is None or instrument_name is None:
-        raise ValueError(
-            "prefix or instrument name not set - have you run config_env.bat?"
-        )
+        raise ValueError("prefix or instrument name not set - have you run config_env.bat?")
 
     broker = os.environ.get("BORZOI_KAFKA_BROKER", "livedata.isis.cclrc.ac.uk:31092")
     topic = os.environ.get("BORZOI_TOPIC", f"{instrument_name}_runInfo")

tests/test_main.py

@@ -1,3 +1,2 @@
-
 def test_that_does_nothing():
     pass