From 4227f82b64f6cb4765ae39108a6b3702ecf2cc23 Mon Sep 17 00:00:00 2001 From: Giles Knap Date: Mon, 10 Aug 2026 06:30:07 +0000 Subject: [PATCH 1/3] Add a hardware-free naming API for static PV-name prediction At runtime the controller names are a side effect of ADS discovery: `_get_ethercat_chains` stamps each slave's ChainLocation and `_resolve_controller_name_and_path` renders it through CATioNameMappings. Both need a live bus. A tool migrating an existing installation onto fastcs-catio needs the same names *before* any hardware exists, so it can rewrite the PV references other IOCs hold. `predict_chain`/`predict_names` derive them from the chain order alone, keyed on terminal type names because that is what a static description of a chain carries, where the runtime keys on (vendor_id, product_code, revision_number) reported over ADS. Purely additive: catio_controller.py is untouched. tests/test_naming.py locks the output against the real runtime path over 6 chain shapes x 3 template shapes, so the two implementations cannot drift apart silently. Written for builder2ibek's `catio` command (epics-containers/builder2ibek#128), which imports predict_names. Co-Authored-By: Claude Opus 5 --- src/fastcs_catio/naming.py | 345 +++++++++++++++++++++++++++++++++++++ tests/test_naming.py | 269 +++++++++++++++++++++++++++++ 2 files changed, 614 insertions(+) create mode 100644 src/fastcs_catio/naming.py create mode 100644 tests/test_naming.py diff --git a/src/fastcs_catio/naming.py b/src/fastcs_catio/naming.py new file mode 100644 index 0000000..dee9f85 --- /dev/null +++ b/src/fastcs_catio/naming.py @@ -0,0 +1,345 @@ +"""Static prediction of the PV names fastcs-catio gives an EtherCAT chain. + +At runtime the names in this module are produced as a side effect of ADS +discovery: :meth:`~fastcs_catio.client.FastCSClient._get_ethercat_chains` +stamps each slave's :class:`~fastcs_catio.devices.ChainLocation`, and +:meth:`~fastcs_catio.catio_controller.CATioServerController._resolve_controller_name_and_path` +renders it through :class:`~fastcs_catio.catio_controller.CATioNameMappings`. +Both need a live bus. + +Tools that migrate an existing installation onto fastcs-catio need the same +names *before* any hardware exists — they have to rewrite the PV references in +other IOCs' databases ahead of time. This module derives them from the chain +order alone, keyed on terminal **type names** (``"EL3104"``), because that is +what a static description of a chain carries; the runtime instead keys on the +``(vendor_id, product_code, revision_number)`` identity reported over ADS. + +Nothing here connects to hardware, runs a coroutine, or touches FastCS +controllers. ``tests/test_naming.py`` locks the output against the runtime +code path so the two cannot drift apart silently. +""" + +from __future__ import annotations + +import re +import string +from dataclasses import dataclass, field + +from fastcs_catio.catio_controller import CATioNameMappings +from fastcs_catio.terminal_config import get_terminal_type + +__all__ = [ + "ChainEntry", + "PredictedSlave", + "UnknownTerminalTypeError", + "predict_chain", + "predict_names", +] + +#: Exact type string that opens a new coupler node (``client.py``). +COUPLER_TYPE = "EK1100" + +#: Bus extension. Before the first coupler it burns one position to reserve the +#: slot of an EK1200 that TwinCAT does not report (``client.py``). +EXTENSION_TYPE = "EK1110" + +#: A box is a coupler-and-terminals in one housing. Anchored, as in ``client.py``. +BOX_TYPE_RE = re.compile(r"(E[PQR]{1}P?\d{4})") + +#: Rendered in place of ``{group_alias}`` when the terminal type has no alias. +DEFAULT_GROUP_ALIAS = "MOD" + +COUPLER = "coupler" +BOX = "box" +SLAVE = "slave" + + +class UnknownTerminalTypeError(KeyError): + """A chain entry names a terminal absent from ``terminal_types.yaml``. + + Raised only when ``strict=True``. The runtime is more forgiving — an + unrecognised identity simply renders as ``MOD`` — but a migration + tool that silently emitted ``MOD`` would rewrite live PV references to + names the IOC never creates, so callers that are generating names for real + hardware should ask for the error. + """ + + +@dataclass(frozen=True) +class ChainEntry: + """One slave in bus order. + + :param type_name: CANopen type string, e.g. ``"EL3104"`` or + ``"EL2024-0010"``. Must match a key of ``terminal_types.yaml``. + :param revision: EtherCAT revision, e.g. ``0x00120000``. Recorded for + diagnostics only: the alias is looked up by type name, which is + revision-independent, mirroring the vendor+product fallback that + :func:`~fastcs_catio.terminal_config.get_terminal_type_by_identity` + applies when a rig runs newer firmware than the cached description. + """ + + type_name: str + revision: int | None = None + + +@dataclass(frozen=True) +class PredictedSlave: + """Where one slave lands, and what it will be called. + + :param node: 1-based coupler/box ordinal; 0 for slaves ahead of the first. + :param position: 0 on the coupler itself, then 1, 2, ... along its terminals. + :param category: :data:`COUPLER`, :data:`BOX` or :data:`SLAVE`. + :param group_alias: alias from ``terminal_types.yaml``, or None when the + type is unknown. + :param index: the number actually rendered into this slave's template. + :param path: PV path segments; ``prefix`` is these joined with ``":"``. + """ + + node: int + position: int + type_name: str + category: str + group_alias: str | None + index: int + path: list[str] = field(default_factory=list) + + @property + def prefix(self) -> str: + """The controller's PV prefix.""" + return ":".join(self.path) + + +def _as_entries(chain) -> list[ChainEntry]: + """Accept ``ChainEntry``, ``(type_name, revision)`` or a bare type name.""" + entries: list[ChainEntry] = [] + for item in chain: + if isinstance(item, ChainEntry): + entries.append(item) + elif isinstance(item, str): + entries.append(ChainEntry(item)) + else: + type_name, revision = item + entries.append(ChainEntry(type_name, revision)) + return entries + + +def _group_alias(type_name: str, strict: bool) -> str | None: + try: + return get_terminal_type(type_name).group_alias + except KeyError as err: + if strict: + raise UnknownTerminalTypeError( + f"terminal type {type_name!r} is not in terminal_types.yaml, so " + f"its group_alias is unknown and its PV names would silently " + f"fall back to {DEFAULT_GROUP_ALIAS!r}" + ) from err + return None + + +def _render(template: str, index: int, context: dict[str, str]) -> str: + """Render one name template exactly as the runtime does. + + The index is passed positionally *and* as ``n``, so ``{}``, ``{:02d}``, + ``{n}`` and ``{n:02d}`` all work. + """ + try: + result = template.format(index, n=index, **context) + except KeyError as err: + raise ValueError( + f"Unknown placeholder {err} in name mapping template {template!r}. " + f"Available keys: {sorted(context)}" + ) from err + except (IndexError, ValueError) as err: + raise ValueError(f"Invalid name mapping template {template!r}: {err}") from err + if "_" in result: + raise ValueError( + f"Rendered PV name segment {result!r} contains an underscore. " + "PV name components must use hyphens, not underscores." + ) + return result + + +def _uses_group_alias(template: str) -> bool: + return any( + key == "group_alias" for _, key, _, _ in string.Formatter().parse(template) + ) + + +def _locate(entries: list[ChainEntry]) -> list[tuple[int, int, str]]: + """Assign ``(node, position, category)`` to each entry, in bus order. + + A port of the loop in ``FastCSClient._get_ethercat_chains``. The two type + tests are kept independent, as they are there; they are mutually exclusive + in practice because ``EK1100`` cannot match :data:`BOX_TYPE_RE`. + """ + located: list[tuple[int, int, str]] = [] + node = 0 + position = 0 + for entry in entries: + type_name = entry.type_name + category = SLAVE + if type_name == EXTENSION_TYPE and node == 0: + position += 1 + if type_name == COUPLER_TYPE: + category = COUPLER + node += 1 + position = 0 + if BOX_TYPE_RE.match(type_name) is not None: + category = BOX + node += 1 + position = 0 + located.append((node, position, category)) + position += 1 + return located + + +def predict_chain( + chain, + mappings: CATioNameMappings | None = None, + root_id: str = "", + device_id: int = 1, + strict: bool = True, +) -> dict[tuple[int, int], PredictedSlave]: + """Predict every controller name for one EtherCAT device's chain. + + :param chain: the device's slaves **in bus order** — :class:`ChainEntry`, + ``(type_name, revision)`` pairs, or bare type-name strings. + :param mappings: name templates; the :class:`CATioNameMappings` defaults + are used when omitted. + :param root_id: the ``id:`` from ``fastcs.yaml``, e.g. + ``"BL21I-VA-CATIO-01"``. Rendered wherever a template says ``{id}``. + :param device_id: the TwinCAT device id rendered into ``device_prefix``. + Not derivable from a static chain — it is read over ADS — so callers + with a single EtherCAT master should leave it at 1. + :param strict: raise :exc:`UnknownTerminalType` for a terminal missing from + ``terminal_types.yaml`` instead of letting its alias fall back to + ``"MOD"``. + + :returns: ``(node, position)`` → :class:`PredictedSlave`, covering couplers + and boxes as well as terminals. + + :raises UnknownTerminalType: in strict mode, per above. + :raises ValueError: for a template that references an unknown placeholder + or renders an underscore into a PV segment. + + Call it once per EtherCAT device: ``node`` restarts at 0 for each, so + merging two devices into one chain would collide. + """ + mappings = mappings or CATioNameMappings() + entries = _as_entries(chain) + located = _locate(entries) + + aliases = [ + _group_alias(entry.type_name, strict) + if category not in (COUPLER, BOX) + else _group_alias(entry.type_name, strict=False) + for entry, (_, _, category) in zip(entries, located, strict=True) + ] + + # Number the modules on each coupler per alias, so "{group_alias}{:02d}" + # yields 24VDI01, 24VDI02, ... independently of the chain position. Keyed + # as the runtime keys it, including the empty-string bucket for no alias. + alias_seq: dict[tuple[int, int], int] = {} + counters: dict[tuple[int, str], int] = {} + for (node, position, category), alias in zip(located, aliases, strict=True): + if category in (COUPLER, BOX): + continue + key = (node, alias or "") + counters[key] = counters.get(key, 0) + 1 + alias_seq[(node, position)] = counters[key] + + device_path = [ + s + for s in _render(mappings.device_prefix, device_id, {"id": root_id}).split(":") + if s + ] + + module_uses_alias = _uses_group_alias(mappings.module_prefix) + + predicted: dict[tuple[int, int], PredictedSlave] = {} + # Only an EK1100 opens a new tree parent. A box gets its own node index but + # stays a leaf, so terminals following it hang off the coupler still open. + coupler_path: list[str] | None = None + + for entry, (node, position, category), alias in zip( + entries, located, aliases, strict=True + ): + if category in (COUPLER, BOX): + parent = coupler_path if category == BOX and coupler_path else device_path + rendered = _render( + mappings.node_prefix, + node, + {"id": root_id, "device_prefix": ":".join(parent)}, + ) + index = node + else: + parent = coupler_path if coupler_path is not None else device_path + index = ( + alias_seq.get((node, position), position) + if module_uses_alias + else position + ) + rendered = _render( + mappings.module_prefix, + index, + { + "id": root_id, + "node_prefix": ":".join(parent), + "device_prefix": ":".join(parent[:-1]) if len(parent) >= 2 else "", + "group_alias": alias or DEFAULT_GROUP_ALIAS, + }, + ) + + path = [s for s in rendered.split(":") if s] + predicted[(node, position)] = PredictedSlave( + node=node, + position=position, + type_name=entry.type_name, + category=category, + group_alias=alias, + index=index, + path=path, + ) + if category == COUPLER: + coupler_path = path + + return predicted + + +def predict_names( + chain, + mappings: CATioNameMappings | None = None, + root_id: str = "", + device_id: int = 1, + strict: bool = True, +) -> dict[tuple[int, int], str]: + """``(node, position)`` → PV prefix, for one EtherCAT device's chain. + + The prefix-only view of :func:`predict_chain`; see it for the arguments and + for the group alias, category and rendered index of each slave. + + >>> from fastcs_catio.naming import predict_names + >>> from fastcs_catio.catio_controller import CATioNameMappings + >>> names = predict_names( + ... ["EK1100", "EL3104", "EL1014"], + ... CATioNameMappings( + ... node_prefix="BL21I-VA-E1RIO-{:02d}", + ... module_prefix="{node_prefix}:{group_alias}{:02d}", + ... ), + ... root_id="BL21I-VA-CATIO-01", + ... ) + >>> names[(1, 1)] + 'BL21I-VA-E1RIO-01:10VAI01' + >>> names[(1, 2)] + 'BL21I-VA-E1RIO-01:24VDI01' + """ + return { + key: slave.prefix + for key, slave in predict_chain( + chain, + mappings=mappings, + root_id=root_id, + device_id=device_id, + strict=strict, + ).items() + } diff --git a/tests/test_naming.py b/tests/test_naming.py new file mode 100644 index 0000000..d0df0ca --- /dev/null +++ b/tests/test_naming.py @@ -0,0 +1,269 @@ +"""Tests for the static naming API. + +The point of :mod:`fastcs_catio.naming` is that it agrees with what the live +IOC does. So most of this file does not assert against hand-written strings — +it drives the *runtime* code path (``AsyncioADSClient._get_ethercat_chains``, +``CATioServerController._compute_module_alias_indices`` and +``_resolve_controller_name_and_path``) over a synthetic chain and asserts that +:func:`~fastcs_catio.naming.predict_chain` produced the same answer. Change the +rule in either place without the other and these fail. +""" + +import asyncio + +import pytest + +from fastcs_catio._constants import DeviceType +from fastcs_catio._types import AmsNetId +from fastcs_catio.catio_controller import CATioNameMappings, CATioServerController +from fastcs_catio.client import AsyncioADSClient +from fastcs_catio.devices import ( + DeviceFrames, + IODevice, + IOIdentity, + IOServer, + IOSlave, + IOTreeNode, + SlaveCRC, + SlaveState, +) +from fastcs_catio.naming import ( + ChainEntry, + UnknownTerminalTypeError, + predict_chain, + predict_names, +) +from fastcs_catio.terminal_config import get_terminal_type + +ROOT_ID = "BL21I-VA-CATIO-01" + +DLS_MAPPINGS = CATioNameMappings( + device_prefix="{id}:ETH{:02d}", + node_prefix="BL21I-VA-E1RIO-{:02d}", + module_prefix="{node_prefix}:{group_alias}{:02d}", +) + +# BL21I-VA-IOC-01's chain, in the order its builder XML declares the slaves. +# Its legacy coupler labels run ERIO-04, -12, -03, -02, -01 -- the legacy +# numbers carry no ordering information, which is the whole reason this +# module exists. +I21_VA_IOC_01 = [ + "EK1100", "EL3104", "EL3104", "EL3104", "EL3104", "EL1014", "EL1014", + "EK1100", "EL3104", "EL3104", + "EK1100", "EL3104", "EL3104", "EL1014", "EL1014", + "EK1100", "EL3104", "EL3104", "EL1014", "EL1014", + "EK1100", "EL3104", "EL3104", "EL3104", "EL3104", "EL1014", "EL1014", +] # fmt: skip + + +def _make_slave(type_name: str, index: int) -> IOSlave: + """An IOSlave carrying the real identity of ``type_name``.""" + terminal = get_terminal_type(type_name) + return IOSlave( + parent_device=1, + type=type_name, + name=f"Slave{index:03d}", + address=1000 + index, + identity=IOIdentity( + vendor_id=terminal.identity.vendor_id, + product_code=terminal.identity.product_code, + revision_number=terminal.identity.revision_number, + serial_number=0, + ), + states=SlaveState(ecat_state=0, link_status=0), + crcs=SlaveCRC(port_a_crc=0, port_b_crc=0, port_c_crc=0, port_d_crc=0), + ) + + +def _make_client(chain: list[str], device_id: int = 1) -> AsyncioADSClient: + """A client with one EtherCAT device, bypassing all connection setup.""" + client = object.__new__(AsyncioADSClient) + device = IODevice( + id=device_id, + type=DeviceType.IODEVICETYPE_ETHERCAT, + name=f"Device{device_id}", + netid=AmsNetId.from_string("127.0.0.1.1.1"), + identity=IOIdentity( + vendor_id=1, product_code=2, revision_number=3, serial_number=4 + ), + frame_counters=DeviceFrames( + time=0, cyclic_sent=0, cyclic_lost=0, acyclic_sent=0, acyclic_lost=0 + ), + slave_count=len(chain), + slaves_states=[], + slaves_crc_counters=[], + slaves=[_make_slave(t, i) for i, t in enumerate(chain)], + ) + client._ecdevices = {device_id: device} + client.ioserver = IOServer(name=ROOT_ID, version="1", build=0, num_devices=1) + return client + + +def _drive(coro): + """Run *coro* to completion without disturbing the suite's event loop. + + ``asyncio.run`` clears the process-wide loop when it finishes. Other tests + here -- ``test_system.py`` in particular, which drives a real ADS simulator + -- install a loop of their own, and clearing it strands that loop with its + sockets still open. It is then collected at some arbitrary later point and + pytest reports the unraisable ResourceWarning against whichever test is + running at the time, which is never this one. So use a private loop, close + it, and put back whatever was installed before. + """ + try: + previous = asyncio.get_event_loop_policy().get_event_loop() + except RuntimeError: + previous = None + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + asyncio.set_event_loop(previous) + + +def _runtime_names( + chain: list[str], mappings: CATioNameMappings, device_id: int = 1 +) -> dict[tuple[int, int], str]: + """Run the production code path and collect ``(node, position) -> prefix``. + + Mirrors ``CATioServerController.get_subcontrollers_from_node`` closely + enough to reach every node, without needing FastCS controllers or a bus. + """ + client = _make_client(chain, device_id) + _drive(client._get_ethercat_chains()) + tree = client._generate_system_tree() + + controller = object.__new__(CATioServerController) + controller._path = [ROOT_ID] + controller._name_mappings = mappings + controller._module_alias_indices = controller._compute_module_alias_indices(tree) + + names: dict[tuple[int, int], str] = {} + + def walk(node: IOTreeNode, parent_path: list[str]) -> None: + for child in node.children: + _, path = controller._resolve_controller_name_and_path(child, parent_path) + data = child.data + if isinstance(data, IOSlave): + key = (int(data.loc_in_chain.node), int(data.loc_in_chain.position)) + names[key] = ":".join(path) + walk(child, path) + + walk(tree, controller._path) + return names + + +@pytest.mark.parametrize( + "chain", + [ + pytest.param(I21_VA_IOC_01, id="i21-va-ioc-01"), + pytest.param(["EK1100", "EL3104", "EL1014", "EL2024-0010"], id="mixed-aliases"), + pytest.param(["EK1100", "EL1014", "EL1014", "EL1014"], id="same-alias-run"), + pytest.param(["EK1110", "EK1100", "EL3104"], id="leading-ek1110"), + pytest.param(["EK1100", "EL3104", "EK1122", "EL3104"], id="ek1122-junction"), + pytest.param(["EK1100", "EL3104", "EK1100", "EL3104"], id="two-couplers"), + ], +) +@pytest.mark.parametrize( + "mappings", + [ + pytest.param(DLS_MAPPINGS, id="dls"), + pytest.param(CATioNameMappings(), id="library-defaults"), + pytest.param( + CATioNameMappings(module_prefix="{node_prefix}:MOD{:02d}"), + id="no-group-alias", + ), + ], +) +def test_prediction_matches_runtime(chain, mappings): + """The static prediction equals what the live discovery path produces.""" + predicted = predict_names(chain, mappings, root_id=ROOT_ID, strict=False) + assert predicted == _runtime_names(chain, mappings) + + +def test_i21_worked_examples(): + """The three migration examples from the DLS I21 conversion design.""" + names = predict_names(I21_VA_IOC_01, DLS_MAPPINGS, root_id=ROOT_ID) + + # legacy BL21I-VA-ERIO-01:MOD1 -- 5th coupler, 1st terminal + assert names[(5, 1)] == "BL21I-VA-E1RIO-05:10VAI01" + # legacy BL21I-VA-ERIO-01:MOD5 -- the EL1014s restart the alias sequence + assert names[(5, 5)] == "BL21I-VA-E1RIO-05:24VDI01" + # legacy BL21I-VA-ERIO-04:MOD1 -- 1st coupler, so ERIO-04 becomes E1RIO-01 + assert names[(1, 1)] == "BL21I-VA-E1RIO-01:10VAI01" + + +def test_alias_sequence_is_per_coupler_not_per_chain(): + """Each coupler restarts the per-alias numbering.""" + names = predict_names(I21_VA_IOC_01, DLS_MAPPINGS, root_id=ROOT_ID) + assert names[(1, 1)] == "BL21I-VA-E1RIO-01:10VAI01" + assert names[(2, 1)] == "BL21I-VA-E1RIO-02:10VAI01" + assert names[(3, 1)] == "BL21I-VA-E1RIO-03:10VAI01" + + +def test_coupler_is_position_zero(): + """A coupler sits at position 0; its first terminal is position 1.""" + chain = predict_chain(["EK1100", "EL3104"], DLS_MAPPINGS, root_id=ROOT_ID) + assert chain[(1, 0)].category == "coupler" + assert chain[(1, 0)].prefix == "BL21I-VA-E1RIO-01" + assert chain[(1, 1)].category == "slave" + + +def test_leading_ek1110_burns_a_position(): + """An EK1110 before the first coupler reserves the unreported EK1200 slot.""" + chain = predict_chain(["EK1110", "EK1100", "EL3104"], DLS_MAPPINGS, strict=False) + assert (0, 1) in chain # the EK1110, shifted off position 0 + assert chain[(1, 1)].type_name == "EL3104" + + +def test_group_aliases_of_the_i21_terminal_types(): + """Locks the aliases the I21 substitution table is built on.""" + aliases = { + t: get_terminal_type(t).group_alias + for t in ("EL3104", "EL1014", "EL2024-0010", "EL3356-0010") + } + assert aliases == { + "EL3104": "10VAI", + "EL1014": "24VDI", + "EL2024-0010": "12VDO", + "EL3356-0010": "AI", + } + + +def test_revision_is_ignored_for_alias_lookup(): + """I21 declares revisions the YAML does not carry; the alias must survive.""" + declared = predict_names( + [ChainEntry("EK1100", 0x00120000), ChainEntry("EL3104", 0x00130000)], + DLS_MAPPINGS, + root_id=ROOT_ID, + ) + assert declared[(1, 1)] == "BL21I-VA-E1RIO-01:10VAI01" + + +def test_unknown_terminal_type_raises_in_strict_mode(): + with pytest.raises(UnknownTerminalTypeError, match="EL9999"): + predict_names(["EK1100", "EL9999"], DLS_MAPPINGS, root_id=ROOT_ID) + + +def test_unknown_terminal_type_falls_back_to_mod_when_not_strict(): + """Matches the runtime, which renders an unrecognised identity as MOD.""" + names = predict_names(["EK1100", "EL9999"], DLS_MAPPINGS, strict=False) + assert names[(1, 1)] == "BL21I-VA-E1RIO-01:MOD01" + + +def test_absolute_node_prefix_keeps_the_path_short_enough_to_avoid_shortening(): + """The DLS templates must not push attribute names past the EPICS budget. + + With the library default ``node_prefix`` the module path gains two extra + segments, the per-attribute budget collapses, and ``shorten_fastcs_name`` + truncates ``ai_standard_channel_1_value`` to ``ai_std_ch`` -- dropping the + channel number and colliding all four EL3104 channels onto one name. + """ + from catio_terminals.utils import shorten_fastcs_name + from fastcs_catio.utils import max_attribute_name_length + + prefix = predict_names(I21_VA_IOC_01, DLS_MAPPINGS, root_id=ROOT_ID)[(5, 1)] + budget = max_attribute_name_length(prefix.split(":"), is_rw=False) + name = "ai_standard_channel_1_value" + assert shorten_fastcs_name(name, budget) == name From d319f1a29011e31c2ceddfc0c60ddc9924135480 Mon Sep 17 00:00:00 2001 From: Giles Knap Date: Mon, 10 Aug 2026 10:01:53 +0000 Subject: [PATCH 2/3] Drive the naming tests on a loop that is never installed globally The helper added with these tests read the current event loop so it could put it back afterwards. `asyncio.get_event_loop()` is deprecated from Python 3.12, and `filterwarnings = "error"` turns that warning into an exception raised before the coroutine is ever awaited, so the file dies with "coroutine _get_ethercat_chains was never awaited". It only fires when no loop has been installed yet, which is why CI did not catch it: in a full run an earlier test has already installed one and the lookup returns it without warning. Running just this file -- what a developer does -- fails on 3.12 and 3.13. `new_event_loop()` does not install a loop and `run_until_complete` drives the loop object directly, so nothing needs saving or restoring. That keeps the original property the helper existed for (not clearing the loop test_system.py installs for its ADS simulator, whose sockets would then be collected unclosed and reported against an unrelated test) without reading global asyncio state at all. Verified on 3.11, 3.12 and 3.13, both for this file alone and for the full suite. Co-Authored-By: Claude Opus 5 --- tests/test_naming.py | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/tests/test_naming.py b/tests/test_naming.py index d0df0ca..7cae823 100644 --- a/tests/test_naming.py +++ b/tests/test_naming.py @@ -100,26 +100,27 @@ def _make_client(chain: list[str], device_id: int = 1) -> AsyncioADSClient: def _drive(coro): - """Run *coro* to completion without disturbing the suite's event loop. - - ``asyncio.run`` clears the process-wide loop when it finishes. Other tests - here -- ``test_system.py`` in particular, which drives a real ADS simulator - -- install a loop of their own, and clearing it strands that loop with its - sockets still open. It is then collected at some arbitrary later point and - pytest reports the unraisable ResourceWarning against whichever test is - running at the time, which is never this one. So use a private loop, close - it, and put back whatever was installed before. + """Run *coro* to completion on a private loop, leaving the global one alone. + + Deliberately not ``asyncio.run``: that clears the process-wide loop when it + finishes. ``test_system.py`` installs a loop of its own to drive the ADS + simulator, and clearing it strands that loop with its sockets still open -- + it is collected at some arbitrary later point and pytest reports the + unraisable ResourceWarning against whichever test happens to be running, + which is never this one. + + Reading the previous loop to restore it afterwards is no good either: + ``get_event_loop()`` is deprecated from 3.12, and with + ``filterwarnings = "error"`` the warning becomes an exception raised before + the coroutine is ever awaited. So never touch the global loop at all -- + ``new_event_loop()`` does not install one, and ``run_until_complete`` + drives the loop object directly. """ - try: - previous = asyncio.get_event_loop_policy().get_event_loop() - except RuntimeError: - previous = None loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro) finally: loop.close() - asyncio.set_event_loop(previous) def _runtime_names( From fdfdd2b5823ceef514a73eab02ae6145ab82110b Mon Sep 17 00:00:00 2001 From: Giles Knap Date: Mon, 10 Aug 2026 10:20:33 +0000 Subject: [PATCH 3/3] Select the EL2595 output current so it gets a PV `dox_current_output_current` was the only unselected symbol anyone actually uses. It is the terminal's whole point -- an EL2595 is an LED constant current driver and the output current is the value you set -- while the two symbols that were selected, `dox_status` and `dox_control`, are packed bitfields. DLS's legacy `ethercat` module exposes it as `$(DEVICE):DOXCURRENT:OUTPUTCURRENT` and BL21I writes to it on four terminals (`BL21I-DI-LED-01/02/03` and `BL21I-OP-LED-01`). Without this the migration onto fastcs-catio has nothing to point those references at, because no PV is created for the attribute at all. Hand-edited rather than regenerated, by the maintainer's explicit decision: this is a single boolean selection, exactly what the `catio-terminals edit` GUI toggles, not an import of new Beckhoff XML. Verified as the only selection change in the file. Co-Authored-By: Claude Opus 5 --- src/catio_terminals/terminals/terminal_types.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/catio_terminals/terminals/terminal_types.yaml b/src/catio_terminals/terminals/terminal_types.yaml index 34524b3..e5adbca 100644 --- a/src/catio_terminals/terminals/terminal_types.yaml +++ b/src/catio_terminals/terminals/terminal_types.yaml @@ -772,7 +772,7 @@ terminal_types: channels: 1 access: Read/Write fastcs_name: dox_current_output_current - selected: false + selected: true bit_offset: 0 - name_template: DOX Impulse length.Impulse length index_group: 61473