From cf155075d2daec56cf7b86d72e68a2ccbabc6125 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Jul 2026 22:18:17 +0000 Subject: [PATCH 1/5] demo: cut-down Eiger REST sim + introspectable controller example Add a FastAPI fake REST sim (demo/simulation/eiger.py) shaped like a detector parameter tree (subsystems of named parameters, a keys listing endpoint, per-parameter GET/PUT), and an EigerDetector controller (demo/eiger.py) that type-hints half its attributes (checked via the current HintedAttribute mechanism) and fills the rest by introspecting the sim's keys endpoints in initialise(). Baseline uses the current API (AttrR/AttrRW + io_ref/AttributeIO); migrates to ControllerFiller when #394 lands. Closes #391 --- src/fastcs/demo/eiger.py | 139 ++++++++++++++++++++++++++++ src/fastcs/demo/simulation/eiger.py | 91 ++++++++++++++++++ tests/demo/test_eiger.py | 89 ++++++++++++++++++ 3 files changed, 319 insertions(+) create mode 100644 src/fastcs/demo/eiger.py create mode 100644 src/fastcs/demo/simulation/eiger.py create mode 100644 tests/demo/test_eiger.py diff --git a/src/fastcs/demo/eiger.py b/src/fastcs/demo/eiger.py new file mode 100644 index 000000000..7584544e6 --- /dev/null +++ b/src/fastcs/demo/eiger.py @@ -0,0 +1,139 @@ +"""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. +""" + +from dataclasses import KW_ONLY, dataclass +from typing import Any + +import httpx + +from fastcs.attributes import AnyAttributeIO, AttributeIO, AttributeIORef, AttrR, AttrRW +from fastcs.controllers import Controller +from fastcs.datatypes import Bool, DataType, 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, +} + + +@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) + await attr.update(attr.dtype(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. + count_time: AttrRW[float] + state: AttrR[str] + + 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_cls = _DATATYPES[data["value_type"]] + io_ref = EigerAttributeIORef(subsystem=subsystem, param=param) + + if data["access_mode"] == "rw": + attr = AttrRW(datatype_cls(), io_ref=io_ref) + else: + attr = AttrR(datatype_cls(), io_ref=io_ref) + + self.add_attribute(param, attr) diff --git a/src/fastcs/demo/simulation/eiger.py b/src/fastcs/demo/simulation/eiger.py new file mode 100644 index 000000000..18282feb2 --- /dev/null +++ b/src/fastcs/demo/simulation/eiger.py @@ -0,0 +1,91 @@ +"""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. +""" + +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" + + +@dataclass +class EigerParameter: + value: Any + value_type: ValueType + access_mode: AccessMode = "r" + + +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"), + "temperature": EigerParameter(22.5, "float", "r"), + "humidity": EigerParameter(32.1, "float", "r"), + }, + } + + +def create_eiger_sim_app() -> FastAPI: + """Create a FastAPI app simulating a cut-down Eiger detector REST API.""" + app = FastAPI() + state = _initial_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) + return { + "value": parameter.value, + "value_type": parameter.value_type, + "access_mode": parameter.access_mode, + } + + @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..f6b8142f8 --- /dev/null +++ b/tests/demo/test_eiger.py @@ -0,0 +1,89 @@ +import httpx +import pytest +import pytest_asyncio +from fastapi.testclient import TestClient + +from fastcs.attributes import AttrR, AttrRW +from fastcs.demo.eiger import EigerDetector +from fastcs.demo.simulation.eiger import API_PREFIX, create_eiger_sim_app + + +@pytest.fixture +def sim_client() -> TestClient: + return TestClient(create_eiger_sim_app()) + + +def test_sim_lists_keys(sim_client: TestClient): + response = sim_client.get(f"{API_PREFIX}/config/keys") + assert response.status_code == 200 + assert set(response.json()) >= {"count_time", "frame_time", "nimages"} + + +def test_sim_get_parameter(sim_client: TestClient): + response = sim_client.get(f"{API_PREFIX}/config/count_time") + assert response.status_code == 200 + body = response.json() + assert body == {"value": 0.1, "value_type": "float", "access_mode": "rw"} + + +def test_sim_put_parameter(sim_client: TestClient): + response = sim_client.put(f"{API_PREFIX}/config/count_time", json={"value": 0.5}) + assert response.status_code == 200 + assert response.json() == {"value": 0.5} + + response = sim_client.get(f"{API_PREFIX}/config/count_time") + assert response.json()["value"] == 0.5 + + +def test_sim_put_read_only_parameter_rejected(sim_client: TestClient): + response = sim_client.put(f"{API_PREFIX}/status/state", json={"value": "busy"}) + assert response.status_code == 403 + + +def test_sim_unknown_parameter_404(sim_client: TestClient): + assert sim_client.get(f"{API_PREFIX}/config/nonexistent").status_code == 404 + assert sim_client.get(f"{API_PREFIX}/nonexistent/keys").status_code == 404 + + +@pytest_asyncio.fixture +async def detector() -> EigerDetector: + transport = httpx.ASGITransport(app=create_eiger_sim_app()) + controller = EigerDetector(transport=transport) + await controller.connect() + await controller.initialise() + controller.post_initialise() + return controller + + +@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) + assert detector.state.datatype.dtype is str + + +@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 + + temperature = detector.attributes["temperature"] + assert isinstance(temperature, AttrR) + await temperature.bind_update_callback()() + assert temperature.get() == 22.5 + + +@pytest.mark.asyncio +async def test_write_attribute_to_device(detector: EigerDetector): + await detector.count_time.put(0.5) + + response = await detector.connection.get("config", "count_time") + assert response["value"] == 0.5 From c4cb0fca347333ce0f951f00ef5f097caf4c623b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Jul 2026 22:21:17 +0000 Subject: [PATCH 2/5] docs: ignore unresolvable httpx/fastapi type refs in nitpicky mode sphinx-build --fail-on-warning was erroring on autodoc cross-references to httpx.AsyncBaseTransport and fastapi.applications.FastAPI, which have no intersphinx mapping - same class of issue already worked around for p4p types. --- docs/conf.py | 3 +++ 1 file changed, 3 insertions(+) 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"), From ff6292b0f9a00ee0998ab01f13ea4e82687bb336 Mon Sep 17 00:00:00 2001 From: Tom Cobb Date: Thu, 23 Jul 2026 14:17:59 +0000 Subject: [PATCH 3/5] demo(#391): address review - idle derived attr, temp oscillation, poll read-only params - eiger.py: add soft `idle: AttrR[bool]` derived from the introspected `state` param (state == "idle"), kept in sync via an on-update callback - shows why we declare `state` as a checked attribute (to build code on top of it). - eiger.py: give read-only params a poll `update_period` in `initialise()`; rw params still read once (ONCE). - simulation/eiger.py: add a lifespan background task that sweeps `temperature` between two values so the front end shows something updating (real server only; the in-process ASGI transport used in tests stays deterministic). - tests: cover idle-from-state, read-only poll vs rw read-once, and oscillation. Co-Authored-By: Claude Opus 4.8 --- src/fastcs/demo/eiger.py | 21 +++++++++++++- src/fastcs/demo/simulation/eiger.py | 41 ++++++++++++++++++++++++++- tests/demo/test_eiger.py | 44 ++++++++++++++++++++++++++++- 3 files changed, 103 insertions(+), 3 deletions(-) diff --git a/src/fastcs/demo/eiger.py b/src/fastcs/demo/eiger.py index 7584544e6..021dad7ca 100644 --- a/src/fastcs/demo/eiger.py +++ b/src/fastcs/demo/eiger.py @@ -27,6 +27,9 @@ "bool": Bool, } +# Poll period (seconds) for read-only status params that change on the device. +UPDATE_PERIOD = 0.2 + @dataclass class EigerConnectionSettings: @@ -106,6 +109,11 @@ class EigerDetector(Controller): count_time: AttrRW[float] state: AttrR[str] + # 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, @@ -129,11 +137,22 @@ async def initialise(self) -> None: for param in await self.connection.keys(subsystem): data = await self.connection.get(subsystem, param) datatype_cls = _DATATYPES[data["value_type"]] - io_ref = EigerAttributeIORef(subsystem=subsystem, param=param) if data["access_mode"] == "rw": + io_ref = EigerAttributeIORef(subsystem=subsystem, param=param) attr = AttrRW(datatype_cls(), 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_cls(), 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: str) -> None: + await self.idle.update(state == "idle") diff --git a/src/fastcs/demo/simulation/eiger.py b/src/fastcs/demo/simulation/eiger.py index 18282feb2..000bf63e6 100644 --- a/src/fastcs/demo/simulation/eiger.py +++ b/src/fastcs/demo/simulation/eiger.py @@ -7,6 +7,11 @@ self-describing backend to introspect. """ +import asyncio +import math +import time +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from dataclasses import dataclass from typing import Any, Literal @@ -42,11 +47,45 @@ def _initial_state() -> dict[Subsystem, dict[str, EigerParameter]]: } +async def _oscillate_temperature( + parameter: EigerParameter, + low: float = 20.0, + high: float = 30.0, + period: float = 10.0, +) -> None: + """Slowly sweep a temperature parameter between two values, forever. + + Gives the front end something visibly changing to poll. Runs as a background + task under the app's lifespan (started by a real server, e.g. uvicorn; not by + the in-process ASGI transport used in tests, which keeps those deterministic). + """ + mid = (low + high) / 2 + amplitude = (high - low) / 2 + start = time.monotonic() + while True: + elapsed = time.monotonic() - start + parameter.value = round( + mid + amplitude * math.sin(2 * math.pi * elapsed / period), 1 + ) + await asyncio.sleep(0.1) + + def create_eiger_sim_app() -> FastAPI: """Create a FastAPI app simulating a cut-down Eiger detector REST API.""" - app = FastAPI() 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) + def _subsystem(subsystem: str) -> dict[str, EigerParameter]: try: return state[subsystem] # type: ignore[index] diff --git a/tests/demo/test_eiger.py b/tests/demo/test_eiger.py index f6b8142f8..8f5e4b6e1 100644 --- a/tests/demo/test_eiger.py +++ b/tests/demo/test_eiger.py @@ -1,11 +1,14 @@ +import asyncio + import httpx import pytest import pytest_asyncio from fastapi.testclient import TestClient from fastcs.attributes import AttrR, AttrRW -from fastcs.demo.eiger import EigerDetector +from fastcs.demo.eiger import UPDATE_PERIOD, EigerDetector from fastcs.demo.simulation.eiger import API_PREFIX, create_eiger_sim_app +from fastcs.util import ONCE @pytest.fixture @@ -87,3 +90,42 @@ async def test_write_attribute_to_device(detector: EigerDetector): response = await detector.connection.get("config", "count_time") assert response["value"] == 0.5 + + +@pytest.mark.asyncio +async def test_idle_derived_from_state(detector: EigerDetector): + # ``idle`` is soft and starts at its default, tracking ``state`` once polled. + assert detector.idle.get() is False + + await detector.state.update("idle") + assert detector.idle.get() is True + + await detector.state.update("acquire") + assert detector.idle.get() is False + + +@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_sim_temperature_oscillates(): + # The background task only runs under the app lifespan (a real server), not the + # bare ASGI transport used elsewhere, so drive the lifespan explicitly here. + app = create_eiger_sim_app() + transport = httpx.ASGITransport(app=app) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient(base_url="http://sim", transport=transport) as c: + readings = [] + for _ in range(4): + await asyncio.sleep(0.3) + response = await c.get(f"{API_PREFIX}/status/temperature") + readings.append(response.json()["value"]) + + assert len(set(readings)) > 1, f"temperature did not change: {readings}" From da9988c5d8ed22a223b87a4c75eeaf8aef988b0c Mon Sep 17 00:00:00 2001 From: Tom Cobb Date: Thu, 23 Jul 2026 14:59:15 +0000 Subject: [PATCH 4/5] demo(#391): address review - simpler sim flip, backdoor, front-door tests - simulation/eiger.py: replace the sine sweep with a simple flip between two known temperatures every 0.5s (predictable); expose the parameter tree via `app.state.sim` as a test backdoor for read-only params with no PUT route. - tests: drop the sim-only tests; drive everything through the controller attributes. Idle test now pokes `state` via the sim backdoor and polls the attribute (rather than calling AttrR.update directly). Oscillation test builds a controller under the app lifespan and observes temperature via subscribe. Co-Authored-By: Claude Opus 4.8 --- src/fastcs/demo/simulation/eiger.py | 35 ++++---- tests/demo/test_eiger.py | 119 +++++++++++++--------------- 2 files changed, 74 insertions(+), 80 deletions(-) diff --git a/src/fastcs/demo/simulation/eiger.py b/src/fastcs/demo/simulation/eiger.py index 000bf63e6..397d31cde 100644 --- a/src/fastcs/demo/simulation/eiger.py +++ b/src/fastcs/demo/simulation/eiger.py @@ -8,8 +8,6 @@ """ import asyncio -import math -import time from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass @@ -23,6 +21,10 @@ 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: @@ -48,26 +50,20 @@ def _initial_state() -> dict[Subsystem, dict[str, EigerParameter]]: async def _oscillate_temperature( - parameter: EigerParameter, - low: float = 20.0, - high: float = 30.0, - period: float = 10.0, + parameter: EigerParameter, period: float = 0.5 ) -> None: - """Slowly sweep a temperature parameter between two values, forever. + """Flip a temperature parameter between two known values forever. - Gives the front end something visibly changing to poll. Runs as a background - task under the app's lifespan (started by a real server, e.g. uvicorn; not by - the in-process ASGI transport used in tests, which keeps those deterministic). + 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. """ - mid = (low + high) / 2 - amplitude = (high - low) / 2 - start = time.monotonic() + index = 0 while True: - elapsed = time.monotonic() - start - parameter.value = round( - mid + amplitude * math.sin(2 * math.pi * elapsed / period), 1 - ) - await asyncio.sleep(0.1) + await asyncio.sleep(period) + index = 1 - index + parameter.value = TEMPERATURES[index] def create_eiger_sim_app() -> FastAPI: @@ -85,6 +81,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: 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: diff --git a/tests/demo/test_eiger.py b/tests/demo/test_eiger.py index 8f5e4b6e1..04029190a 100644 --- a/tests/demo/test_eiger.py +++ b/tests/demo/test_eiger.py @@ -3,59 +3,35 @@ import httpx import pytest import pytest_asyncio -from fastapi.testclient import TestClient from fastcs.attributes import AttrR, AttrRW from fastcs.demo.eiger import UPDATE_PERIOD, EigerDetector -from fastcs.demo.simulation.eiger import API_PREFIX, create_eiger_sim_app +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.fixture -def sim_client() -> TestClient: - return TestClient(create_eiger_sim_app()) - -def test_sim_lists_keys(sim_client: TestClient): - response = sim_client.get(f"{API_PREFIX}/config/keys") - assert response.status_code == 200 - assert set(response.json()) >= {"count_time", "frame_time", "nimages"} - - -def test_sim_get_parameter(sim_client: TestClient): - response = sim_client.get(f"{API_PREFIX}/config/count_time") - assert response.status_code == 200 - body = response.json() - assert body == {"value": 0.1, "value_type": "float", "access_mode": "rw"} - - -def test_sim_put_parameter(sim_client: TestClient): - response = sim_client.put(f"{API_PREFIX}/config/count_time", json={"value": 0.5}) - assert response.status_code == 200 - assert response.json() == {"value": 0.5} - - response = sim_client.get(f"{API_PREFIX}/config/count_time") - assert response.json()["value"] == 0.5 - - -def test_sim_put_read_only_parameter_rejected(sim_client: TestClient): - response = sim_client.put(f"{API_PREFIX}/status/state", json={"value": "busy"}) - assert response.status_code == 403 +@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() -def test_sim_unknown_parameter_404(sim_client: TestClient): - assert sim_client.get(f"{API_PREFIX}/config/nonexistent").status_code == 404 - assert sim_client.get(f"{API_PREFIX}/nonexistent/keys").status_code == 404 +@pytest_asyncio.fixture +async def detector(_eiger) -> EigerDetector: + return _eiger[0] @pytest_asyncio.fixture -async def detector() -> EigerDetector: - transport = httpx.ASGITransport(app=create_eiger_sim_app()) - controller = EigerDetector(transport=transport) - await controller.connect() - await controller.initialise() - controller.post_initialise() - return controller +async def sim(_eiger) -> SimState: + return _eiger[1] @pytest.mark.asyncio @@ -78,29 +54,33 @@ async def test_read_attribute_from_device(detector: EigerDetector): await detector.count_time.bind_update_callback()() assert detector.count_time.get() == 0.1 - temperature = detector.attributes["temperature"] - assert isinstance(temperature, AttrR) - await temperature.bind_update_callback()() - assert temperature.get() == 22.5 + 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) - response = await detector.connection.get("config", "count_time") - assert response["value"] == 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): +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 - await detector.state.update("idle") + # Poke the read-only ``state`` via the sim backdoor, then poll the attribute. + sim["status"]["state"].value = "idle" + await detector.state.bind_update_callback()() assert detector.idle.get() is True - await detector.state.update("acquire") + sim["status"]["state"].value = "acquire" + await detector.state.bind_update_callback()() assert detector.idle.get() is False @@ -115,17 +95,32 @@ async def test_read_only_params_poll_but_rw_read_once(detector: EigerDetector): @pytest.mark.asyncio -async def test_sim_temperature_oscillates(): - # The background task only runs under the app lifespan (a real server), not the - # bare ASGI transport used elsewhere, so drive the lifespan explicitly here. +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() - transport = httpx.ASGITransport(app=app) async with app.router.lifespan_context(app): - async with httpx.AsyncClient(base_url="http://sim", transport=transport) as c: - readings = [] - for _ in range(4): - await asyncio.sleep(0.3) - response = await c.get(f"{API_PREFIX}/status/temperature") - readings.append(response.json()["value"]) - - assert len(set(readings)) > 1, f"temperature did not change: {readings}" + 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}" From 34196c6a62377cd335c12ded388398aaf0b57236 Mon Sep 17 00:00:00 2001 From: Tom Cobb Date: Thu, 30 Jul 2026 16:27:51 +0000 Subject: [PATCH 5/5] demo(#391): introspect state as an enum from allowed_values; drop IO cast - Sim: EigerParameter gains allowed_values, reported by GET only for discrete params (as the real detector does). state now advertises [idle, ready, acquire]. - Controller: a param reporting allowed_values is introspected as an Enum over an enum class built from those values. The members are only knowable over the wire, so state's hint drops to a bare AttrR - the exact-dtype hint check has no author-time class to match against. - EigerAttributeIO.update no longer casts to the dtype; attr.update validates, which is the one place a bad device value should be coerced or complained about. Co-Authored-By: Claude Opus 5 --- src/fastcs/demo/eiger.py | 44 ++++++++++++++++++++++------- src/fastcs/demo/simulation/eiger.py | 16 +++++++++-- tests/demo/test_eiger.py | 25 ++++++++++++---- 3 files changed, 68 insertions(+), 17 deletions(-) diff --git a/src/fastcs/demo/eiger.py b/src/fastcs/demo/eiger.py index 021dad7ca..7ebea02fe 100644 --- a/src/fastcs/demo/eiger.py +++ b/src/fastcs/demo/eiger.py @@ -9,14 +9,15 @@ examples. """ +import enum from dataclasses import KW_ONLY, dataclass -from typing import Any +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, Float, Int, String +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 @@ -31,6 +32,25 @@ 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" @@ -95,7 +115,9 @@ def __init__(self, connection: EigerConnection): async def update(self, attr: AttrR[Any, EigerAttributeIORef]) -> None: data = await self._connection.get(attr.io_ref.subsystem, attr.io_ref.param) - await attr.update(attr.dtype(data["value"])) + # 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) @@ -105,9 +127,11 @@ 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. + # 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[str] + 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 @@ -136,23 +160,23 @@ 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_cls = _DATATYPES[data["value_type"]] + datatype = _datatype(param, data) if data["access_mode"] == "rw": io_ref = EigerAttributeIORef(subsystem=subsystem, param=param) - attr = AttrRW(datatype_cls(), io_ref=io_ref) + 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_cls(), io_ref=io_ref) + 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: str) -> None: - await self.idle.update(state == "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 index 397d31cde..b70488d7e 100644 --- a/src/fastcs/demo/simulation/eiger.py +++ b/src/fastcs/demo/simulation/eiger.py @@ -31,6 +31,12 @@ 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]]: @@ -42,7 +48,9 @@ def _initial_state() -> dict[Subsystem, dict[str, EigerParameter]]: "description": EigerParameter("Simulated Eiger", "string", "r"), }, "status": { - "state": EigerParameter("idle", "string", "r"), + "state": EigerParameter( + "idle", "string", "r", allowed_values=["idle", "ready", "acquire"] + ), "temperature": EigerParameter(22.5, "float", "r"), "humidity": EigerParameter(32.1, "float", "r"), }, @@ -108,11 +116,15 @@ async def get_keys(subsystem: str) -> list[str]: @app.get(API_PREFIX + "/{subsystem}/{param}") async def get_parameter(subsystem: str, param: str) -> dict[str, Any]: parameter = _parameter(subsystem, param) - return { + 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( diff --git a/tests/demo/test_eiger.py b/tests/demo/test_eiger.py index 04029190a..17c76be8c 100644 --- a/tests/demo/test_eiger.py +++ b/tests/demo/test_eiger.py @@ -1,10 +1,12 @@ 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 @@ -40,7 +42,20 @@ async def test_hinted_attributes_are_introspected(detector: EigerDetector): assert detector.count_time.datatype.dtype is float assert isinstance(detector.state, AttrR) - assert detector.state.datatype.dtype is str + # ``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 @@ -75,14 +90,14 @@ async def test_idle_derived_from_state(detector: EigerDetector, sim: SimState): assert detector.idle.get() is False # Poke the read-only ``state`` via the sim backdoor, then poll the attribute. - sim["status"]["state"].value = "idle" - await detector.state.bind_update_callback()() - assert detector.idle.get() is True - 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):