Move Eiger internals from apischema to pydantic

pyproject.toml

@@ -15,8 +15,7 @@ dependencies = [
     "tickit==0.2.3",
     "typing_extensions",
     "softioc",
-    "pydantic>1",
-    "apischema"
+    "pydantic>1"
 ] 
 dynamic = ["version"]
 license.file = "LICENSE"

src/tickit_devices/eiger/eiger_adapters.py

@@ -1,14 +1,21 @@
 import logging
+from typing import Any, Dict, List, Union
 
 from aiohttp import web
-from apischema import serialize
+from pydantic.v1 import BaseModel
 from tickit.adapters.httpadapter import HttpAdapter
 from tickit.adapters.interpreters.endpoints.http_endpoint import HttpEndpoint
 from tickit.adapters.zeromq.push_adapter import ZeroMqPushAdapter
 
 from tickit_devices.eiger.eiger import EigerDevice
-from tickit_devices.eiger.eiger_schema import SequenceComplete, Value, construct_value
+from tickit_devices.eiger.eiger_schema import (
+    AccessMode,
+    SequenceComplete,
+    Value,
+    construct_value,
+)
 from tickit_devices.eiger.eiger_status import State
+from tickit_devices.utils import serialize
 
 API_VERSION = "1.8.0"
 DETECTOR_API = f"detector/api/{API_VERSION}"
@@ -41,9 +48,11 @@ async def get_config(self, request: web.Request) -> web.Response:
             data = construct_value(self.device.settings, param)
 
         else:
-            data = serialize(Value("None", "string", access_mode="None"))
+            data = serialize(
+                Value(value="None", value_type="string", access_mode=AccessMode.NONE)
+            )
 
-        return web.json_response(data)
+        return web.json_response(data.dict())
 
     @HttpEndpoint.put(f"/{DETECTOR_API}" + "/config/{parameter_name}")
     async def put_config(self, request: web.Request) -> web.Response:
@@ -59,11 +68,11 @@ async def put_config(self, request: web.Request) -> web.Response:
         """
         param = request.match_info["parameter_name"]
 
-        response = await request.json()
+        response = await request.dict()
 
         if self.device.get_state() is not State.IDLE:
             LOGGER.warning("Eiger not initialized or is currently running.")
-            return web.json_response(serialize([]))
+            return web.json_response([])
         elif (
             hasattr(self.device.settings, param)
             and self.device.get_state() is State.IDLE
@@ -75,10 +84,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]))
+            return web.json_response([param])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
-            return web.json_response(serialize([]))
+            return web.json_response([])
 
     @HttpEndpoint.get(f"/{DETECTOR_API}" + "/status/{status_param}")
     async def get_status(self, request: web.Request) -> web.Response:
@@ -97,9 +106,11 @@ async def get_status(self, request: web.Request) -> web.Response:
             data = construct_value(self.device.status, param)
 
         else:
-            data = serialize(Value("None", "string", access_mode="None"))
+            data = serialize(
+                Value(value="None", value_type="string", access_mode=AccessMode.NONE)
+            )
 
-        return web.json_response(data)
+        return web.json_response(data.dict())
 
     @HttpEndpoint.get(f"/{DETECTOR_API}" + "/status/board_000/{status_param}")
     async def get_board_000_status(self, request: web.Request) -> web.Response:
@@ -141,7 +152,7 @@ async def initialize_eiger(self, request: web.Request) -> web.Response:
         await self.device.initialize()
 
         LOGGER.debug("Initializing Eiger...")
-        return web.json_response(serialize(SequenceComplete(1)))
+        return web.json_response(serialize(SequenceComplete.number(1)))
 
     @HttpEndpoint.put(f"/{DETECTOR_API}" + "/command/arm", interrupt=True)
     async def arm_eiger(self, request: web.Request) -> web.Response:
@@ -157,7 +168,7 @@ async def arm_eiger(self, request: web.Request) -> web.Response:
         await self.device.arm()
 
         LOGGER.debug("Arming Eiger...")
-        return web.json_response(serialize(SequenceComplete(2)))
+        return web.json_response(serialize(SequenceComplete.number(2)))
 
     @HttpEndpoint.put(f"/{DETECTOR_API}" + "/command/disarm", interrupt=True)
     async def disarm_eiger(self, request: web.Request) -> web.Response:
@@ -173,7 +184,7 @@ async def disarm_eiger(self, request: web.Request) -> web.Response:
         await self.device.disarm()
 
         LOGGER.debug("Disarming Eiger...")
-        return web.json_response(serialize(SequenceComplete(3)))
+        return web.json_response(serialize(SequenceComplete.number(3)))
 
     @HttpEndpoint.put(f"/{DETECTOR_API}" + "/command/trigger", interrupt=False)
     async def trigger_eiger(self, request: web.Request) -> web.Response:
@@ -192,7 +203,7 @@ async def trigger_eiger(self, request: web.Request) -> web.Response:
         await self.raise_interrupt()
         await self.device.finished_aquisition.wait()
 
-        return web.json_response(serialize(SequenceComplete(4)))
+        return web.json_response(serialize(SequenceComplete.number(4)))
 
     @HttpEndpoint.put(f"/{DETECTOR_API}" + "/command/cancel", interrupt=True)
     async def cancel_eiger(self, request: web.Request) -> web.Response:
@@ -208,7 +219,7 @@ async def cancel_eiger(self, request: web.Request) -> web.Response:
         await self.device.cancel()
 
         LOGGER.debug("Cancelling Eiger...")
-        return web.json_response(serialize(SequenceComplete(5)))
+        return web.json_response(serialize(SequenceComplete.number(5)))
 
     @HttpEndpoint.put(f"/{DETECTOR_API}" + "/command/abort", interrupt=True)
     async def abort_eiger(self, request: web.Request) -> web.Response:
@@ -224,7 +235,7 @@ async def abort_eiger(self, request: web.Request) -> web.Response:
         await self.device.abort()
 
         LOGGER.debug("Aborting Eiger...")
-        return web.json_response(serialize(SequenceComplete(6)))
+        return web.json_response(serialize(SequenceComplete.number(6)))
 
     @HttpEndpoint.get(f"/{STREAM_API}" + "/status/{param}")
     async def get_stream_status(self, request: web.Request) -> web.Response:
@@ -274,7 +285,7 @@ async def put_stream_config(self, request: web.Request) -> web.Response:
         """
         param = request.match_info["param"]
 
-        response = await request.json()
+        response = await request.dict()
 
         if hasattr(self.device.stream.config, param):
             attr = response["value"]
@@ -284,10 +295,10 @@ async def put_stream_config(self, request: web.Request) -> web.Response:
             self.device.stream.config[param] = attr
 
             LOGGER.debug("Set " + str(param) + " to " + str(attr))
-            return web.json_response(serialize([param]))
+            return web.json_response([param])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
-            return web.json_response(serialize([]))
+            return web.json_response([])
 
     @HttpEndpoint.get(f"/{MONITOR_API}" + "/config/{param}")
     async def get_monitor_config(self, request: web.Request) -> web.Response:
@@ -320,7 +331,7 @@ async def put_monitor_config(self, request: web.Request) -> web.Response:
         """
         param = request.match_info["param"]
 
-        response = await request.json()
+        response = await request.dict()
 
         if hasattr(self.device.monitor_config, param):
             attr = response["value"]
@@ -330,10 +341,10 @@ 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([param])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
-            return web.json_response(serialize([]))
+            return web.json_response([])
 
     @HttpEndpoint.get(f"/{MONITOR_API}" + "/status/{param}")
     async def get_monitor_status(self, request: web.Request) -> web.Response:
@@ -383,7 +394,7 @@ async def put_filewriter_config(self, request: web.Request) -> web.Response:
         """
         param = request.match_info["param"]
 
-        response = await request.json()
+        response = await request.dict()
 
         if hasattr(self.device.filewriter_config, param):
             attr = response["value"]
@@ -393,10 +404,10 @@ 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([param])
         else:
             LOGGER.debug("Eiger has no config variable: " + str(param))
-            return web.json_response(serialize([]))
+            return web.json_response([])
 
     @HttpEndpoint.get(f"/{FILEWRITER_API}" + "/status/{param}")
     async def get_filewriter_status(self, request: web.Request) -> web.Response:

src/tickit_devices/eiger/eiger_schema.py

@@ -1,13 +1,11 @@
 import logging
-from dataclasses import dataclass, field
 from enum import Enum
 from functools import partial
 from typing import Any, Generic, List, Mapping, Optional, TypeVar
 
-from apischema import serialized
-from apischema.fields import with_fields_set
-from apischema.metadata import skip
-from apischema.serialization import serialize
+from pydantic.v1 import BaseModel, Field
+
+from tickit_devices.utils import serialize
 
 T = TypeVar("T")
 
@@ -29,15 +27,16 @@ def field_config(**kwargs) -> Mapping[str, Any]:
     return dict(**kwargs)
 
 
-class AccessMode(Enum):
+class AccessMode(str, Enum):
     """Possible access modes for field metadata."""
 
     READ_ONLY: str = "r"
     WRITE_ONLY: str = "w"
     READ_WRITE: str = "rw"
+    NONE: str = "None"
 
 
-class ValueType(Enum):
+class ValueType(str, Enum):
     """Possible value types for field metadata."""
 
     FLOAT: str = "float"
@@ -105,53 +104,60 @@ class ValueType(Enum):
 )
 
 
-@with_fields_set
-@dataclass
-class Value(Generic[T]):
+class Value(BaseModel, Generic[T]):
     """Schema for a value to be returned by the API. Most fields are optional."""
 
     value: T
     value_type: str
-    access_mode: Optional[str] = None
+    access_mode: Optional[AccessMode] = None
     unit: Optional[str] = None
     min: Optional[T] = None
     max: Optional[T] = None
     allowed_values: Optional[List[str]] = None
 
 
-def construct_value(obj, param):  # noqa: D103
+def construct_value(obj, param) -> Value:  # noqa: D103
     value = obj[param]["value"]
     meta = obj[param]["metadata"]
 
     if "allowed_values" in meta:
-        data = serialize(
-            Value(
-                value,
-                meta["value_type"].value,
-                access_mode=meta["access_mode"].value,
-                allowed_values=meta["allowed_values"],
-            )
+        data = Value(
+            value=value,
+            value_type=meta["value_type"].value,
+            access_mode=meta["access_mode"].value,
+            allowed_values=meta["allowed_values"],
         )
 
     else:
-        data = serialize(
-            Value(
-                value,
-                meta["value_type"].value,
-                access_mode=meta["access_mode"].value,
-            )
+        data = Value(
+            value=value,
+            value_type=meta["value_type"].value,
+            access_mode=meta["access_mode"].value,
         )
 
     return data
 
 
-@dataclass
-class SequenceComplete:
+class SequenceComplete(BaseModel):
     """Schema for confirmation returned by operations that do not return values."""
 
-    _sequence_id: int = field(default=1, metadata=skip, init=True, repr=False)
+    sequence_id: int = Field(default=1, alias="sequence id")
+
+    @classmethod
+    def number(cls, number: int) -> "SequenceComplete":
+        """Create a new completion document with the given ID.
+
+        This function exists as a workaround for mypy ignoring aliases.
+        See https://github.com/pydantic/pydantic/discussions/2889
+
+        Args:
+            number: The sequence ID
+
+        Returns:
+            SequenceComplete: Document describing a completed sequence of operations
+        """
+        return SequenceComplete(sequence_id=number)  # type: ignore
 
-    @serialized("sequence id")  # type: ignore
-    @property
-    def sequence_id(self) -> int:  # noqa: D102
-        return self._sequence_id
+    class Config:
+        allow_population_by_field_name = True
+        fields = {"sequence_id": "sequence id"}

src/tickit_devices/eiger/eiger_settings.py

@@ -112,6 +112,7 @@ class EigerSettings:
     trigger_mode: str = field(
         default="exts", metadata=rw_str(allowed_values=["exts", "ints", "exte", "inte"])
     )
+    trigger_start_delay: float = field(default=0.0, metadata=rw_float())
     two_theta_increment: float = field(default=0.0, metadata=rw_float())
     two_theta_start: float = field(default=0.0, metadata=rw_float())
     wavelength: float = field(default=1.0, metadata=rw_float())

src/tickit_devices/eiger/eiger_status.py

@@ -6,7 +6,7 @@
 from .eiger_schema import ro_str_list, rw_datetime, rw_float, rw_state
 
 
-class State(Enum):
+class State(str, Enum):
     """Possible states of the Eiger detector."""
 
     NA = "na"

src/tickit_devices/eiger/stream/eiger_stream.py

@@ -24,6 +24,9 @@
 
 _Message = Union[BaseModel, Mapping[str, Any], bytes]
 
+_Sendable = Union[bytes, Frame, memoryview]
+_Message = Union[_Sendable, str, Mapping[str, Any], BaseModel]
+
 
 class EigerStream:
     """Simulation of an Eiger stream."""
@@ -34,6 +37,8 @@ class EigerStream:
 
     _message_buffer: Queue[_Message]
 
+    _message_buffer: Queue[_Sendable]
+
     #: An empty typed mapping of input values
     Inputs: TypedDict = TypedDict("Inputs", {})
     #: A typed mapping containing the 'value' output value