controllers: ControllerRunner, plus native timestamps and severity on attributes

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.

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

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,15 +154,24 @@ 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
 
         Raises:
             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 -= {

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"""

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

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

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

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