The Tango arm: a second control substrate, and a typed address at the registry seam

apps/api/pyproject.toml

@@ -104,6 +104,21 @@ dependencies = [
 bo = [
     "botorch>=0.16,<1",
 ]
+# tango (Actuate-axis Tango arm): the optional dependency group for
+# `TangoControlPort` at `cora/operation/adapters/`, the ControlPort adapter
+# for Tango-floor deployments (ESRF BLISS / Beacon, MAX IV Tango / Sardana).
+# PyTango binds the C++ Tango core plus omniORB: a heavy native stack that
+# only Tango deployments need, so it is kept OUT of the base install (EPICS
+# deployments go through aioca / p4p). The adapter imports `tango.asyncio`
+# lazily and probes importability at construction (`_optional_tango.py`), so
+# selecting the `tango` substrate without this group installed fails as a
+# `ValueError` at wiring time, never as an uncaught `ImportError` mid-loop.
+# The port is built once in the FastAPI lifespan, so a missing group fails
+# startup rather than a request. LGPL-3.0 (PyTango binding) over the C++ Tango core.
+# Pin <11 to flag a future major SDK shift.
+tango = [
+    "pytango>=10,<11",
+]
 
 [dependency-groups]
 dev = [

apps/api/src/cora/infrastructure/control_port_route.py

@@ -32,12 +32,13 @@
 
 from pydantic import BaseModel, Field
 
-Substrate = Literal["in_memory", "epics_ca", "epics_pva"]
+Substrate = Literal["in_memory", "epics_ca", "epics_pva", "tango"]
 """The closed set of ControlPort substrates a deployment can select.
 
 `in_memory` is the test + opt-out default; `epics_ca` + `epics_pva`
-are the production EPICS adapters. New substrates land here as
-additive literal values plus a matching arm in the BC-side
+are the production EPICS adapters; `tango` is the PyTango adapter for
+Tango-floor deployments (ESRF BLISS, MAX IV). New substrates land here
+as additive literal values plus a matching arm in the BC-side
 `build_control_port` factory's `_build_substrate` switch.
 """
 

apps/api/src/cora/operation/adapters/__init__.py

@@ -3,8 +3,9 @@
 `InMemoryControlPort` is the unit-tier `ControlPort` adapter per
 [[project_control_port_generalization_research]]. Substrate adapters
 (`CaprotoControlPort` test-tier; `EpicsCaControlPort`,
-`EpicsPvaControlPort` production-tier) cover the EPICS family.
-`ControlPortRegistry` routes addresses to the right substrate by
-longest-prefix match so the executor sees a single `ControlPort`
-regardless of how many substrates a deployment runs.
+`EpicsPvaControlPort` production-tier) cover the EPICS family, and
+`TangoControlPort` (PyTango, optional `tango` extra) covers Tango-floor
+deployments. `ControlPortRegistry` routes addresses to the right
+substrate by longest-prefix match so the executor sees a single
+`ControlPort` regardless of how many substrates a deployment runs.
 """

apps/api/src/cora/operation/adapters/_optional_tango.py

@@ -0,0 +1,32 @@
+"""Construction-time probe for the optional `tango` dependency group.
+
+The `TangoControlPort` adapter needs PyTango (`tango`), which binds the C++
+Tango core plus omniORB: a heavy native stack that only Tango-floor
+deployments (ESRF BLISS, MAX IV Tango/Sardana) need. To keep the base install
+lean and the `cora` import cheap on EPICS-only deployments, the adapter does
+not import `tango` at module load; it probes importability in its `__init__`
+via the helper here.
+
+The probe raises `ValueError` (not `ImportError`) on a missing dependency so
+the gap surfaces where the ControlPort is materialised (`build_control_port`,
+called once at wiring) rather than as an uncaught `ImportError` deep in the
+conduct loop, after a procedure has already started. Selecting the substrate
+without the group installed therefore fails startup, loudly and before any
+run begins. Mirrors the `bo` group's
+`_optional_torch.py` posture.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+
+_INSTALL_HINT = "install the optional 'tango' dependency group (e.g. `uv sync --extra tango`)"
+
+
+def require_tango(substrate: str) -> None:
+    """Raise ValueError if `tango` (PyTango) is not importable, naming the substrate."""
+    if importlib.util.find_spec("tango") is None:
+        raise ValueError(f"the {substrate!r} substrate needs PyTango; {_INSTALL_HINT}")
+
+
+__all__ = ["require_tango"]

apps/api/src/cora/operation/adapters/caproto_control_port.py

@@ -109,6 +109,8 @@
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator
 
+    from cora.operation.ports.control_address import EpicsPvAddress
+
 
 _log = get_logger(__name__)
 _DISPATCH_EVENT = "controlport.dispatch"
@@ -239,57 +241,59 @@ async def _connected_pv(self, address: str) -> Any:
             raise ControlNotConnectedError(address) from exc
         return pv
 
-    async def read(self, address: str) -> Measurement:
-        pv = await self._connected_pv(address)
+    async def read(self, address: EpicsPvAddress) -> Measurement:
+        name = address.pv
+        pv = await self._connected_pv(name)
         try:
             response = await pv.read(data_type="time", timeout=self._default_timeout_s)
         except CaprotoTimeoutError as exc:
-            raise ControlTimeoutError(address, self._default_timeout_s) from exc
+            raise ControlTimeoutError(name, self._default_timeout_s) from exc
         return _to_reading(response)
 
     async def write(
         self,
-        address: str,
+        address: EpicsPvAddress,
         value: int | float | bool | str | tuple[Any, ...],
         *,
         wait: bool = True,
         timeout_s: float = 30.0,
     ) -> None:
+        name = address.pv
         correlation_id = get_dispatch_correlation_id()
         _log.info(
             _DISPATCH_EVENT,
-            address=address,
+            address=name,
             operation="write",
             correlation_id=str(correlation_id) if correlation_id is not None else None,
             status="started",
         )
         try:
-            pv = await self._connected_pv(address)
+            pv = await self._connected_pv(name)
             await pv.write(value, wait=wait, timeout=timeout_s)
         except CaprotoTimeoutError as exc:
             _log.info(
                 _DISPATCH_FAILED_EVENT,
-                address=address,
+                address=name,
                 operation="write",
                 correlation_id=str(correlation_id) if correlation_id is not None else None,
                 status="failed",
                 error_class=ControlTimeoutError.__name__,
             )
-            raise ControlTimeoutError(address, timeout_s) from exc
+            raise ControlTimeoutError(name, timeout_s) from exc
         except ErrorResponseReceived as exc:
             _log.info(
                 _DISPATCH_FAILED_EVENT,
-                address=address,
+                address=name,
                 operation="write",
                 correlation_id=str(correlation_id) if correlation_id is not None else None,
                 status="failed",
                 error_class=ControlWriteRejectedError.__name__,
             )
-            raise ControlWriteRejectedError(address, str(exc)) from exc
+            raise ControlWriteRejectedError(name, str(exc)) from exc
         except ControlNotConnectedError:
             _log.info(
                 _DISPATCH_FAILED_EVENT,
-                address=address,
+                address=name,
                 operation="write",
                 correlation_id=str(correlation_id) if correlation_id is not None else None,
                 status="failed",
@@ -298,13 +302,13 @@ async def write(
             raise
         _log.info(
             _DISPATCH_COMPLETED_EVENT,
-            address=address,
+            address=name,
             operation="write",
             correlation_id=str(correlation_id) if correlation_id is not None else None,
             status="completed",
         )
 
-    def subscribe(self, address: str) -> AsyncGenerator[Measurement]:
+    def subscribe(self, address: EpicsPvAddress) -> AsyncGenerator[Measurement]:
         """Return type narrows the Protocol's `AsyncIterator` to `AsyncGenerator`.
 
         Covariant return lets tests close subscriptions via the
@@ -314,7 +318,7 @@ def subscribe(self, address: str) -> AsyncGenerator[Measurement]:
         + connect + `pv.subscribe`) runs on the generator's first
         `__anext__`.
         """
-        return self._drain(address)
+        return self._drain(address.pv)
 
     async def _drain(self, address: str) -> AsyncGenerator[Measurement]:
         pv = await self._connected_pv(address)

apps/api/src/cora/operation/adapters/control_port_config.py

@@ -22,10 +22,14 @@
     reachable but no real substrate is exercised).
   - `epics_ca`: production CA via aioca (`EpicsCaControlPort`).
   - `epics_pva`: production PVA via p4p (`EpicsPvaControlPort`).
+  - `tango`: Tango device attributes via PyTango (`TangoControlPort`).
+    Ships under the optional `tango` dependency group; selecting it
+    without the extra installed raises `ValueError` at construction
+    (probed via `_optional_tango.require_tango`).
 
-Future substrates (`tango`, `opc_ua`) land as additive code edits
-in `cora.infrastructure.control_port_route` (literal) plus a new
-arm in `_build_substrate` here.
+Future substrates (`opc_ua`) land as additive code edits in
+`cora.infrastructure.control_port_route` (literal) plus a new arm in
+`_build_substrate` here.
 
 ## Lifecycle
 
@@ -36,13 +40,15 @@
 """
 
 from collections.abc import Sequence
+from typing import Any
 
 from cora.infrastructure.control_port_route import ControlPortRoute, Substrate
 from cora.operation.adapters.control_port_registry import ControlPortRegistry
 from cora.operation.adapters.epics_ca_control_port import EpicsCaControlPort
 from cora.operation.adapters.epics_pva_control_port import EpicsPvaControlPort
 from cora.operation.adapters.in_memory_control_port import InMemoryControlPort
-from cora.operation.ports.control_port import ControlPort
+from cora.operation.adapters.tango_control_port import TangoControlPort
+from cora.operation.ports.control_port import ControlPort, SubstrateControlPort
 
 
 def build_control_port(routes: Sequence[ControlPortRoute]) -> ControlPort:
@@ -51,33 +57,45 @@ def build_control_port(routes: Sequence[ControlPortRoute]) -> ControlPort:
     Empty routes returns a single `InMemoryControlPort` (legacy
     default + test convenience). Non-empty routes returns a
     `ControlPortRegistry` populated with the configured substrate
-    adapters per prefix.
+    adapters per prefix. `in_memory` routes go through
+    `register_control_port` (the registry wraps the `str`-surfaced adapter);
+    the real substrate adapters register directly on the typed surface.
     """
     if not routes:
         return InMemoryControlPort()
     registry = ControlPortRegistry()
     for route in routes:
-        registry.register(
-            route.prefix,
-            _build_substrate(route.substrate),
-            is_simulated=route.is_simulated,
-        )
+        if route.substrate == "in_memory":
+            registry.register_control_port(
+                route.prefix, InMemoryControlPort(), is_simulated=route.is_simulated
+            )
+        else:
+            registry.register_substrate_port(
+                route.prefix,
+                _build_substrate(route.substrate),
+                route.substrate,
+                is_simulated=route.is_simulated,
+            )
     return registry
 
 
-def _build_substrate(substrate: Substrate) -> ControlPort:
-    """Construct the per-substrate adapter with deployment defaults.
+def _build_substrate(substrate: Substrate) -> SubstrateControlPort[Any]:
+    """Construct the per-substrate typed adapter with deployment defaults.
+
+    Handles the typed-address substrate adapters only; `in_memory` is
+    registered through `ControlPortRegistry.register_control_port` in
+    `build_control_port` because it is `str`-surfaced.
 
     Per-adapter constructor kwargs (timeouts, etc.) ride on the
     adapter defaults today; a future iteration may widen
     `ControlPortRoute` with optional per-route overrides
     (`timeout_s`, etc.) when a real deployment surfaces the need.
     """
-    if substrate == "in_memory":
-        return InMemoryControlPort()
     if substrate == "epics_ca":
         return EpicsCaControlPort()
-    return EpicsPvaControlPort()
+    if substrate == "epics_pva":
+        return EpicsPvaControlPort()
+    return TangoControlPort()
 
 
 __all__ = ["build_control_port"]