from __future__ import annotations
import inspect
import re
from collections.abc import Mapping
from pydantic import (
Field,
field_validator,
)
from ophyd_async.core import (
ConfinedModel,
Device,
DeviceFiller,
DeviceMap,
DeviceVector,
LazyMock,
Signal,
SignalR,
SignalRW,
SignalW,
TriggerableCommand,
gather_dict,
)
from ._epics_connector import _PvPrefixDeviceConnector, fill_children_with_prefix
from ._signal import PvaCommandBackend, PvaSignalBackend, pvget_with_timeout
from ._util import EpicsCommandBackend
[docs]
class PviDeviceConnector(_PvPrefixDeviceConnector):
"""Connect a `Device` to a PVI structure served over PVA.
At construction, create the type-hinted signals, commands and sub-devices
from the `Device`'s annotations. At connection, discover the served PVI tree
and fill in each child, checking the type-hinted ones and adding any extra
ones the tree provides.
:param prefix:
The PV prefix of the device; "PVI" is appended to it to get the PVI PV.
:param error_hint:
If given, appended to the error message when a type-hinted child is
missing from the served tree.
"""
mock_device_vector_children: list[int] = [1, 2]
pvi_tree: PviTree | None = None
def __init__(self, prefix: str = "", error_hint: str = "") -> None:
# TODO: what happens if we get a leading "pva://" here?
super().__init__(prefix)
self.error_hint = error_hint
[docs]
def set_prefix(self, prefix: str) -> None:
super().set_prefix(prefix)
self.pvi_pv = prefix + "PVI"
[docs]
def create_children_from_annotations(self, device: Device) -> None:
# Per the DeviceConnector contract this may be called more than once;
# having built the filler, subsequent calls are a no-op.
if hasattr(self, "filler"):
return
def command_backend_factory(
sig: inspect.Signature | None,
) -> EpicsCommandBackend:
# EPICS only supports void/void commands (a plain PV put); a typed
# Command[P, T] annotation is a mistake on an EPICS device.
if sig is not None:
raise TypeError(
f"{device.name}: EPICS only supports TriggerableCommand /"
" Command[[], None]; typed Command with parameters is not"
" yet supported over EPICS"
)
return PvaCommandBackend()
self.filler = DeviceFiller(
device=device,
signal_backend_factory=PvaSignalBackend,
device_connector_factory=PviDeviceConnector,
command_backend_factory=command_backend_factory,
)
# Address any PvSuffix-annotated children now; everything else is
# addressed from the served PVI tree at connect, which also cross-checks
# the ones addressed here.
fill_children_with_prefix(self.prefix, self.filler, filled=False)
self.filler.check_created()
[docs]
async def connect_mock(self, device: Device, mock: LazyMock) -> None:
# A DeviceVector gets fabricated mock children; a DeviceMap does not (its
# entries come from the served PVI tree, so in mock mode it stays empty).
if isinstance(device, DeviceVector):
self.filler.create_device_dict_entries_to_mock(
self.mock_device_vector_children
)
device.set_name(device.name)
return await super().connect_mock(device, mock)
[docs]
async def connect_real(
self, device: Device, timeout: float, force_reconnect: bool
) -> None:
if self.pvi_tree is None:
# Top-level device: discover the served PVI tree.
self.pvi_tree = await PviTree.build_device_tree(
pvi_pv=self.pvi_pv, timeout=timeout
)
tree = self.pvi_tree
# A DeviceMap is filled from its node's normal named entries, keyed by
# name. Anything else fills its named entries as attributes, plus any
# "__N" entries into the int-keyed DeviceVector that it is.
is_map = isinstance(device, DeviceMap)
if is_map and tree.vector_children:
raise TypeError(
f"{self.pvi_pv}: {device.name} is annotated as a DeviceMap, but "
f"PVI serves integer-keyed entries {sorted(tree.vector_children)} "
"here, which only a DeviceVector can hold"
)
self._fill_named_entries(tree, as_map=is_map)
if not is_map:
self._fill_vector_children(device, tree)
suffix = f"\n{self.error_hint}" if self.error_hint else ""
self.filler.check_filled(f"{self.pvi_pv}: {tree}{suffix}")
device.set_name(device.name)
return await super().connect_real(device, timeout, force_reconnect)
def _fill_named_entries(self, tree: PviTree, *, as_map: bool) -> None:
"""Fill a node's normal (non-"__N") entries.
`as_map=True` adds each entry to a `DeviceMap` keyed by its name;
`as_map=False` sets it as a plain attribute on the device.
"""
for name, details in tree.signals.items():
self._fill_signal(name, details, map_key=name if as_map else None)
for name, execute_pv in tree.commands.items():
self._fill_command(name, execute_pv, map_key=name if as_map else None)
for name, sub_tree in tree.sub_devices.items():
# A sub-tree with "__N" entries is an (undeclared) DeviceVector;
# for declared children the filler already knows the type.
device_type = DeviceVector if sub_tree.vector_children else Device
self._fill_sub_device(
name,
sub_tree,
device_type=device_type,
map_key=name if as_map else None,
)
def _fill_vector_children(self, device: Device, tree: PviTree) -> None:
"""Fill the int-keyed "__N" entries into the `DeviceVector` `device`.
`PviTree` validates the entries are all one kind, so the child value type
selects how to fill it.
"""
for key, child in tree.vector_children.items():
if isinstance(child, SignalDetails):
self._fill_signal(device.name, child, map_key=key)
elif isinstance(child, PviTree):
self._fill_sub_device(device.name, child, map_key=key)
else: # a bare execute-PV str is a command
self._fill_command(device.name, child, map_key=key)
def _check_agrees_with_pvi(self, name: str, annotated: str, served: str) -> None:
"""Raise if a child's PvSuffix annotation disagrees with the PVI tree.
`annotated` is empty for the children PVI alone addresses, which is the
usual case; a child that was given a `PvSuffix` as well must agree with
what the tree says, or one of the two is wrong.
"""
if annotated and annotated != served:
raise TypeError(
f"{self.pvi_pv}: {name} is addressed at {annotated} by its "
f"PvSuffix annotation, but PVI serves it at {served}"
)
def _fill_signal(
self, name: str, details: SignalDetails, map_key: int | str | None = None
) -> None:
backend = self.filler.fill_child_signal(name, details.signal_type, map_key)
self._check_agrees_with_pvi(name, backend.read_pv, details.read_pv)
self._check_agrees_with_pvi(name, backend.write_pv, details.write_pv)
backend.read_pv = details.read_pv
backend.write_pv = details.write_pv
def _fill_command(
self, name: str, execute_pv: str, map_key: int | str | None = None
) -> None:
backend = self.filler.fill_child_command(name, TriggerableCommand, map_key)
self._check_agrees_with_pvi(name, backend.write_pv, execute_pv)
backend.write_pv = execute_pv
def _fill_sub_device(
self,
name: str,
sub_tree: PviTree,
*,
device_type: type[Device] = Device,
map_key: int | str | None = None,
) -> None:
connector = self.filler.fill_child_device(
name, device_type=device_type, map_key=map_key
)
# An unaddressed connector still has the "PVI" its empty prefix makes,
# so only cross-check the ones a PvSuffix gave a prefix to.
self._check_agrees_with_pvi(
name, connector.pvi_pv if connector.prefix else "", sub_tree.pvi_pv
)
connector.pvi_tree = sub_tree
connector.pvi_pv = sub_tree.pvi_pv
[docs]
class SignalDetails(ConfinedModel):
"""Representation of a Signal to be constructed."""
signal_type: type[Signal]
read_pv: str
write_pv: str
[docs]
@classmethod
def from_entry(cls, entry: dict[str, str]) -> SignalDetails:
match entry:
case {"r": read_pv, "w": write_pv}:
return cls(signal_type=SignalRW, read_pv=read_pv, write_pv=write_pv)
case {"rw": pv}:
return cls(signal_type=SignalRW, read_pv=pv, write_pv=pv)
case {"r": read_pv}:
return cls(signal_type=SignalR, read_pv=read_pv, write_pv=read_pv)
case {"w": write_pv}:
return cls(signal_type=SignalW, read_pv=write_pv, write_pv=write_pv)
case _:
raise TypeError(f"Can't process entry {entry}")
[docs]
class PviTree(ConfinedModel):
"""Representation of a PVI structure of devices and signals in a PVI query.
Example 1: A device with sub-devices and signals
--------------------------------------
For a PVI structure such as:
```json
{
"bit": {"d": "TEST-PANDA:Bits:PVI"},
"calc": {"d": "TEST-PANDA:Calc:PVI"},
"a": {"rw": "TEST-PANDA:Bits:A"}
}
```
From "TEST-PANDA:PVI", This would be represented as:
```python
PviTree(
pvi_pv="TEST-PANDA:PVI",
signals={
"a": SignalDetails(
signal_type=SignalRW,
read_pv="TEST-PANDA:Bits:A",
write_pv="TEST-PANDA:Bits:A")
},
sub_devices={
"bit": PviTree(...),
"calc": PviTree(...)
},
vector_children=[]
)
```
Example 2: A device with vector children
-----------------------------------------
If an entry like `"calc"` is a **DeviceVector**
(e.g., mirroring a fastCS controller vector), the PVI entries will look like this:
```json
{
"__1": {"d": "TEST-PANDA:Calc:2:PVI"},
"__2": {"d": "TEST-PANDA:Calc:1:PVI"}
}
```
This would be represented as:
```python
PviTree(
pvi_pv="TEST-PANDA:Calc:PVI",
signals={},
sub_devices={},
vector_children=[
PviTree(pvi_pv="TEST-PANDA:Calc:2:PVI", signals={}, ...),
PviTree(pvi_pv="TEST-PANDA:Calc:1:PVI", signals={}, ...)
]
)
```
This is similar for vectors of signals, where `vector_children` would instead
be populated with `SignalDetails`
Example 3: A device with legacy vector children
-----------------------------------------
Legacy PVI vector structure is supported, for backwards compatability
with pandablocks-ioc, where vector children are represented as:
```
{
"calc": [None, {"d": "TEST-PANDA:Calc1:PVI"}, {"d": "TEST-PANDA:Calc2:PVI"}],
}
```
generate the same PviTree as in Example 2, excluding a PVI PV.
:param pvi_pv:
The PVI PV of the device.
:param signals:
A mapping of signal names to `SignalDetails` objects.
:param commands:
A mapping of command names to their execute PV strings (from "x" PVI entries).
:param sub_devices:
A mapping of sub-device names to their corresponding `PviTree` objects.
:param vector_children:
A mapping of int to `PviTree` objects representing child devices of a vector
device.
"""
pvi_pv: str = Field(default="")
signals: Mapping[str, SignalDetails] = Field(default_factory=dict)
commands: Mapping[str, str] = Field(default_factory=dict)
sub_devices: Mapping[str, PviTree] = Field(default_factory=dict)
# A `str` vector_child is the execute PV of a DeviceVector of commands
# (a "__N" entry whose only key is "x").
vector_children: Mapping[int, PviTree | SignalDetails | str] = Field(
default_factory=dict
)
[docs]
@classmethod
async def build_device_tree(cls, pvi_pv: str, timeout: float) -> PviTree:
"""Recursively build a PviTree from a top level device.
Starting from the top-level device, this classmethod performs
post-order traversal over the served PVI structure, populating
a PviTree from the bottom up.
:param name: Device name
:param pvi_pv: Device PVI PV
:param timeout: Timeout on pvget
"""
pvi_structure = await pvget_with_timeout(pvi_pv, timeout)
# An example entry is: {"d": "Prefix:Device:PVI", "rw": "Prefix:A"}
# these entries are stored under the parent PVI structure name
# for example, {"device": {"d": "Prefix:Device:PVI", "rw": "Prefix:A"}}
entries: dict[str, dict[str, str]] = pvi_structure["value"].todict()
# Separate "x" (command) entries before processing signals
commands: dict[str, str] = {
entry_name: entries.pop(entry_name)["x"]
for entry_name in list(entries)
if not isinstance(entries[entry_name], list)
and set(entries[entry_name]) == {"x"}
}
signal_details = {
entry_name: SignalDetails.from_entry(entries.pop(entry_name))
for entry_name in list(entries)
if not isinstance(entries[entry_name], list)
and set(entries[entry_name]) != {"d"}
}
sub_trees = await gather_dict(
{
entry_name: cls._handle_legacy_entry(entry, timeout)
if isinstance(entry, list) # Found a legacy entry, try to handle
else cls.build_device_tree(entry["d"], timeout)
for entry_name, entry in entries.items()
}
)
vector_children: dict[int, PviTree | SignalDetails | str] = {}
# Filter DeviceVector children ("__N" integer entries) out of the
# stand-alone devices/signals/commands ("commands" holds bare execute-PV
# strings for "__N" entries here, i.e. a DeviceVector of commands). There
# is deliberately no string-keyed equivalent: a DeviceMap is only ever
# created from an explicit `DeviceMap[...]` annotation, and is filled from
# a node's normal named entries (see PviDeviceConnector._fill_device_map).
for processed_entries in (sub_trees, signal_details, commands):
for child_name in list(processed_entries):
if m := re.match(r"^__(\d+)$", child_name):
vector_children[int(m.group(1))] = processed_entries.pop(child_name)
return PviTree(
pvi_pv=pvi_pv,
signals=signal_details,
commands=commands,
sub_devices=sub_trees,
vector_children=vector_children,
)
@classmethod
async def _handle_legacy_entry(
cls, legacy_entry: list[None | dict[str, str]], timeout: float
) -> PviTree:
"""Handle legacy vector entries. Cannot be converted to map as have no str keys.
For example;
```
{
"calc": [None, {"d": "TEST-PANDA:Calc1:PVI"}, {"d": "TEST-PANDA:Calc2:PVI"}]
}
```
a `PviTree` is built for each device entry in this list.
"""
sub_trees = await gather_dict(
{
vector_index: cls.build_device_tree(vector_entry["d"], timeout)
for vector_index, vector_entry in enumerate(legacy_entry)
if vector_entry is not None
}
)
# Legacy FastCS vector should not contain child signals,
# devices, or its own PVI PV.
return PviTree(
vector_children=sub_trees,
)
def __str__(self) -> str:
"""Print a readable top layer of the PviTree."""
children = {
**{
child_name: tree.pvi_pv for child_name, tree in self.sub_devices.items()
},
**{
child_name: vector_child.pvi_pv
for child_name, vector_child in self.vector_children.items()
if isinstance(vector_child, PviTree)
},
}
signals = {
**{
signal_name: (
detail.signal_type.__name__,
detail.read_pv,
detail.write_pv,
)
for signal_name, detail in self.signals.items()
},
**{
signal_name: (
vector_child.signal_type.__name__,
vector_child.read_pv,
vector_child.write_pv,
)
for signal_name, vector_child in self.vector_children.items()
if isinstance(vector_child, SignalDetails)
},
}
return f"sub_devices={children}\nsignals={signals}"
@field_validator("vector_children")
@classmethod
def _check_consistency_of_vector_children_type(
cls,
vector_children: Mapping[int, PviTree | SignalDetails | str],
):
"""Validates that parsed vector children are all of the same type."""
if not (
all(
isinstance(vector_child, SignalDetails)
for vector_child in vector_children.values()
)
or all(
isinstance(vector_child, PviTree)
for vector_child in vector_children.values()
)
or all(
isinstance(vector_child, str)
for vector_child in vector_children.values()
)
):
raise ValueError(
"Failed to validate PviTree. "
"vector_children must all be of type `SignalDetails`, `PviTree` "
f"or `str`. Received mixed type: {vector_children=}"
)
return vector_children
