`ophyd_async.epics.testing`#

Package Contents#

Classes#

`EpicsTestCaDevice`	Device for use in a channel access test IOC.
`EpicsTestEnum`	For testing strict enum values in test IOCs.
`EpicsTestSubsetEnum`	For testing subset enum values in test IOCs.
`EpicsTestSupersetEnum`	For testing superset enum values in test IOCs.
`EpicsTestPvaDevice`	Device for use in a pv access test IOC.
`EpicsTestPviDevice`	Device for use in a generic PVI test IOC, independent of PandA/FastCS.
`EpicsTestPviLeafDevice`	Leaf sub-device for `child` and `device_vector` on EpicsTestPviNestedDevice.
`EpicsTestPviNestedDevice`	Real-IOC-backed structural PVI test device.
`EpicsTestPviNestedDeviceMissingChild`	Connects to the same served PVI tree as EpicsTestPviNestedDevice.
`EpicsTestTable`	An abstraction of a Table where each field is a column.

Functions#

`generate_random_pv_prefix`	For generating random PV names in test devices.
`start_ioc`	Start an EPICS IOC subprocess by running `script` with `args`.

Data#

`CA_PVA_RECORDS`
`IOC`
`PVA_RECORDS`
`PVI_NESTED_RECORDS`

API#

ophyd_async.epics.testing.CA_PVA_RECORDS#: None

ophyd_async.epics.testing.IOC#: None

ophyd_async.epics.testing.PVA_RECORDS#: None

ophyd_async.epics.testing.PVI_NESTED_RECORDS#: None

class ophyd_async.epics.testing.EpicsTestCaDevice(prefix: str, with_pvi: bool = False, name: str = '')[source]#

Bases: ophyd_async.epics.core.EpicsDevice

Device for use in a channel access test IOC.

a_int: Annotated[SignalRW[int], PvSuffix('int')]#

None

A thing

a_float: Annotated[SignalRW[float], PvSuffix('float')]#: None

float_prec_0: Annotated[SignalRW[int], PvSuffix('float_prec_0')]#: None

a_str: Annotated[SignalRW[str], PvSuffix('str')]#: None

longstr: Annotated[SignalRW[str], PvSuffix('longstr')]#: None

longstr2: Annotated[SignalRW[str], PvSuffix('~longstr2.VAL$')]#: None

a_bool: Annotated[SignalRW[bool], PvSuffix('bool')]#: None

slowseq: Annotated[SignalRW[int], PvSuffix('slowseq')]#: None

enum: Annotated[SignalRW[EpicsTestEnum], PvSuffix('enum')]#: None

enum2: Annotated[SignalRW[EpicsTestEnum], PvSuffix('enum2')]#: None

subset_enum: Annotated[SignalRW[EpicsTestSubsetEnum], PvSuffix('subset_enum')]#: None

superset_enum: Annotated[SignalRW[EpicsTestSupersetEnum], PvSuffix('subset_enum')]#: None

enum_str_fallback: Annotated[SignalRW[str], PvSuffix('enum_str_fallback')]#: None

bool_unnamed: Annotated[SignalRW[bool], PvSuffix('bool_unnamed')]#: None

partialint: Annotated[SignalRW[int], PvSuffix('partialint')]#: None

lessint: Annotated[SignalRW[int], PvSuffix('lessint')]#: None

uint8a: Annotated[SignalRW[Array1D[uint8]], PvSuffix('uint8a')]#: None

int16a: Annotated[SignalRW[Array1D[int16]], PvSuffix('int16a')]#: None

int32a: Annotated[SignalRW[Array1D[int32]], PvSuffix('int32a')]#: None

float32a: Annotated[SignalRW[Array1D[float32]], PvSuffix('float32a')]#: None

float64a: Annotated[SignalRW[Array1D[float64]], PvSuffix('float64a')]#: None

stra: Annotated[SignalRW[Sequence[str]], PvSuffix('stra')]#: None

mbb_direct_bit_r: Annotated[SignalR[bool], PvSuffix('~mbb_direct.B0')]#: None

mbb_direct_bit: Annotated[SignalRW[bool], PvSuffix('~mbb_direct_rw.B0')]#: None

go: Annotated[TriggerableCommand, PvSuffix('go')]#: None

class ophyd_async.epics.testing.EpicsTestEnum[source]#

Bases: ophyd_async.core.StrictEnum

For testing strict enum values in test IOCs.

A#: ‘Aaa’

B#: ‘Bbb’

C#: ‘Ccc’

class ophyd_async.epics.testing.EpicsTestSubsetEnum[source]#

Bases: ophyd_async.core.SubsetEnum

For testing subset enum values in test IOCs.

The backing PV (mbbo with choices Aaa/Bbb/Ccc) has an extra choice (“Ccc”) beyond what this enum declares.

A#: ‘Aaa’

B#: ‘Bbb’

class ophyd_async.epics.testing.EpicsTestSupersetEnum[source]#

Bases: ophyd_async.core.SupersetEnum

For testing superset enum values in test IOCs.

Deliberately points at the same backing PV as EpicsTestSubsetEnum (mbbo with choices Aaa/Bbb/Ccc): this enum declares a choice (“Ddd”) that does not exist in the backend, which is exactly what a superset enum allows.

A#: ‘Aaa’

B#: ‘Bbb’

C#: ‘Ccc’

D#: ‘Ddd’

class ophyd_async.epics.testing.EpicsTestPvaDevice(prefix: str, with_pvi: bool = False, name: str = '')[source]#

Bases: ophyd_async.epics.testing._devices.EpicsTestCaDevice

Device for use in a pv access test IOC.

int8a: Annotated[SignalRW[Array1D[int8]], PvSuffix('int8a')]#: None

uint16a: Annotated[SignalRW[Array1D[uint16]], PvSuffix('uint16a')]#: None

uint32a: Annotated[SignalRW[Array1D[uint32]], PvSuffix('uint32a')]#: None

int64a: Annotated[SignalRW[Array1D[int64]], PvSuffix('int64a')]#: None

uint64a: Annotated[SignalRW[Array1D[uint64]], PvSuffix('uint64a')]#: None

table: Annotated[SignalRW[EpicsTestTable], PvSuffix('table')]#: None

ntndarray: Annotated[SignalR[ndarray], PvSuffix('ntndarray')]#: None

class ophyd_async.epics.testing.EpicsTestPviDevice(prefix: str, with_pvi: bool = False, name: str = '')[source]#

Bases: ophyd_async.epics.core.EpicsDevice

Device for use in a generic PVI test IOC, independent of PandA/FastCS.

Construct with with_pvi=True: signals here are plain type annotations with no PvSuffix (discovered and filled in from the IOC’s PVI structure by PviDeviceConnector at connect time), except overridden_float, which deliberately also carries a PvSuffix pointing at a different record — this checks that the PVI-supplied PV wins once connected.

Most fields (mbb_direct_bit_r, a_float, table, ntndarray, go) reuse the same attribute names, PVI entries and backing records as EpicsTestCaDevice/EpicsTestPvaDevice, rather than declaring new synthetic ones. Only wo_float (no write-only field exists on those devices to reuse) and overridden_float (inherently synthetic by design, see above) are bespoke to this device.

mbb_direct_bit_r: SignalR[bool]#: None

a_float: SignalRW[float]#: None

wo_float: SignalW[float]#: None

overridden_float: Annotated[SignalRW[float], PvSuffix('float_prec_1')]#: None

table: SignalRW[EpicsTestTable]#: None

ntndarray: SignalR[ndarray]#: None

go: TriggerableCommand#: None

class ophyd_async.epics.testing.EpicsTestPviLeafDevice(name: str = '', connector: DeviceConnector | None = None)[source]#

Bases: ophyd_async.core.Device

Leaf sub-device for child and device_vector on EpicsTestPviNestedDevice.

signal_rw: SignalRW[int]#: None

signal_x: TriggerableCommand#: None

class ophyd_async.epics.testing.EpicsTestPviNestedDevice(prefix: str, with_pvi: bool = False, name: str = '')[source]#

Bases: ophyd_async.epics.core.EpicsDevice

Real-IOC-backed structural PVI test device.

Exercises PviDeviceConnector’s handling of nested sub-devices, DeviceVector of devices, DeviceVector of signals, DeviceVector of commands, and an Optional signal absent from the served tree – the structural mechanics that tests/unit_tests/epics/pvi/test_pvi.py’s Block1-Block5 classes exercise only against a mocked PVI tree. Construct with with_pvi=True.

The served PVI tree also includes an extra_devices DeviceVector that is deliberately not annotated here, to prove that PviDeviceConnector fills in undeclared DeviceVectors of devices dynamically.

Legacy list-based vector encoding (PviTree._handle_legacy_entry) is deliberately not covered here: it’s exclusively produced by pandablocks-ioc’s own hand-rolled PVA server, not something a QSRV2/db-file-based IOC can construct declaratively, so it stays covered by the existing mock-only unit test instead.

signal_rw: SignalRW[int]#: None

signal_x: TriggerableCommand#: None

child: EpicsTestPviLeafDevice#: None

device_vector: DeviceVector[EpicsTestPviLeafDevice]#: None

signal_vector: DeviceVector[SignalRW[float]]#: None

command_vector: DeviceVector[TriggerableCommand]#: None

optional_signal: SignalRW[int] | None#: None

class ophyd_async.epics.testing.EpicsTestPviNestedDeviceMissingChild(prefix: str, with_pvi: bool = False, name: str = '')[source]#

Bases: ophyd_async.epics.core.EpicsDevice

Connects to the same served PVI tree as EpicsTestPviNestedDevice.

Declares an extra required (non-Optional) sub-device the server doesn’t actually provide – demonstrates that connecting raises a clear RuntimeError when a required field can’t be resolved via PVI (the mirror image of optional_signal above, which is allowed to be absent).

signal_rw: SignalRW[int]#: None

missing_child: EpicsTestPviLeafDevice#: None

class ophyd_async.epics.testing.EpicsTestTable(**kwargs)[source]#

Bases: ophyd_async.core.Table

An abstraction of a Table where each field is a column.

For example:

>>> from ophyd_async.core import Table, Array1D
>>> import numpy as np
>>> from collections.abc import Sequence
>>> class MyTable(Table):
...     a: Array1D[np.int8]
...     b: Sequence[str]
...
>>> t = MyTable(a=[1, 2], b=["x", "y"])
>>> len(t)  # the length is the number of rows
2
>>> t2 = t + t  # adding tables together concatenates them
>>> t2.a
array([1, 2, 1, 2], dtype=int8)
>>> t2.b
['x', 'y', 'x', 'y']
>>> t2[1]  # slice a row
array([(2, b'y')], dtype=[('a', 'i1'), ('b', 'S40')])

a_bool: Array1D[bool_]#: None

a_int: Array1D[int32]#: None

a_float: Array1D[float64]#: None

a_str: Sequence[str]#: None

a_enum: Sequence[EpicsTestEnum]#: None

ophyd_async.epics.testing.generate_random_pv_prefix() → str[source]#: For generating random PV names in test devices.

ophyd_async.epics.testing.start_ioc(script: Path, *args: str) → ManagedSubprocess[source]#

Start an EPICS IOC subprocess by running script with args.

Mirrors ophyd_async.tango.testing.start_tango_device_servers’s shape and subprocess handling: rather than building the IOC’s full argv inline, this always launches a real, independently-runnable file - the exact command a human would type by hand, so there’s no separate “does this even work standalone” code path to keep correct.

Parameters:

script – The script implementing the IOC topology to host, e.g. ophyd_async.epics.demo.IOC or ophyd_async.epics.testing.IOC.
args – Positional args for that script’s __main__, e.g. prefix or prefix, str(num_channels) - whatever it expects. Overriding which executable actually hosts the IOC (defaulting to the bundled epicscorelibs.ioc) is a command-line-only concern of that script itself - pass --softioc ... as one of args if you need that (see the script’s own --help); this function doesn’t thread it through.

ophyd_async.epics.testing#

Package Contents#

Classes#

Functions#

Data#

API#

`ophyd_async.epics.testing`#