diff --git a/.env.example b/.env.example index e458b598841..4922ddb9dd7 100644 --- a/.env.example +++ b/.env.example @@ -85,19 +85,31 @@ LOG_LEVEL=INFO # LONGEST prefix, and each prefix must be distinct (a duplicate is refused # at boot rather than silently discarding the earlier route's flags). # -# Per-route flags, both declared facts and never inferred from substrate: -# is_simulated this route drives a simulator, even over real Channel -# Access (a soft IOC speaks real CA). Feeds the Dataset -# provenance gate that blocks promoting simulator data. -# read_only CORA may read and subscribe here but never write. -# Per-route expressiveness INSIDE a writable deployment -# ("drive the stage, never the shutter"). It defaults to -# false, so it is NOT how you make a deployment -# observe-only: use CONTROL_WRITES_ENABLED=false for that. +# Per-route declarations, all declared facts and never inferred from substrate: +# is_simulated this route drives a simulator, even over real Channel +# Access (a soft IOC speaks real CA). Feeds the Dataset +# provenance gate that blocks promoting simulator data. +# read_only CORA may read and subscribe here but never write. +# Per-route expressiveness INSIDE a writable deployment +# ("drive the stage, never the shutter"). It defaults to +# false, so it is NOT how you make a deployment +# observe-only: use CONTROL_WRITES_ENABLED=false for that. +# text_addresses epics_ca only. Addresses whose EPICS DBR_CHAR waveform +# carries a NUL-terminated string (e.g. tomoscan's +# ScanStatus, FileName) rather than raw bytes (e.g. an +# NTNDArray image). EPICS gives both the same wire type, +# so undeclared addresses read as Measurement(kind= +# "Array", value=); declared ones decode +# to Measurement(kind="Scalar", value=). A no-op on +# other substrates and on addresses that never resolve +# to DBR_CHAR. # # CONTROL_PORT_ROUTES='[ # {"prefix":"2bma:cam1:image","substrate":"epics_pva"}, # {"prefix":"2bma:shutter:","substrate":"epics_ca","read_only":true}, +# {"prefix":"2bmb:TomoScan:","substrate":"epics_ca","read_only":true, +# "text_addresses":["2bmb:TomoScan:ScanStatus","2bmb:TomoScan:FileName", +# "2bmb:TomoScan:FilePath","2bmb:TomoScan:FullFileName"]}, # {"prefix":"2bma:","substrate":"epics_ca"} # ]' diff --git a/apps/api/src/cora/infrastructure/control_port_route.py b/apps/api/src/cora/infrastructure/control_port_route.py index c22b2ddf56e..229a1ce6b20 100644 --- a/apps/api/src/cora/infrastructure/control_port_route.py +++ b/apps/api/src/cora/infrastructure/control_port_route.py @@ -70,6 +70,18 @@ class the factory will construct for this route. `is_simulated` `Settings.control_writes_enabled`, which cannot be partially applied. Reach for the switch to make a deployment observe-only; reach for this field only to carve a hole in a writable one. + + `text_addresses` declares which addresses on this route carry text + in an EPICS DBR_CHAR waveform. EPICS gives a char waveform holding + a NUL-terminated string (tomoscan's `ScanStatus`, `FileName`, ...) + and one holding raw bytes (an NTNDArray image) the same wire type, + so the adapter cannot tell them apart and must be told. Meaningful + only for `epics_ca` (`EpicsCaControlPort` is the sole reader of it; + PVA's NTScalar and Tango's DevString already carry this distinction + on the wire); a route on another substrate that sets it is inert + rather than rejected, since the address list still describes a + true fact about the deployment, just one this substrate's adapter + has no ambiguity to resolve. """ prefix: str = Field(..., min_length=1) @@ -90,6 +102,16 @@ class the factory will construct for this route. `is_simulated` "deployment observe-only, set CONTROL_WRITES_ENABLED=false instead." ), ) + text_addresses: tuple[str, ...] = Field( + default=(), + description=( + "Addresses on this route whose EPICS DBR_CHAR waveform carries a " + "NUL-terminated string rather than raw bytes. Declared per deployment, " + "never inferred: EPICS gives a string-bearing char waveform and a " + "byte-bearing one (e.g. NTNDArray image data) the same wire type. " + "Applies to epics_ca routes only; a no-op on other substrates." + ), + ) model_config = {"extra": "forbid"} diff --git a/apps/api/src/cora/operation/adapters/control_port_config.py b/apps/api/src/cora/operation/adapters/control_port_config.py index 6a97d833326..daf72070a16 100644 --- a/apps/api/src/cora/operation/adapters/control_port_config.py +++ b/apps/api/src/cora/operation/adapters/control_port_config.py @@ -62,7 +62,7 @@ from collections.abc import Sequence from typing import Any, Literal -from cora.infrastructure.control_port_route import ControlPortRoute, Substrate +from cora.infrastructure.control_port_route import ControlPortRoute 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 @@ -130,7 +130,7 @@ def build_control_port(routes: Sequence[ControlPortRoute], *, writes_enabled: bo registry.register_substrate_port( route.prefix, _guarded_substrate( - _build_substrate(route.substrate), + _build_substrate(route), read_only=read_only, scope=scope, prefix=prefix, @@ -190,21 +190,26 @@ def _guarded_substrate( return ReadOnlySubstratePort(port, scope=scope, prefix=prefix) -def _build_substrate(substrate: Substrate) -> SubstrateControlPort[Any]: +def _build_substrate(route: ControlPortRoute) -> 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. + `route.text_addresses` is CA-specific (see `ControlPortRoute` + docstring) and reaches only `EpicsCaControlPort`; PVA and Tango + routes carry the field but nothing here reads it for them, which + is the field's declared inert-elsewhere contract, not an omission. + + Remaining per-adapter constructor kwargs (timeouts, etc.) ride on + the adapter defaults today; a future iteration may widen + `ControlPortRoute` with further per-route overrides when a real + deployment surfaces the need. """ - if substrate == "epics_ca": - return EpicsCaControlPort() - if substrate == "epics_pva": + if route.substrate == "epics_ca": + return EpicsCaControlPort(text_addresses=route.text_addresses) + if route.substrate == "epics_pva": return EpicsPvaControlPort() return TangoControlPort() diff --git a/apps/api/src/cora/operation/adapters/epics_ca_control_port.py b/apps/api/src/cora/operation/adapters/epics_ca_control_port.py index 7f2e808e3df..2f28ebb41d1 100644 --- a/apps/api/src/cora/operation/adapters/epics_ca_control_port.py +++ b/apps/api/src/cora/operation/adapters/epics_ca_control_port.py @@ -62,6 +62,28 @@ `.enums` for label resolution, then caches the labels per-address so subsequent reads stay on the cheap FORMAT_TIME path. +## DBR_CHAR waveforms: bytes, or a string wearing bytes' clothes + +A DBR_CHAR waveform (`aioca.DBR_CHAR`, `element_count > 1`) reaches +`_kind_for` indistinguishably from any other array type and is +unpacked as a tuple of small integers: correct for a byte-array +payload (an NTNDArray image plugin's raw data), wrong for what many +IOCs use the same record shape for, a NUL-terminated ASCII string +too long for a `stringout`'s 40-character limit (tomoscan's +`ScanStatus`, `FileName`, `FilePath`, `FullFileName`). EPICS assigns +both the same wire type, so nothing in the reading itself says which +this is; `EpicsCaControlPort` is told via `text_addresses`, a +deployment-declared set of PVs to decode as text rather than an +integer tuple. Declaring an address that never resolves to DBR_CHAR +is inert, not an error: the declaration describes what the PV +carries, and a route that also lists the wrong PVs merely finds +nothing to apply it to. + +A decoded string is cut at the first NUL: a fixed-size waveform +written with a shorter string than its NELM leaves trailing NULs +(or whatever was in the buffer previously) past the terminator, and +everything from the first NUL on is padding, not payload. + ## Error mapping aioca raises ONE exception class, `CANothing(name, errorcode)`, @@ -110,6 +132,7 @@ from typing import TYPE_CHECKING, Any from aioca import ( + DBR_CHAR, DBR_ENUM, FORMAT_CTRL, FORMAT_TIME, @@ -134,7 +157,7 @@ ) if TYPE_CHECKING: - from collections.abc import AsyncGenerator + from collections.abc import AsyncGenerator, Iterable from cora.operation.ports.control_address import EpicsPvAddress @@ -280,10 +303,48 @@ def _produced_at_for(timestamp: float) -> datetime | None: return datetime.fromtimestamp(timestamp, tz=UTC) -def _to_reading(augmented: Any, enum_labels: tuple[str, ...] | None) -> Measurement: - """Translate an aioca `AugmentedValue` (FORMAT_TIME) to `Measurement`.""" - kind = _kind_for(augmented.datatype, augmented.element_count) - value = _unpack_value(augmented, kind, enum_labels) +def _decode_char_waveform(augmented: Any) -> str: + """Decode a DBR_CHAR waveform holding a NUL-terminated string. + + Only called once the caller has confirmed both that the reading is + actually DBR_CHAR and that the deployment declared this address as + text (see module docstring, "DBR_CHAR waveforms"). `.tolist()` + covers the numpy-array case aioca returns for a populated waveform; + the plain `list()` fallback covers a zero-length reading, which + aioca can hand back as an empty non-numpy sequence. + """ + raw = augmented.tolist() if hasattr(augmented, "tolist") else list(augmented) + return bytes(raw).split(b"\x00", 1)[0].decode("utf-8", errors="replace") + + +def _to_reading( + augmented: Any, + enum_labels: tuple[str, ...] | None, + *, + as_text: bool = False, +) -> Measurement: + """Translate an aioca `AugmentedValue` (FORMAT_TIME) to `Measurement`. + + `as_text` is the caller's `text_addresses` declaration for this + specific address, not a property of the reading. It only takes + effect when the reading is actually a DBR_CHAR *waveform* + (`element_count > 1`, matching `_kind_for`'s own Array threshold + and the module docstring's "DBR_CHAR waveforms" scope): a + declaration cannot manufacture a wire type the substrate did not + send, so a stale or misdirected declaration is inert rather than + corrupting. The `element_count > 1` half of that guard matters on + its own: aioca collapses a length-1 DBR_CHAR waveform to its + scalar `ca_int` type, which has neither `.tolist()` nor + `__iter__`, so decoding it as a waveform would raise `TypeError` + instead of falling through inert. + """ + is_text = as_text and int(augmented.datatype) == DBR_CHAR and augmented.element_count > 1 + kind: MeasurementKind = ( + "Scalar" if is_text else _kind_for(augmented.datatype, augmented.element_count) + ) + value = ( + _decode_char_waveform(augmented) if is_text else _unpack_value(augmented, kind, enum_labels) + ) severity = int(getattr(augmented, "severity", 0)) status = int(getattr(augmented, "status", 0)) timestamp = float(getattr(augmented, "timestamp", 0.0)) @@ -315,8 +376,14 @@ class EpicsCaControlPort: See module docstring for the connection model + ACL table. """ - def __init__(self, *, default_timeout_s: float = _DEFAULT_TIMEOUT_S) -> None: + def __init__( + self, + *, + default_timeout_s: float = _DEFAULT_TIMEOUT_S, + text_addresses: Iterable[str] = (), + ) -> None: self._default_timeout_s = default_timeout_s + self._text_addresses = frozenset(text_addresses) self._enum_labels: dict[str, tuple[str, ...]] = {} self._closed = False @@ -356,7 +423,7 @@ async def read(self, address: EpicsPvAddress) -> Measurement: labels: tuple[str, ...] | None = None if _kind_for(augmented.datatype, augmented.element_count) == "Categorical": labels = await self._resolve_enum_labels(pv) - return _to_reading(augmented, labels) + return _to_reading(augmented, labels, as_text=pv in self._text_addresses) async def write( self, @@ -426,6 +493,7 @@ async def _drain(self, address: str) -> AsyncGenerator[Measurement]: format=FORMAT_TIME, notify_disconnect=True, ) + as_text = address in self._text_addresses try: labels: tuple[str, ...] | None = None while True: @@ -437,7 +505,7 @@ async def _drain(self, address: str) -> AsyncGenerator[Measurement]: and _kind_for(update.datatype, update.element_count) == "Categorical" ): labels = await self._resolve_enum_labels(address) - yield _to_reading(update, labels) + yield _to_reading(update, labels, as_text=as_text) finally: with contextlib.suppress(Exception): sub.close() diff --git a/apps/api/tests/integration/_softioc.py b/apps/api/tests/integration/_softioc.py index 91bd4c8baa6..654870c8318 100644 --- a/apps/api/tests/integration/_softioc.py +++ b/apps/api/tests/integration/_softioc.py @@ -25,6 +25,16 @@ - `long_value` (DBR_LONG, `longout`) -> `Measurement(kind="Scalar")` - `string_value` (DBR_STRING, `stringout`) -> `Measurement(kind="Scalar")` - `waveform` (DBR_DOUBLE x 4, `waveform`) -> `Measurement(kind="Array")` + - `text_waveform` (DBR_CHAR x 256, `waveform`) -> `Measurement(kind="Array")` + by default; `Measurement(kind="Scalar", value=)` when the caller + declares it via `EpicsCaControlPort(text_addresses={...})`, exercising + the tomoscan `ScanStatus`-shaped ambiguity (see that adapter's + "DBR_CHAR waveforms" module-docstring section) + - `text_waveform_nelm1` (DBR_CHAR x 1, `waveform`) -> aioca collapses a + length-1 char waveform to its scalar `ca_int` type; declaring it via + `text_addresses` must stay inert (no `.tolist()` / not iterable, so + the naive "any DBR_CHAR is a waveform" version of the decode crashed + here before the `element_count > 1` guard was added) - `enum_value` (DBR_ENUM, `mbbo` with 3 strings) -> `Measurement(kind="Categorical")` - `major_alarm_value` (`ao` with HIHI threshold tripped, HHSV=MAJOR) -> `Measurement(quality="Uncertain")` @@ -136,6 +146,22 @@ field(PINI, "YES") } +record(waveform, "$(P)text_waveform") { + field(DESC, "DBR_CHAR waveform, NUL-terminated string") + field(DTYP, "Soft Channel") + field(NELM, "256") + field(FTVL, "UCHAR") + field(PINI, "YES") +} + +record(waveform, "$(P)text_waveform_nelm1") { + field(DESC, "DBR_CHAR waveform, NELM=1") + field(DTYP, "Soft Channel") + field(NELM, "1") + field(FTVL, "UCHAR") + field(PINI, "YES") +} + record(mbbo, "$(P)enum_value") { field(DESC, "DBR_ENUM with closed label set") field(DTYP, "Soft Channel") diff --git a/apps/api/tests/integration/test_epics_ca_control_port.py b/apps/api/tests/integration/test_epics_ca_control_port.py index 5f49085141c..4414c843fb8 100644 --- a/apps/api/tests/integration/test_epics_ca_control_port.py +++ b/apps/api/tests/integration/test_epics_ca_control_port.py @@ -22,6 +22,9 @@ - Protocol conformance via `isinstance` (no IOC) - Every `MeasurementKind` branch (Scalar / Array / Categorical) + - DBR_CHAR waveform: undeclared (Array-of-bytes), declared via + `text_addresses` (decoded Scalar str, both read + subscribe), + NUL-padding trim, and the inert case (declaring a non-char PV) - `Quality=Bad` via `bad_quality_value` (HIHI threshold tripped) - caput-callback round-trip on scalar + long - subscribe initial-value + post-write fan-out @@ -121,6 +124,127 @@ async def test_read_waveform_returns_array_as_tuple(softioc: str) -> None: await port.aclose() +@pytest.mark.integration +async def test_read_char_waveform_without_declaration_returns_array_of_bytes( + softioc: str, +) -> None: + """DBR_CHAR waveform is Array-of-int by default: the ambiguous case, undeclared. + + Same shape a byte-array payload (an NTNDArray image) would take; + this is the reading a deployment gets before it tells the adapter + which addresses actually carry text via `text_addresses`. + """ + port = EpicsCaControlPort() + try: + await port.write( + EpicsPvAddress(f"{softioc}text_waveform"), + tuple(b"hello"), + wait=True, + ) + reading = await port.read(EpicsPvAddress(f"{softioc}text_waveform")) + assert reading.kind == "Array" + assert reading.value == (104, 101, 108, 108, 111) + finally: + await port.aclose() + + +@pytest.mark.integration +async def test_read_char_waveform_declared_as_text_returns_decoded_string( + softioc: str, +) -> None: + """A declared `text_addresses` entry decodes the same wire reading as text. + + Models tomoscan's `ScanStatus` / `FileName` / `FullFileName`: a + DBR_CHAR waveform an operator knows carries a NUL-terminated + string, told to the adapter because EPICS gives it no way to tell + that apart from a byte-array payload on its own. + """ + pv = f"{softioc}text_waveform" + port = EpicsCaControlPort(text_addresses={pv}) + try: + await port.write(EpicsPvAddress(pv), tuple(b"fdt file transfer complete"), wait=True) + reading = await port.read(EpicsPvAddress(pv)) + assert reading.kind == "Scalar" + assert reading.value == "fdt file transfer complete" + finally: + await port.aclose() + + +@pytest.mark.integration +async def test_read_char_waveform_declared_as_text_trims_trailing_nul_padding( + softioc: str, +) -> None: + """A shorter message than NELM leaves trailing padding; decode stops at the first NUL.""" + pv = f"{softioc}text_waveform" + port = EpicsCaControlPort(text_addresses={pv}) + try: + padded = tuple(b"hi") + (0,) * 254 + await port.write(EpicsPvAddress(pv), padded, wait=True) + reading = await port.read(EpicsPvAddress(pv)) + assert reading.value == "hi" + finally: + await port.aclose() + + +@pytest.mark.integration +async def test_declaring_a_non_char_address_as_text_is_inert(softioc: str) -> None: + """`text_addresses` naming a non-DBR_CHAR PV changes nothing: it cannot manufacture a type.""" + pv = f"{softioc}waveform" + port = EpicsCaControlPort(text_addresses={pv}) + try: + await port.write(EpicsPvAddress(pv), (1.0, 2.0, 3.0, 4.0), wait=True) + reading = await port.read(EpicsPvAddress(pv)) + assert reading.kind == "Array" + assert reading.value == (1.0, 2.0, 3.0, 4.0) + finally: + await port.aclose() + + +@pytest.mark.integration +async def test_declaring_a_length_one_char_waveform_as_text_does_not_raise( + softioc: str, +) -> None: + """A length-1 DBR_CHAR waveform must stay inert under `text_addresses`, not crash. + + aioca collapses `element_count == 1` to its scalar `ca_int` type, + which is neither iterable nor has `.tolist()`. `_to_reading` gates + `as_text` on `element_count > 1` for exactly this reason: without + it, decoding this reading as a waveform raises `TypeError` instead + of falling through to the ordinary (already-Scalar) path. + """ + pv = f"{softioc}text_waveform_nelm1" + port = EpicsCaControlPort(text_addresses={pv}) + try: + await port.write(EpicsPvAddress(pv), (ord("Q"),), wait=True) + reading = await port.read(EpicsPvAddress(pv)) + assert reading.kind == "Scalar" + finally: + await port.aclose() + + +@pytest.mark.integration +async def test_subscribe_char_waveform_declared_as_text_yields_decoded_strings( + softioc: str, +) -> None: + """Subscribe honours the same declaration as read, per update.""" + pv = f"{softioc}text_waveform" + port = EpicsCaControlPort(text_addresses={pv}) + try: + await port.write(EpicsPvAddress(pv), tuple(b"scan started"), wait=True) + iterator = port.subscribe(EpicsPvAddress(pv)) + first = await asyncio.wait_for(anext(iterator), timeout=2.0) + assert first.kind == "Scalar" + assert first.value == "scan started" + + await port.write(EpicsPvAddress(pv), tuple(b"scan complete"), wait=True) + second = await asyncio.wait_for(anext(iterator), timeout=2.0) + assert second.value == "scan complete" + + await iterator.aclose() + finally: + await port.aclose() + + @pytest.mark.integration async def test_read_enum_returns_categorical_with_label(softioc: str) -> None: """DBR_ENUM lands as Measurement(kind='Categorical', value=