Add further API fixes and initial stream2 support

.codecov.yml

@@ -0,0 +1,2 @@
+ignore:
+  - "src/tickit_devices/eiger/stream/stream2.py"

pyproject.toml

@@ -16,7 +16,8 @@ dependencies = [
     "typing_extensions",
     "softioc",
     "pydantic>1",
-    "apischema"
+    "apischema",
+    "cbor2",
 ]
 dynamic = ["version"]
 license.file = "LICENSE"
@@ -68,6 +69,7 @@ version_file = "src/tickit_devices/_version.py"
 
 [tool.mypy]
 ignore_missing_imports = true # Ignore missing stubs in imported modules
+exclude = ["src/tickit_devices/eiger/stream/stream2.py"]
 
 [tool.pytest.ini_options]
 # Run pytest with all our checkers, and don't spam us with massive tracebacks on error
@@ -84,6 +86,7 @@ testpaths = "docs src tests"
 
 [tool.coverage.run]
 data_file = "/tmp/tickit_devices.coverage"
+omit = ["src/tickit_devices/eiger/stream/stream2.py"]
 
 [tool.coverage.paths]
 # Tests are run from installed location, map back to the src directory

src/tickit_devices/eiger/__init__.py

@@ -8,6 +8,7 @@
 
 from tickit_devices.eiger.eiger import EigerDevice
 from tickit_devices.eiger.eiger_adapters import EigerRESTAdapter, EigerZMQAdapter
+from tickit_devices.eiger.stream.stream_config import CBOR_STREAM, LEGACY_STREAM
 
 
 @pydantic.v1.dataclasses.dataclass
@@ -16,8 +17,9 @@ class Eiger(ComponentConfig):
 
     host: str = "0.0.0.0"
     port: int = 8081
-    zmq_host: str = "127.0.0.1"
-    zmq_port: int = 9999
+    stream_host: str = "127.0.0.1"
+    stream_legacy_port: int = 9999
+    stream_cbor_port: int = 31001
 
     def __call__(self) -> Component:  # noqa: D102
         logging.getLogger("aiohttp.access").setLevel(logging.WARNING)
@@ -31,10 +33,17 @@ def __call__(self) -> Component:  # noqa: D102
                 ),
             ),
             AdapterContainer(
-                EigerZMQAdapter(device),
+                EigerZMQAdapter(device.streams[LEGACY_STREAM]),
                 ZeroMqPushIo(
-                    self.zmq_host,
-                    self.zmq_port,
+                    self.stream_host,
+                    self.stream_legacy_port,
+                ),
+            ),
+            AdapterContainer(
+                EigerZMQAdapter(device.streams[CBOR_STREAM]),
+                ZeroMqPushIo(
+                    self.stream_host,
+                    self.stream_cbor_port,
                 ),
             ),
         ]

src/tickit_devices/eiger/data/stream2/end.cbor

@@ -0,0 +1 @@
+ÙÙ÷£dtypecendiseries_id<þpseries_unique_idx01HBV3JPF9T4ZDPADX6EMK6XMZ

src/tickit_devices/eiger/eiger.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+from collections.abc import Mapping
 from queue import Queue
 
 from tickit.core.device import Device, DeviceUpdate
@@ -13,6 +14,13 @@
 from tickit_devices.eiger.monitor.monitor_config import MonitorConfig
 from tickit_devices.eiger.monitor.monitor_status import MonitorStatus
 from tickit_devices.eiger.stream.eiger_stream import EigerStream
+from tickit_devices.eiger.stream.eiger_stream_2 import EigerStream2
+from tickit_devices.eiger.stream.stream_config import (
+    CBOR_STREAM,
+    LEGACY_STREAM,
+    StreamConfig,
+)
+from tickit_devices.eiger.stream.stream_status import StreamStatus
 
 from .eiger_status import EigerStatus, State
 
@@ -35,7 +43,8 @@ class EigerDevice(Device):
 
     settings: EigerSettings
     status: EigerStatus
-    stream: EigerStream
+    stream: EigerStream | EigerStream2
+    streams: Mapping[str, EigerStream | EigerStream2]
 
     _num_frames_left: int
     _data_queue: Queue
@@ -49,7 +58,7 @@ def __init__(
         self,
         settings: EigerSettings | None = None,
         status: EigerStatus | None = None,
-        stream: EigerStream | None = None,
+        stream: EigerStream | EigerStream2 | None = None,
     ) -> None:
         """Construct a new eiger.
 
@@ -61,7 +70,13 @@ def __init__(
         self.settings = settings or EigerSettings()
         self.status = status or EigerStatus()
 
-        self.stream = stream or EigerStream(callback_period=SimTime(int(1e9)))
+        self.stream_status: StreamStatus = StreamStatus()
+        self.stream_config: StreamConfig = StreamConfig()
+        self.streams = {
+            LEGACY_STREAM: stream or EigerStream(callback_period=SimTime(int(1e9))),
+            CBOR_STREAM: stream or EigerStream2(callback_period=SimTime(int(1e9))),
+        }
+        self.stream = self.streams[CBOR_STREAM]
 
         self.filewriter_status: FileWriterStatus = FileWriterStatus()
         self.filewriter_config: FileWriterConfig = FileWriterConfig()
@@ -102,8 +117,11 @@ async def arm(self) -> None:
 
         Required for triggering.
         """
+        self.stream = self.streams[self.stream_config.format]
         self._series_id += 1
-        self.stream.begin_series(self.settings, self._series_id)
+        self.stream.begin_series(
+            self.settings, self._series_id, self.stream_config.header_detail
+        )
         self._num_frames_left = self.settings.nimages
         self._num_triggers_left = self.settings.ntrigger
         self._set_state(State.READY)
@@ -224,3 +242,73 @@ def _set_state(self, state: State) -> None:
 
     def _is_in_state(self, state: State) -> bool:
         return self.get_state() is state
+
+
+def get_changed_parameters(key: str) -> list[str]:
+    """Get the list of parameters that may have changed as a result of putting
+    to the parameter provided.
+
+    Args:
+        key: string key of the changed parameter within the detector subsystem
+
+    Returns:
+        list[str]: a list of keys which may have been changed after a PUT request
+    """
+    match key:
+        case "auto_summation":
+            return ["auto_summation", "frame_count_time"]
+        case "count_time" | "frame_time":
+            return [
+                "bit_depth_image",
+                "bit_depth_readout",
+                "count_time",
+                "countrate_correction_count_cutoff",
+                "frame_count_time",
+                "frame_time",
+            ]
+        case "flatfield":
+            return ["flatfield", "threshold/1/flatfield"]
+        case "incident_energy" | "photon_energy":
+            return [
+                "element",
+                "flatfield",
+                "incident_energy",
+                "photon_energy",
+                "threshold/1/energy",
+                "threshold/1/flatfield",
+                "threshold/2/energy",
+                "threshold/2/flatfield",
+                "threshold_energy",
+                "wavelength",
+            ]
+        case "pixel_mask":
+            return ["pixel_mask", "threshold/1/pixel_mask"]
+        case "threshold/1/flatfield":
+            return ["flatfield", "threshold/1/flatfield"]
+        case "roi_mode":
+            return ["count_time", "frame_time", "roi_mode"]
+        case "threshold_energy" | "threshold/1/energy":
+            return [
+                "flatfield",
+                "threshold/1/energy",
+                "threshold/1/flatfield",
+                "threshold/2/flatfield",
+                "threshold_energy",
+            ]
+        case "threshold/2/energy":
+            return [
+                "flatfield",
+                "threshold/1/flatfield",
+                "threshold/2/energy",
+                "threshold/2/flatfield",
+            ]
+        case "threshold/1/mode":
+            return ["threshold/1/mode", "threshold/difference/mode"]
+        case "threshold/2/mode":
+            return ["threshold/2/mode", "threshold/difference/mode"]
+        case "threshold/1/pixel_mask":
+            return ["pixel_mask", "threshold/1/pixel_mask"]
+        case "threshold/difference/mode":
+            return ["difference_mode"]  # replicating API inconsistency
+        case _:
+            return [key]

src/tickit_devices/eiger/eiger_adapters.py

@@ -6,8 +6,10 @@
 from tickit.adapters.specifications import HttpEndpoint
 from tickit.adapters.zmq import ZeroMqPushAdapter
 
-from tickit_devices.eiger.eiger import EigerDevice
+from tickit_devices.eiger.eiger import EigerDevice, get_changed_parameters
 from tickit_devices.eiger.eiger_schema import SequenceComplete, construct_value
+from tickit_devices.eiger.stream.eiger_stream import EigerStream
+from tickit_devices.eiger.stream.eiger_stream_2 import EigerStream2
 
 API_VERSION = "1.8.0"
 DETECTOR_API = f"detector/api/{API_VERSION}"
@@ -73,7 +75,10 @@ async def put_config(self, request: web.Request) -> web.Response:
             self.device.settings[param] = attr
 
             LOGGER.debug("Set " + str(param) + " to " + str(attr))
-            return web.json_response(serialize([param]))
+
+            changed_parameters = get_changed_parameters(param)
+
+            return web.json_response(serialize(changed_parameters))
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
             return web.json_response(status=404)
@@ -121,15 +126,14 @@ async def put_threshold_config(self, request: web.Request) -> web.Response:
         config = self.device.settings.threshold_config
         if threshold in config and hasattr(config[threshold], param):
             attr = response["value"]
-
-            LOGGER.debug(
-                f"Changing to {str(attr)} for threshold/{threshold}{str(param)}"
-            )
-
             config[threshold][param] = attr
 
             LOGGER.debug(f"Set threshold/{threshold}{str(param)} to {str(attr)}")
-            return web.json_response(serialize([param]))
+
+            full_param = f"threshold/{threshold}/{param}"
+            changed_parameters = get_changed_parameters(full_param)
+
+            return web.json_response(serialize(changed_parameters))
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
             return web.json_response(status=404)
@@ -319,8 +323,8 @@ async def get_stream_status(self, request: web.Request) -> web.Response:
         """
         param = request.match_info["param"]
 
-        if hasattr(self.device.stream.status, param):
-            return web.json_response(construct_value(self.device.stream.status, param))
+        if hasattr(self.device.stream_status, param):
+            return web.json_response(construct_value(self.device.stream_status, param))
         else:
             return web.json_response(status=404)
 
@@ -337,8 +341,8 @@ async def get_stream_config(self, request: web.Request) -> web.Response:
         """
         param = request.match_info["param"]
 
-        if hasattr(self.device.stream.config, param):
-            return web.json_response(construct_value(self.device.stream.config, param))
+        if hasattr(self.device.stream_config, param):
+            return web.json_response(construct_value(self.device.stream_config, param))
         else:
             return web.json_response(status=404)
 
@@ -358,15 +362,15 @@ async def put_stream_config(self, request: web.Request) -> web.Response:
 
         response = await request.json()
 
-        if hasattr(self.device.stream.config, param):
+        if hasattr(self.device.stream_config, param):
             attr = response["value"]
 
             LOGGER.debug(f"Changing to {attr} for {param}")
 
-            self.device.stream.config[param] = attr
+            self.device.stream_config[param] = attr
 
             LOGGER.debug("Set " + str(param) + " to " + str(attr))
-            return web.json_response(serialize([param]))
+            return web.json_response([])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
             return web.json_response(status=404)
@@ -413,7 +417,7 @@ async def put_monitor_config(self, request: web.Request) -> web.Response:
             self.device.monitor_config[param] = attr
 
             LOGGER.debug("Set " + str(param) + " to " + str(attr))
-            return web.json_response(serialize([param]))
+            return web.json_response([])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
             return web.json_response(status=404)
@@ -480,7 +484,7 @@ async def put_filewriter_config(self, request: web.Request) -> web.Response:
             self.device.filewriter_config[param] = attr
 
             LOGGER.debug("Set " + str(param) + " to " + str(attr))
-            return web.json_response(serialize([param]))
+            return web.json_response([])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
             return web.json_response(status=404)
@@ -511,11 +515,11 @@ class EigerZMQAdapter(ZeroMqPushAdapter):
 
     device: EigerDevice
 
-    def __init__(self, device: EigerDevice) -> None:
+    def __init__(self, stream: EigerStream | EigerStream2) -> None:
         super().__init__()
-        self.device = device
+        self.stream = stream
 
     def after_update(self) -> None:
         """Updates IOC values immediately following a device update."""
-        if buffered_data := list(self.device.stream.consume_data()):
+        if buffered_data := list(self.stream.consume_data()):
             self.add_message_to_stream(buffered_data)

src/tickit_devices/eiger/eiger_schema.py

@@ -33,26 +33,26 @@ def field_config(**kwargs) -> Mapping[str, Any]:
 class AccessMode(Enum):
     """Possible access modes for field metadata."""
 
-    READ_ONLY: str = "r"
-    WRITE_ONLY: str = "w"
-    READ_WRITE: str = "rw"
+    READ_ONLY = "r"
+    WRITE_ONLY = "w"
+    READ_WRITE = "rw"
 
 
 class ValueType(Enum):
     """Possible value types for field metadata."""
 
-    FLOAT: str = "float"
-    INT: str = "int"
-    UINT: str = "uint"
-    STRING: str = "string"
-    STR_LIST: str = "string[]"
-    BOOL: str = "bool"
-    FLOAT_GRID: str = "float[][]"
-    UINT_GRID: str = "uint[][]"
-    DATE: str = "date"
-    DATETIME: str = "datetime"
-    NONE: str = "none"
-    STATE: str = "State"
+    FLOAT = "float"
+    INT = "int"
+    UINT = "uint"
+    STRING = "string"
+    STR_LIST = "string[]"
+    BOOL = "bool"
+    FLOAT_GRID = "float[][]"
+    UINT_GRID = "uint[][]"
+    DATE = "date"
+    DATETIME = "datetime"
+    NONE = "none"
+    STATE = "State"
 
 
 #