diff --git a/docs/conf.py b/docs/conf.py index 3ee2ad966..99b82e5cd 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -89,6 +89,9 @@ ("py:class", "p4p.nt.enum.NTEnum"), ("py:class", "p4p.nt.ndarray.NTNDArray"), ("py:class", "p4p.nt.NTTable"), + # httpx and fastapi don't have intersphinx mappings + ("py:class", "httpx.AsyncBaseTransport"), + ("py:class", "fastapi.applications.FastAPI"), # Problems in FastCS itself ("py:class", "BaseController"), ("py:class", "AttrIOUpdateCallback"), diff --git a/src/fastcs/demo/eiger.py b/src/fastcs/demo/eiger.py new file mode 100644 index 000000000..7ebea02fe --- /dev/null +++ b/src/fastcs/demo/eiger.py @@ -0,0 +1,182 @@ +"""Example 5 - introspectable controller: a cut-down Eiger over the fake REST sim. + +Half the attributes (``count_time``, ``state``) are declared as type hints and +checked by the current ``HintedAttribute`` introspection-validation mechanism; the +rest of the parameter tree is discovered at ``initialise()`` time by walking the +sim's ``keys`` endpoints and is added dynamically, with no static check. A device +that describes itself over the wire is exactly the case where introspection earns +its complexity - contrast with the (deliberately non-introspectable) SCPI/temperature +examples. +""" + +import enum +from dataclasses import KW_ONLY, dataclass +from typing import Any, cast + +import httpx + +from fastcs.attributes import AnyAttributeIO, AttributeIO, AttributeIORef, AttrR, AttrRW +from fastcs.controllers import Controller +from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String +from fastcs.demo.simulation.eiger import API_PREFIX, Subsystem, ValueType +from fastcs.util import ONCE + +_DATATYPES: dict[ValueType, type[DataType]] = { + "float": Float, + "int": Int, + "string": String, + "bool": Bool, +} + +# Poll period (seconds) for read-only status params that change on the device. +UPDATE_PERIOD = 0.2 + + +def _datatype(param: str, data: dict[str, Any]) -> DataType: + """Build a datatype for a parameter from the metadata the device reports. + + A parameter that reports ``allowed_values`` is discrete, so it becomes an `Enum` + over an enum class built from those values. The members are only knowable over the + wire, which is exactly the case introspection exists for. + """ + allowed_values = data.get("allowed_values") + if allowed_values is None: + return _DATATYPES[data["value_type"]]() + + name = "".join(part.title() for part in param.split("_")) + # The functional API builds a class; type checkers only see the instance signature. + enum_cls = cast( + type[enum.Enum], enum.Enum(name, {value: value for value in allowed_values}) + ) + return Enum(enum_cls) + + +@dataclass +class EigerConnectionSettings: + base_url: str = "http://localhost:8000" + + +class EigerConnection: + """Thin async HTTP client wrapper for the Eiger REST sim. + + A ``transport`` can be supplied to point directly at an in-process ASGI app + (e.g. in tests), bypassing the network entirely. + """ + + def __init__(self, transport: httpx.AsyncBaseTransport | None = None): + self._transport = transport + self._client: httpx.AsyncClient | None = None + + async def connect(self, settings: EigerConnectionSettings) -> None: + self._client = httpx.AsyncClient( + base_url=settings.base_url, transport=self._transport + ) + + async def close(self) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + + @property + def client(self) -> httpx.AsyncClient: + if self._client is None: + raise RuntimeError("EigerConnection is not connected") + return self._client + + async def keys(self, subsystem: Subsystem) -> list[str]: + response = await self.client.get(f"{API_PREFIX}/{subsystem}/keys") + response.raise_for_status() + return response.json() + + async def get(self, subsystem: Subsystem, param: str) -> dict: + response = await self.client.get(f"{API_PREFIX}/{subsystem}/{param}") + response.raise_for_status() + return response.json() + + async def put(self, subsystem: Subsystem, param: str, value) -> None: + response = await self.client.put( + f"{API_PREFIX}/{subsystem}/{param}", json={"value": value} + ) + response.raise_for_status() + + +@dataclass +class EigerAttributeIORef(AttributeIORef): + subsystem: Subsystem + param: str + _: KW_ONLY + update_period: float | None = ONCE + + +class EigerAttributeIO(AttributeIO[Any, EigerAttributeIORef]): + def __init__(self, connection: EigerConnection): + super().__init__() + self._connection = connection + + async def update(self, attr: AttrR[Any, EigerAttributeIORef]) -> None: + data = await self._connection.get(attr.io_ref.subsystem, attr.io_ref.param) + # No cast here - ``update`` validates against the datatype, which is the one + # place a bad value from the device should be coerced or complained about. + await attr.update(data["value"]) + + async def send(self, attr, value) -> None: + await self._connection.put(attr.io_ref.subsystem, attr.io_ref.param, value) + + +class EigerDetector(Controller): + """Cut-down Eiger controller: half declared, half introspected.""" + + # Declared (checked): must exist, with this access mode and dtype, after + # initialise() introspects the parameter tree. ``state`` is discrete, and its + # enum class is built from the ``allowed_values`` the device reports, so there + # is no author-time type to hint - only the access mode can be pinned here. + count_time: AttrRW[float] + state: AttrR + + # Derived (soft): built on top of the introspected ``state`` param. Declaring + # ``state`` as a checked attribute is what lets us reference it in code and + # publish something computed from it - here, whether the detector is idle. + idle = AttrR(Bool()) + + def __init__( + self, + settings: EigerConnectionSettings | None = None, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self.connection = EigerConnection(transport=transport) + ios: list[AnyAttributeIO] = [EigerAttributeIO(self.connection)] + super().__init__(ios=ios) + + self._settings = settings or EigerConnectionSettings() + + async def connect(self) -> None: + await self.connection.connect(self._settings) + self._connected = True + + async def disconnect(self) -> None: + await self.connection.close() + + async def initialise(self) -> None: + for subsystem in ("config", "status"): + for param in await self.connection.keys(subsystem): + data = await self.connection.get(subsystem, param) + datatype = _datatype(param, data) + + if data["access_mode"] == "rw": + io_ref = EigerAttributeIORef(subsystem=subsystem, param=param) + attr = AttrRW(datatype, io_ref=io_ref) + else: + # Read-only params are status values that change on the device, + # so poll them periodically rather than reading once. + io_ref = EigerAttributeIORef( + subsystem=subsystem, param=param, update_period=UPDATE_PERIOD + ) + attr = AttrR(datatype, io_ref=io_ref) + + self.add_attribute(param, attr) + + # Keep the derived ``idle`` flag in sync with the introspected ``state``. + self.state.add_on_update_callback(self._update_idle) + + async def _update_idle(self, state: enum.Enum) -> None: + await self.idle.update(state.value == "idle") diff --git a/src/fastcs/demo/simulation/eiger.py b/src/fastcs/demo/simulation/eiger.py new file mode 100644 index 000000000..b70488d7e --- /dev/null +++ b/src/fastcs/demo/simulation/eiger.py @@ -0,0 +1,141 @@ +"""A cut-down, Eiger-shaped fake REST device for the introspectable controller demo. + +Mimics the shape of a real Eiger detector's parameter-tree REST API (subsystems of +named parameters, a ``keys`` listing endpoint, per-parameter GET/PUT) without any of +the real detector logic. Introspection earns its complexity only when a device's +parameters aren't knowable at author time - this sim exists to give that a genuine, +self-describing backend to introspect. +""" + +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from dataclasses import dataclass +from typing import Any, Literal + +from fastapi import FastAPI, HTTPException + +ValueType = Literal["float", "int", "string", "bool"] +AccessMode = Literal["r", "rw"] +Subsystem = Literal["config", "status"] + +API_PREFIX = "/detector/api/1.8.0" + +# The sim flips its temperature between these two values so the front end has +# something visibly changing to poll. +TEMPERATURES = (20.0, 30.0) + + +@dataclass +class EigerParameter: + value: Any + value_type: ValueType + access_mode: AccessMode = "r" + allowed_values: list[str] | None = None + """The permitted values of a discrete parameter, as the real detector reports them. + + Only discrete parameters carry this, and it is the metadata a client needs to + introspect the parameter as an enum rather than a bare string. + """ + + +def _initial_state() -> dict[Subsystem, dict[str, EigerParameter]]: + return { + "config": { + "count_time": EigerParameter(0.1, "float", "rw"), + "frame_time": EigerParameter(0.1, "float", "rw"), + "nimages": EigerParameter(1, "int", "rw"), + "description": EigerParameter("Simulated Eiger", "string", "r"), + }, + "status": { + "state": EigerParameter( + "idle", "string", "r", allowed_values=["idle", "ready", "acquire"] + ), + "temperature": EigerParameter(22.5, "float", "r"), + "humidity": EigerParameter(32.1, "float", "r"), + }, + } + + +async def _oscillate_temperature( + parameter: EigerParameter, period: float = 0.5 +) -> None: + """Flip a temperature parameter between two known values forever. + + Runs as a background task under the app's lifespan (started by a real server, + e.g. uvicorn). The in-process ASGI transport used by the controller in tests + does not start lifespan events, so a test that wants the task running drives + the lifespan explicitly. + """ + index = 0 + while True: + await asyncio.sleep(period) + index = 1 - index + parameter.value = TEMPERATURES[index] + + +def create_eiger_sim_app() -> FastAPI: + """Create a FastAPI app simulating a cut-down Eiger detector REST API.""" + state = _initial_state() + + @asynccontextmanager + async def lifespan(app: FastAPI) -> AsyncIterator[None]: + task = asyncio.create_task( + _oscillate_temperature(state["status"]["temperature"]) + ) + try: + yield + finally: + task.cancel() + + app = FastAPI(lifespan=lifespan) + # Backdoor: expose the parameter tree so tests can set read-only values (e.g. + # ``state``, which has no PUT route) and then poll them through the controller. + app.state.sim = state + + def _subsystem(subsystem: str) -> dict[str, EigerParameter]: + try: + return state[subsystem] # type: ignore[index] + except KeyError: + raise HTTPException( + status_code=404, detail=f"Unknown subsystem '{subsystem}'" + ) from None + + def _parameter(subsystem: str, param: str) -> EigerParameter: + try: + return _subsystem(subsystem)[param] + except KeyError: + raise HTTPException( + status_code=404, detail=f"Unknown parameter '{param}'" + ) from None + + @app.get(API_PREFIX + "/{subsystem}/keys") + async def get_keys(subsystem: str) -> list[str]: + return list(_subsystem(subsystem)) + + @app.get(API_PREFIX + "/{subsystem}/{param}") + async def get_parameter(subsystem: str, param: str) -> dict[str, Any]: + parameter = _parameter(subsystem, param) + data: dict[str, Any] = { + "value": parameter.value, + "value_type": parameter.value_type, + "access_mode": parameter.access_mode, + } + # Only discrete parameters report their options, as on the real detector. + if parameter.allowed_values is not None: + data["allowed_values"] = parameter.allowed_values + return data + + @app.put(API_PREFIX + "/{subsystem}/{param}") + async def put_parameter( + subsystem: str, param: str, body: dict[str, Any] + ) -> dict[str, Any]: + parameter = _parameter(subsystem, param) + if parameter.access_mode != "rw": + raise HTTPException( + status_code=403, detail=f"Parameter '{param}' is read-only" + ) + parameter.value = body["value"] + return {"value": parameter.value} + + return app diff --git a/tests/demo/test_eiger.py b/tests/demo/test_eiger.py new file mode 100644 index 000000000..17c76be8c --- /dev/null +++ b/tests/demo/test_eiger.py @@ -0,0 +1,141 @@ +import asyncio +import enum + +import httpx +import pytest +import pytest_asyncio + +from fastcs.attributes import AttrR, AttrRW +from fastcs.datatypes import Enum +from fastcs.demo.eiger import UPDATE_PERIOD, EigerDetector +from fastcs.demo.simulation.eiger import EigerParameter, create_eiger_sim_app +from fastcs.util import ONCE + +# Backdoor to the sim's parameter tree, keyed by subsystem then param name. +SimState = dict[str, dict[str, EigerParameter]] + + +@pytest_asyncio.fixture +async def _eiger(): + app = create_eiger_sim_app() + controller = EigerDetector(transport=httpx.ASGITransport(app=app)) + await controller.connect() + await controller.initialise() + controller.post_initialise() + yield controller, app.state.sim + await controller.disconnect() + + +@pytest_asyncio.fixture +async def detector(_eiger) -> EigerDetector: + return _eiger[0] + + +@pytest_asyncio.fixture +async def sim(_eiger) -> SimState: + return _eiger[1] + + +@pytest.mark.asyncio +async def test_hinted_attributes_are_introspected(detector: EigerDetector): + assert isinstance(detector.count_time, AttrRW) + assert detector.count_time.datatype.dtype is float + + assert isinstance(detector.state, AttrR) + # ``state`` reports ``allowed_values``, so it is introspected as an enum whose + # members come from the device rather than as a bare string. + assert isinstance(detector.state.datatype, Enum) + assert detector.state.datatype.names == ["idle", "ready", "acquire"] + + +@pytest.mark.asyncio +async def test_enum_attribute_reads_as_member(detector: EigerDetector, sim: SimState): + sim["status"]["state"].value = "acquire" + await detector.state.bind_update_callback()() + + state = detector.state.get() + assert isinstance(state, enum.Enum) + assert state.value == "acquire" + + +@pytest.mark.asyncio +async def test_unhinted_attributes_are_also_introspected(detector: EigerDetector): + for name in ("frame_time", "nimages", "description", "temperature", "humidity"): + assert name in detector.attributes + + +@pytest.mark.asyncio +async def test_read_attribute_from_device(detector: EigerDetector): + await detector.count_time.bind_update_callback()() + assert detector.count_time.get() == 0.1 + + humidity = detector.attributes["humidity"] + assert isinstance(humidity, AttrR) + await humidity.bind_update_callback()() + assert humidity.get() == 32.1 + + +@pytest.mark.asyncio +async def test_write_attribute_to_device(detector: EigerDetector): + await detector.count_time.put(0.5) + + # Read it back through the attribute to confirm the round-trip to the device. + await detector.count_time.bind_update_callback()() + assert detector.count_time.get() == 0.5 + + +@pytest.mark.asyncio +async def test_idle_derived_from_state(detector: EigerDetector, sim: SimState): + # ``idle`` is soft and starts at its default, tracking ``state`` once polled. + assert detector.idle.get() is False + + # Poke the read-only ``state`` via the sim backdoor, then poll the attribute. + sim["status"]["state"].value = "acquire" + await detector.state.bind_update_callback()() + assert detector.idle.get() is False + + sim["status"]["state"].value = "idle" + await detector.state.bind_update_callback()() + assert detector.idle.get() is True + + +@pytest.mark.asyncio +async def test_read_only_params_poll_but_rw_read_once(detector: EigerDetector): + for name in ("state", "temperature", "humidity", "description"): + attr = detector.attributes[name] + assert isinstance(attr, AttrR) and not isinstance(attr, AttrRW) + assert attr.io_ref.update_period == UPDATE_PERIOD + + assert detector.count_time.io_ref.update_period is ONCE + + +@pytest.mark.asyncio +async def test_temperature_oscillation_seen_via_subscribe(): + # The oscillation task runs under the app lifespan, so drive the lifespan here + # (the bare ASGI transport used elsewhere does not start it). Observe it through + # the controller's temperature attribute, subscribing for updates. + app = create_eiger_sim_app() + async with app.router.lifespan_context(app): + controller = EigerDetector(transport=httpx.ASGITransport(app=app)) + await controller.connect() + await controller.initialise() + controller.post_initialise() + + temperature = controller.attributes["temperature"] + assert isinstance(temperature, AttrR) + + seen: list[float] = [] + + async def record(value: float) -> None: + seen.append(value) + + temperature.add_on_update_callback(record) + + # Poll across several sim flips (every 0.5s) so the value changes under us. + for _ in range(8): + await temperature.bind_update_callback()() + await asyncio.sleep(0.2) + + await controller.disconnect() + + assert len(set(seen)) > 1, f"temperature did not change: {seen}"