From a5914c736018ed5681ffb3cbb21badaaac126482 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 23:09:45 +0000 Subject: [PATCH 1/2] controllers(#395): ControllerRunner, native timestamps and severity - `ControllerRunner` owns the controller lifecycle - initialise, connect, the initial and periodic tasks, reconnect, disconnect - with no transport or interactive-shell concerns. `FastCS.serve` becomes a caller of it. Starting is in two halves so a transport can be wired to the APIs before the first values are read; `start()` alone does both, for an embedder that does not need them in between. - The runner also owns reconnect. A scan task that raises marks its controller disconnected and pauses; until now nothing ever called `reconnect()`, so it stayed paused unless the driver wired its own recovery. - `Controller.connected` exposes the connection state that was only readable through the private `_connected`. - A value entering an attribute may carry when it was obtained and how wrong it is, via `Update(timestamp=..., severity=...)`; a bare value is stamped on arrival and reported as no alarm. `Severity` is a FastCS enum using the same strings as EPICS. `AttrR.timestamp` and `AttrR.severity` read them back. - Documents the stable interface an embedder is restricted to. The `AttrW` setpoint cache the issue also lists was already delivered by #412, as the `.setpoint` property ADR 0016 settled on. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017zpFjvbhUisfVUq3NH8M1G --- docs/explanations/stable-interface.md | 112 ++++++++++++++++ src/fastcs/attributes/__init__.py | 1 + src/fastcs/attributes/attr_r.py | 33 +++++ src/fastcs/attributes/severity.py | 20 +++ src/fastcs/attributes/update.py | 7 + src/fastcs/control_system.py | 60 ++------- src/fastcs/controllers/__init__.py | 1 + src/fastcs/controllers/controller.py | 9 ++ src/fastcs/controllers/runner.py | 157 ++++++++++++++++++++++ tests/test_attributes.py | 76 ++++++++++- tests/test_control_system.py | 13 +- tests/test_controller_runner.py | 180 ++++++++++++++++++++++++++ tests/test_multi_controller.py | 10 +- 13 files changed, 616 insertions(+), 63 deletions(-) create mode 100644 docs/explanations/stable-interface.md create mode 100644 src/fastcs/attributes/severity.py create mode 100644 src/fastcs/controllers/runner.py create mode 100644 tests/test_controller_runner.py diff --git a/docs/explanations/stable-interface.md b/docs/explanations/stable-interface.md new file mode 100644 index 000000000..52f9c4b21 --- /dev/null +++ b/docs/explanations/stable-interface.md @@ -0,0 +1,112 @@ +# The Stable Interface + +Most of FastCS is free to change while it is pre-1.0. A narrow part of it is +not: the surface an *embedder* uses — code that runs FastCS controllers inside +another framework rather than serving them over a transport, such as the +ophyd-async connector. That surface is listed here, and an embedder should use +nothing outside it. In particular, nothing should reach into `BaseController`. + +## Running controllers: `ControllerRunner` + +`ControllerRunner` owns the controller lifecycle and nothing else — no +transports, no interactive shell. `FastCS` is a caller of it. + +```python +from fastcs.controllers import ControllerRunner + +runner = ControllerRunner(controller) +apis = await runner.setup() # initialise, and build the ControllerAPIs +await runner.start() # connect, run initial tasks, start scanning +... +await runner.stop() # stop the tasks, disconnect +``` + +- **`setup()`** runs `initialise()` and `post_initialise()` on each controller + and builds their `ControllerAPI`s. It exists as a separate step because + anything serving the controllers has to register its callbacks *before* the + first values are read, or it misses them. +- **`start()`** connects the controllers, runs the initial (`ONCE`) tasks, and + starts the periodic ones. It runs `setup()` first if you have not, so an + embedder that does not need the APIs in between can just call `start()`. +- **`stop()`** cancels the tasks and disconnects. + +**Idempotency is the caller's responsibility.** Starting a running runner, or +stopping a stopped one, is not defined — an embedder whose own connect may run +more than once has to keep track itself. + +The runner also owns **reconnect**. A scan task whose callback raises marks its +controller disconnected and pauses rather than dying; the runner notices and +calls `Controller.reconnect()` until it comes back. This is deliberately not +left to each controller, so every controller recovers the same way. + +## Reading the structure: `ControllerAPI` + +`ControllerAPI` is the read-only view of a controller: + +- `attributes` — the `Attribute`s, by name +- `command_methods` — the `Command`s, by name +- `scan_methods` — the `Scan`s, by name +- `sub_apis` — child `ControllerAPI`s, by name +- `path` and `description` +- `walk_api()` — this API and every descendant + +## Reading and writing values: the attribute surface + +For an `AttrR` (and so an `AttrRW`): + +- `readback` — the last known value +- `timestamp` — when that value was obtained, as a unix timestamp: the time the + source reported if it reported one, otherwise the time the update arrived +- `severity` — how wrong that value is, as a `Severity` +- `await poll()` — read a fresh value from the getter, cache it, return it +- `await update(value)` — push a value into the cache without any IO. Accepts a + bare value or an `Update`, which may carry a timestamp and severity +- `add_readback_callback(cb)` — be told when the readback changes + +For an `AttrW` (and so an `AttrRW`): + +- `setpoint` — the last value asked for. Cached by `set()` *before* the setter + runs and regardless of whether it succeeds, so it answers "what did we last + ask for", distinct from `readback`'s "what did we last read" +- `await set(value)` — cache the setpoint and apply it through the setter +- `add_setpoint_callback(cb)` — be told when the setpoint changes + +For any attribute: `dtype`, `access_mode`, `description`, `group`, and the +metadata it carries. + +## Calling actions: the command surface + +- `await command()` — call it +- `command.signature` — what it takes and returns + +## Timestamps and severity + +A value entering an attribute may carry when it was obtained and how wrong it +is, by arriving as an `Update`: + +```python +from fastcs.attributes import Severity, Update + +async def get_temperature() -> Update[float]: + value, device_time = await protocol.read_with_timestamp() + return Update(readback=value, timestamp=device_time) + +async def get_status() -> Update[float]: + value, fault = await protocol.read_status() + return Update( + readback=value, + severity=Severity.MAJOR if fault else Severity.NO_ALARM, + ) +``` + +A bare value is stamped with the time it arrived and reported as +`Severity.NO_ALARM`. This matters because a device that already knows when a +value was measured — an EPICS record timestamp, a Tango event — otherwise has +nowhere to say so, and the reading silently becomes "whenever FastCS heard +about it". + +`Severity` is a FastCS enum that uses the same strings as EPICS alarm +severities, so a driver or transport speaking EPICS does not have to translate. +It is not EPICS-specific. The value/timestamp/severity trio follows the shape of +bluesky's `Reading` so that the two read the same way, but shares no code with +it. diff --git a/src/fastcs/attributes/__init__.py b/src/fastcs/attributes/__init__.py index e968192b2..b9e08edcd 100644 --- a/src/fastcs/attributes/__init__.py +++ b/src/fastcs/attributes/__init__.py @@ -9,4 +9,5 @@ from .attribute import Attribute as Attribute from .attribute import AttributeAccessMode as AttributeAccessMode from .hinted_attribute import HintedAttribute as HintedAttribute +from .severity import Severity as Severity from .update import Update as Update diff --git a/src/fastcs/attributes/attr_r.py b/src/fastcs/attributes/attr_r.py index 3e4243d5a..3386cc3b3 100644 --- a/src/fastcs/attributes/attr_r.py +++ b/src/fastcs/attributes/attr_r.py @@ -1,12 +1,14 @@ from __future__ import annotations import asyncio +import time from collections.abc import Awaitable, Callable, Coroutine from dataclasses import KW_ONLY, dataclass, replace from typing import Any, Generic from fastcs.attributes._infer_datatype import infer_datatype_from_getter from fastcs.attributes.attribute import Attribute, AttributeAccessMode +from fastcs.attributes.severity import Severity from fastcs.attributes.update import Update from fastcs.attributes.util import AttrValuePredicate, PredicateEvent from fastcs.datatypes import DataType, DType_T @@ -95,6 +97,10 @@ def __init__( self._value: DType_T = ( self._datatype.initial_value if initial_value is None else initial_value ) + self._timestamp: float = time.time() + """When the cached value was obtained, or when the attribute was created""" + self._severity: Severity = Severity.NO_ALARM + """How wrong the cached value is, as last reported""" self._getter = resolved_getter self._poll_period: float | None = poll_period """Period in seconds between calls to poll(), or ONCE, or None (on-demand)""" @@ -110,6 +116,20 @@ def readback(self) -> DType_T: """The last known value of the attribute.""" return self._value + @property + def timestamp(self) -> float: + """When the last known value was obtained, as a unix timestamp. + + The time the source reported, if it reported one, and otherwise the + time the update reached FastCS. + """ + return self._timestamp + + @property + def severity(self) -> Severity: + """How wrong the last known value is, as the source last reported.""" + return self._severity + def has_getter(self) -> bool: return self._getter is not None @@ -134,6 +154,10 @@ async def update(self, value: DType_T | Update[DType_T]) -> None: To request a change to the setpoint of the attribute, use the ``set`` method, which will attempt to apply the change to the underlying source. + A value that arrives as an ``Update`` may carry the time it was obtained + and how wrong it is; a bare value is stamped with the time it arrived and + reported as ``Severity.NO_ALARM``. + Args: value: The new value of the attribute, or an ``Update`` wrapping it @@ -141,8 +165,13 @@ async def update(self, value: DType_T | Update[DType_T]) -> None: ValueError: If the value fails to be validated to DType_T """ + received_at = time.time() if isinstance(value, Update): + timestamp = received_at if value.timestamp is None else value.timestamp + severity = Severity.NO_ALARM if value.severity is None else value.severity value = value.readback + else: + timestamp, severity = received_at, Severity.NO_ALARM self.log_event("Attribute set", value=repr(value), attribute=self) @@ -153,6 +182,10 @@ async def update(self, value: DType_T | Update[DType_T]) -> None: logger.error("Failed to validate value", value=repr(value), attribute=self) raise + # Only once the value is known good, so a rejected update leaves the + # cached value and the time it was obtained agreeing with each other. + self._timestamp, self._severity = timestamp, severity + self.log_event("Value validated", value=repr(self._value), attribute=self) self._on_update_events -= { diff --git a/src/fastcs/attributes/severity.py b/src/fastcs/attributes/severity.py new file mode 100644 index 000000000..cbead56b7 --- /dev/null +++ b/src/fastcs/attributes/severity.py @@ -0,0 +1,20 @@ +from enum import Enum + + +class Severity(Enum): + """How wrong a value is, if at all. + + A FastCS-native enum that happens to use the same strings as EPICS alarm + severities, so a driver or transport speaking EPICS does not have to + translate. It is not EPICS-specific: a Tango event push or any other IO can + report severity the same way, through `Update`. + """ + + NO_ALARM = "NO_ALARM" + """The value is good""" + MINOR = "MINOR" + """The value is outside its warning range, or the device reports a minor fault""" + MAJOR = "MAJOR" + """The value is outside its alarm range, or the device reports a major fault""" + INVALID = "INVALID" + """The value could not be read, or cannot be trusted""" diff --git a/src/fastcs/attributes/update.py b/src/fastcs/attributes/update.py index be84f4be2..dde4addb4 100644 --- a/src/fastcs/attributes/update.py +++ b/src/fastcs/attributes/update.py @@ -3,6 +3,7 @@ from dataclasses import dataclass from typing import Generic +from fastcs.attributes.severity import Severity from fastcs.datatypes import DType_T @@ -15,14 +16,20 @@ class Update(Generic[DType_T]): - ``timestamp`` - when the value was obtained. ``None`` means the framework should stamp it with the time the update was received. + - ``severity`` - how wrong the value is, if the source says. ``None`` means + the source did not report one, which is read as ``Severity.NO_ALARM``. - ``setpoint`` - a setpoint to publish alongside the readback. ``None`` leaves the cached setpoint untouched. A bare value returned from a setter is equivalent to ``Update(readback=value, setpoint=value)`` - the device's accepted or clamped value, which is both what it will report and what was asked of it. + + The value/timestamp/severity trio follows the shape of bluesky's ``Reading`` + so the two read the same way, but shares no code with it. """ readback: DType_T timestamp: float | None = None setpoint: DType_T | None = None + severity: Severity | None = None diff --git a/src/fastcs/control_system.py b/src/fastcs/control_system.py index e8604d703..44467fe63 100644 --- a/src/fastcs/control_system.py +++ b/src/fastcs/control_system.py @@ -7,9 +7,8 @@ from IPython.terminal.embed import InteractiveShellEmbed -from fastcs.controllers import Controller, ControllerAPI +from fastcs.controllers import Controller, ControllerAPI, ControllerRunner from fastcs.logging import logger -from fastcs.methods import ScanCallback from fastcs.tracer import Tracer from fastcs.transports import Transport @@ -62,10 +61,7 @@ def __init__( self._transports = transports self._loop = loop or asyncio.get_event_loop() - self._scan_coros: list[ScanCallback] = [] - self._initial_coros: list[ScanCallback] = [] - - self._scan_tasks: set[asyncio.Task] = set() + self._runner = ControllerRunner(self._controllers, self._loop) self.controller_apis: list[ControllerAPI] = [] def run(self, interactive: bool = True): @@ -84,25 +80,6 @@ def run(self, interactive: bool = True): self._loop.add_signal_handler(signal.SIGTERM, serve.cancel) self._loop.run_until_complete(serve) - async def _run_initial_coros(self): - for coro in self._initial_coros: - await coro() - - async def _start_scan_tasks(self): - self._scan_tasks = {self._loop.create_task(coro()) for coro in self._scan_coros} - - def _stop_scan_tasks(self): - for task in self._scan_tasks: - if not task.done(): - try: - task.cancel() - except (asyncio.CancelledError, RuntimeError): - pass - except Exception as e: - raise RuntimeError("Unhandled exception in stop scan tasks") from e - - self._scan_tasks.clear() - async def serve(self, interactive: bool = True) -> None: """Serve the control system over the given transports on the current event loop @@ -118,18 +95,10 @@ async def serve(self, interactive: bool = True) -> None: interactive: Whether to create an interactive IPython shell """ - for controller in self._controllers: - await controller.initialise() - controller.post_initialise() - - self.controller_apis = [] - self._scan_coros = [] - self._initial_coros = [] - for controller in self._controllers: - api, scan_coros, initial_coros = controller.create_api_and_tasks() - self.controller_apis.append(api) - self._scan_coros.extend(scan_coros) - self._initial_coros.extend(initial_coros) + # Build the APIs before wiring transports to them: a transport + # registers its callbacks when it connects, and would miss the first + # readback if the controllers had already started. + self.controller_apis = await self._runner.setup() context = { "controllers": {_context_key(c): c for c in self._controllers}, @@ -172,10 +141,7 @@ async def block_forever(): transports=f"[{', '.join(str(t) for t in self._transports)}]", ) - for controller in self._controllers: - await controller.connect() - await self._run_initial_coros() - await self._start_scan_tasks() + await self._runner.start() try: await asyncio.gather(*coros) @@ -185,15 +151,7 @@ async def block_forever(): logger.exception("Unhandled exception in serve") finally: logger.info("Shutting down FastCS") - self._stop_scan_tasks() - for controller in self._controllers: - try: - await controller.disconnect() - except Exception: - logger.exception( - "Exception during disconnect", - controller=_context_key(controller), - ) + await self._runner.stop() async def _interactive_shell(self, context: dict[str, Any]): """Spawn interactive shell in another thread and wait for it to complete.""" @@ -222,4 +180,4 @@ async def interactive_shell( await stop_event.wait() def __del__(self): - self._stop_scan_tasks() + self._runner._cancel_tasks() # noqa: SLF001 diff --git a/src/fastcs/controllers/__init__.py b/src/fastcs/controllers/__init__.py index b982292de..e3fe4106e 100644 --- a/src/fastcs/controllers/__init__.py +++ b/src/fastcs/controllers/__init__.py @@ -2,3 +2,4 @@ from .controller import Controller as Controller from .controller_api import ControllerAPI as ControllerAPI from .controller_vector import ControllerVector as ControllerVector +from .runner import ControllerRunner as ControllerRunner diff --git a/src/fastcs/controllers/controller.py b/src/fastcs/controllers/controller.py index b03793db6..0bee8d7d8 100755 --- a/src/fastcs/controllers/controller.py +++ b/src/fastcs/controllers/controller.py @@ -28,6 +28,15 @@ def add_sub_controller(self, name: str, sub_controller: BaseController): ) return super().add_sub_controller(name, sub_controller) + @property + def connected(self) -> bool: + """Whether the controller believes it can talk to its device. + + Set by `connect`/`reconnect`, and cleared when a scan task raises. The + `ControllerRunner` reads it to decide when to reconnect. + """ + return self._connected + async def connect(self) -> None: """Hook to perform initial connection to device diff --git a/src/fastcs/controllers/runner.py b/src/fastcs/controllers/runner.py new file mode 100644 index 000000000..03de97411 --- /dev/null +++ b/src/fastcs/controllers/runner.py @@ -0,0 +1,157 @@ +import asyncio +from collections.abc import Sequence + +from fastcs.controllers.controller import Controller +from fastcs.controllers.controller_api import ControllerAPI +from fastcs.logging import logger +from fastcs.methods import ScanCallback + +RECONNECT_PERIOD = 1.0 +"""Seconds between checks for a controller that has dropped its connection""" + + +class ControllerRunner: + """Runs one or more `Controller` s, without serving them anywhere. + + This owns the whole controller lifecycle - initialising, connecting, + running the initial and periodic tasks, reconnecting after a failure, and + tidying up - and nothing about how the controllers are presented. `FastCS` + uses it and adds transports on top; an embedded caller that only wants the + controllers running can use it on its own:: + + runner = ControllerRunner(controller) + await runner.start() + ... + await runner.stop() + + Starting has two halves, because anything serving the controllers needs + their `ControllerAPI` before the first values are read: ``setup`` initialises + them and builds the APIs, and ``start`` connects and starts the tasks. + Calling ``start`` on its own does both. + + **Idempotency is the caller's responsibility.** Starting a running runner, + or stopping a stopped one, is not defined. + + Args: + controllers: The controller(s) to run. Accepts either a single + ``Controller`` or a sequence of them. + loop: Optional event loop to create the tasks in + + """ + + def __init__( + self, + controllers: Controller | Sequence[Controller], + loop: asyncio.AbstractEventLoop | None = None, + ) -> None: + if isinstance(controllers, Controller): + controllers = [controllers] + self._controllers: list[Controller] = list(controllers) + self._loop = loop + + self._controller_apis: list[ControllerAPI] = [] + self._scan_coros: list[ScanCallback] = [] + self._initial_coros: list[ScanCallback] = [] + self._tasks: set[asyncio.Task] = set() + + @property + def controllers(self) -> list[Controller]: + return self._controllers + + @property + def controller_apis(self) -> list[ControllerAPI]: + """The API of each controller. Empty until ``setup`` has run.""" + return self._controller_apis + + async def setup(self) -> list[ControllerAPI]: + """Initialise the controllers and build their APIs. + + Runs before anything connects, so that a transport can be wired to the + APIs and catch the first readback. + + Returns: + The API of each controller, in the order they were given + + """ + for controller in self._controllers: + await controller.initialise() + controller.post_initialise() + + self._controller_apis = [] + self._scan_coros = [] + self._initial_coros = [] + for controller in self._controllers: + api, scan_coros, initial_coros = controller.create_api_and_tasks() + self._controller_apis.append(api) + self._scan_coros.extend(scan_coros) + self._initial_coros.extend(initial_coros) + + return self._controller_apis + + async def start(self) -> None: + """Connect the controllers and start their tasks. + + Runs ``setup`` first if it has not already run. + """ + if not self._controller_apis: + await self.setup() + + for controller in self._controllers: + await controller.connect() + + for coro in self._initial_coros: + await coro() + + loop = self._loop or asyncio.get_event_loop() + self._tasks = {loop.create_task(coro()) for coro in self._scan_coros} + self._tasks |= { + loop.create_task(self._reconnect_loop(controller)) + for controller in self._controllers + } + + async def stop(self) -> None: + """Stop the tasks and disconnect the controllers.""" + self._cancel_tasks() + + for controller in self._controllers: + try: + await controller.disconnect() + except Exception: + logger.exception( + "Exception during disconnect", controller=controller.path + ) + + async def _reconnect_loop(self, controller: Controller) -> None: + """Bring a controller back after its scan tasks hit an error. + + A scan task that raises marks its controller disconnected and pauses + rather than dying, so something has to try to bring it back. That is the + runner's job rather than the controller's, so that every controller + reconnects the same way whether or not its author thought about it. + """ + while True: + await asyncio.sleep(RECONNECT_PERIOD) + + if controller.connected: + continue + + logger.info("Attempting to reconnect", controller=controller.path) + try: + await controller.reconnect() + except Exception: + logger.exception("Reconnect failed", controller=controller.path) + + def _cancel_tasks(self) -> None: + for task in self._tasks: + if not task.done(): + try: + task.cancel() + except (asyncio.CancelledError, RuntimeError): + pass + except Exception as e: + raise RuntimeError("Unhandled exception in stop tasks") from e + + self._tasks.clear() + + def __del__(self): + self._cancel_tasks() diff --git a/tests/test_attributes.py b/tests/test_attributes.py index e78d5c59e..595bbf443 100644 --- a/tests/test_attributes.py +++ b/tests/test_attributes.py @@ -1,10 +1,19 @@ import asyncio +import time from functools import partial import pytest from pytest_mock import MockerFixture -from fastcs.attributes import AttrR, AttrRW, AttrW, NotPolled, Polled, Update +from fastcs.attributes import ( + AttrR, + AttrRW, + AttrW, + NotPolled, + Polled, + Severity, + Update, +) from fastcs.controllers import Controller from fastcs.datatypes import Float, Int, String from fastcs.util import ONCE @@ -439,3 +448,68 @@ async def setter(value, uri=uri): await c.int_parameter.set(20) assert c.int_parameter.readback == 20 + + +@pytest.mark.asyncio +async def test_a_bare_value_is_stamped_with_the_time_it_arrived(): + attr = AttrR(Int()) + before = time.time() + + await attr.update(3) + + assert before <= attr.timestamp <= time.time() + assert attr.severity is Severity.NO_ALARM + + +@pytest.mark.asyncio +async def test_an_update_can_carry_the_time_the_value_was_obtained(): + attr = AttrR(Int()) + + await attr.update(Update(readback=3, timestamp=1234.5)) + + assert attr.timestamp == 1234.5 + + +@pytest.mark.asyncio +async def test_an_update_can_carry_a_severity(): + attr = AttrR(Int()) + + await attr.update(Update(readback=3, severity=Severity.MAJOR)) + + assert attr.severity is Severity.MAJOR + + +@pytest.mark.asyncio +async def test_an_update_with_no_timestamp_is_stamped_on_arrival(): + attr = AttrR(Int()) + before = time.time() + + await attr.update(Update(readback=3)) + + assert before <= attr.timestamp <= time.time() + assert attr.severity is Severity.NO_ALARM + + +@pytest.mark.asyncio +async def test_a_getter_can_report_a_timestamp_and_severity(): + async def get_value() -> Update[int]: + return Update(readback=7, timestamp=99.0, severity=Severity.MINOR) + + attr = AttrR(Int(), getter=get_value) + + assert await attr.poll() == 7 + assert attr.timestamp == 99.0 + assert attr.severity is Severity.MINOR + + +@pytest.mark.asyncio +async def test_a_rejected_value_leaves_the_timestamp_alone(): + """The cached value and the time it was obtained must agree.""" + attr = AttrR(Int(min=0), getter=None) + await attr.update(Update(readback=1, timestamp=10.0)) + + with pytest.raises(ValueError): + await attr.update(Update(readback=-1, timestamp=20.0)) + + assert attr.readback == 1 + assert attr.timestamp == 10.0 diff --git a/tests/test_control_system.py b/tests/test_control_system.py index 66b78b192..4572a1fa0 100644 --- a/tests/test_control_system.py +++ b/tests/test_control_system.py @@ -95,18 +95,19 @@ def __init__(self): assert controller.update_once.readback == 1 assert controller.update_never.readback == 0 - assert len(fastcs._scan_tasks) == 1 - assert len(fastcs._initial_coros) == 1 + # One periodic scan task per distinct period, plus one reconnect watcher + assert len(fastcs._runner._scan_coros) == 1 + assert len(fastcs._runner._initial_coros) == 1 @pytest.mark.asyncio async def test_controller_connect_disconnect(): class MyTestController(Controller): async def connect(self): - self.connected = True + self.connect_called = True async def disconnect(self): - self.connected = False + self.connect_called = False controller = MyTestController() @@ -117,10 +118,10 @@ async def disconnect(self): # connect is called at the start of serve await asyncio.sleep(0.1) - assert controller.connected + assert controller.connect_called task.cancel() # disconnect is called at the end of serve await asyncio.sleep(0.1) - assert not controller.connected + assert not controller.connect_called diff --git a/tests/test_controller_runner.py b/tests/test_controller_runner.py new file mode 100644 index 000000000..1b696bb65 --- /dev/null +++ b/tests/test_controller_runner.py @@ -0,0 +1,180 @@ +import asyncio + +import pytest + +from fastcs.attributes import AttrR +from fastcs.controllers import Controller, ControllerRunner +from fastcs.controllers.runner import RECONNECT_PERIOD +from fastcs.datatypes import Int +from fastcs.methods import scan +from fastcs.util import ONCE + + +class LifecycleController(Controller): + """Records every lifecycle hook the runner is supposed to call.""" + + def __init__(self): + super().__init__() + self.events: list[str] = [] + self.count = AttrR(Int()) + + async def initialise(self): + self.events.append("initialise") + + def post_initialise(self): + self.events.append("post_initialise") + + async def connect(self): + self.events.append("connect") + await super().connect() + + async def disconnect(self): + self.events.append("disconnect") + + @scan(ONCE) + async def read_once(self): + self.events.append("initial") + await self.count.update(self.count.readback + 1) + + +@pytest.mark.asyncio +async def test_the_runner_drives_the_whole_lifecycle(): + controller = LifecycleController() + runner = ControllerRunner(controller) + + await runner.start() + try: + assert controller.events == [ + "initialise", + "post_initialise", + "connect", + "initial", + ] + assert controller.count.readback == 1 + finally: + await runner.stop() + + assert controller.events[-1] == "disconnect" + + +@pytest.mark.asyncio +async def test_setup_builds_the_apis_before_anything_connects(): + """A transport is wired to the APIs between setup and start.""" + controller = LifecycleController() + runner = ControllerRunner(controller) + + apis = await runner.setup() + + assert [api.path for api in apis] == [[]] + assert "count" in apis[0].attributes + assert controller.events == ["initialise", "post_initialise"] + assert runner.controller_apis == apis + + +@pytest.mark.asyncio +async def test_start_sets_up_when_setup_has_not_run(): + runner = ControllerRunner(LifecycleController()) + + await runner.start() + try: + assert len(runner.controller_apis) == 1 + finally: + await runner.stop() + + +@pytest.mark.asyncio +async def test_a_runner_takes_several_controllers(): + controllers = [LifecycleController(), LifecycleController()] + runner = ControllerRunner(controllers) + + await runner.start() + try: + assert len(runner.controller_apis) == 2 + assert all(controller.count.readback == 1 for controller in controllers) + finally: + await runner.stop() + + +@pytest.mark.asyncio +async def test_stop_reports_a_failing_disconnect_without_raising(): + class UndisconnectableController(LifecycleController): + async def disconnect(self): + raise RuntimeError("no") + + runner = ControllerRunner(UndisconnectableController()) + await runner.start() + + await runner.stop() + + +@pytest.mark.asyncio +async def test_the_runner_reconnects_a_controller_that_dropped_out(monkeypatch): + """Nothing else calls reconnect, so a paused controller would stay paused.""" + monkeypatch.setattr("fastcs.controllers.runner.RECONNECT_PERIOD", 0.01) + + class DroppingController(LifecycleController): + reconnects = 0 + + async def reconnect(self): + type(self).reconnects += 1 + await super().reconnect() + + controller = DroppingController() + runner = ControllerRunner(controller) + await runner.start() + try: + assert controller.connected + + # What a scan task does when its callback raises + controller._connected = False + + await asyncio.sleep(0.05) + + assert DroppingController.reconnects >= 1 + assert controller.connected + finally: + await runner.stop() + + +@pytest.mark.asyncio +async def test_a_failing_reconnect_does_not_stop_the_runner(monkeypatch): + monkeypatch.setattr("fastcs.controllers.runner.RECONNECT_PERIOD", 0.01) + + class UnreconnectableController(LifecycleController): + attempts = 0 + + async def reconnect(self): + type(self).attempts += 1 + raise RuntimeError("still down") + + controller = UnreconnectableController() + runner = ControllerRunner(controller) + await runner.start() + try: + controller._connected = False + await asyncio.sleep(0.05) + + # It keeps trying rather than dying on the first failure + assert UnreconnectableController.attempts > 1 + assert not controller.connected + finally: + await runner.stop() + + +@pytest.mark.asyncio +async def test_stop_cancels_the_tasks(): + controller = LifecycleController() + runner = ControllerRunner(controller) + await runner.start() + tasks = set(runner._tasks) + assert tasks + + await runner.stop() + await asyncio.sleep(0) + + assert all(task.cancelled() or task.done() for task in tasks) + assert not runner._tasks + + +def test_reconnect_period_is_a_second_by_default(): + assert RECONNECT_PERIOD == 1.0 diff --git a/tests/test_multi_controller.py b/tests/test_multi_controller.py index bde9b2f38..2906c96f8 100644 --- a/tests/test_multi_controller.py +++ b/tests/test_multi_controller.py @@ -307,7 +307,7 @@ class _LifecycleController(Controller): def __init__(self): super().__init__() - self.connected = False + self.connect_called = False self.initialised = False self.post_initialised = False @@ -318,10 +318,10 @@ def post_initialise(self): self.post_initialised = True async def connect(self): - self.connected = True + self.connect_called = True async def disconnect(self): - self.connected = False + self.connect_called = False class _OtherLifecycleController(_LifecycleController): @@ -350,7 +350,7 @@ async def test_fastcs_serves_two_controllers_end_to_end(mocker: MockerFixture): for controller in (a, b): assert controller.initialised assert controller.post_initialised - assert controller.connected + assert controller.connect_called with TestClient(transport._server._app) as client: assert client.get("/alpha/foo").status_code == 200 @@ -370,4 +370,4 @@ async def test_fastcs_serves_two_controllers_end_to_end(mocker: MockerFixture): pass for controller in (a, b): - assert not controller.connected + assert not controller.connect_called From 0cb710db2f10919fa5cf04ae3231e2bf86c96ae0 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 14 Aug 2026 23:14:36 +0000 Subject: [PATCH 2/2] refactor(#395): drop dead code from the new runner The `controllers` property has no caller, and `Task.cancel` does not raise - the guards `FastCS._stop_scan_tasks` wrapped it in never fired, so moving them across only moved unreachable code. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017zpFjvbhUisfVUq3NH8M1G --- src/fastcs/controllers/runner.py | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/src/fastcs/controllers/runner.py b/src/fastcs/controllers/runner.py index 03de97411..ead653e9e 100644 --- a/src/fastcs/controllers/runner.py +++ b/src/fastcs/controllers/runner.py @@ -54,10 +54,6 @@ def __init__( self._initial_coros: list[ScanCallback] = [] self._tasks: set[asyncio.Task] = set() - @property - def controllers(self) -> list[Controller]: - return self._controllers - @property def controller_apis(self) -> list[ControllerAPI]: """The API of each controller. Empty until ``setup`` has run.""" @@ -142,14 +138,12 @@ async def _reconnect_loop(self, controller: Controller) -> None: logger.exception("Reconnect failed", controller=controller.path) def _cancel_tasks(self) -> None: + # ``Task.cancel`` does not raise - it returns whether the task was + # cancellable - so the guards the old FastCS._stop_scan_tasks wrapped + # this in never fired. for task in self._tasks: if not task.done(): - try: - task.cancel() - except (asyncio.CancelledError, RuntimeError): - pass - except Exception as e: - raise RuntimeError("Unhandled exception in stop tasks") from e + task.cancel() self._tasks.clear()