pyaml.control.abstract_impl#

Control-system implementations of PyAML read/write interfaces.

The classes in this module adapt control-system devices to PyAML scalar and array interfaces, including conversions between hardware values, magnet strengths, BPM readings, and RF quantities.

Functions

`check_range`(values, dev_range)	Check whether values are within given ranges.
`format_out_of_range_message`(values, devs, *)	Build a user-friendly error message for out-of-range values.

Classes

`CSScalarAggregator`(devs)	Aggregate scalar control-system devices into one PyAML interface.
`CSStrengthScalarAggregator`(peer)	Aggregate magnet strengths while avoiding duplicate hardware writes.
`RBetatronTuneArray`(tune_monitor, devs)	Expose horizontal and vertical betatron tunes as a read-only array.
`RBpmArray`(hDev, vDev)	Expose horizontal and vertical BPM positions as an array.
`RWBpmOffsetArray`(hDev, vDev)	Expose horizontal and vertical BPM offsets as a writable array.
`RWBpmTiltScalar`(dev)	Class providing read access to a BPM tilt of a control system
`RWHardwareArray`(model, devs)	Expose multiple magnet hardware setpoints as an array interface.
`RWHardwareScalar`(model, dev)	Expose one magnet hardware setpoint as a readable/writable scalar.
`RWRFFrequencyScalar`(rf, dev)	Class providing read write access to RF frequency of a control system.
`RWRFPhaseScalar`(transmitter, dev)	Class providing read write access to cavity phase for a transmitter of a control system.
`RWRFVoltageScalar`(transmitter, dev)	Class providing read write access to cavity voltage for a transmitter of a control system.
`RWStrengthArray`(model, devs)	Expose multiple magnet strengths with hardware-value conversion.
`RWStrengthScalar`(model, dev)	Expose one magnet strength with hardware-value conversion.

class pyaml.control.abstract_impl.CSScalarAggregator(devs)#

Bases: ScalarAggregator

Aggregate scalar control-system devices into one PyAML interface.

Parameters:: devs (DeviceAccessList) – Devices managed by the aggregator.

Methods

`add_devices()`	Add one or more devices to the scalar aggregator.
`set()`	Write scalar values to the managed control-system devices.
`set_and_wait()`	Set values and wait for readback confirmation.
`get()`	Read the current values from all managed devices.
`readback()`	Read the last available values from all managed devices.
`unit()`	Return the unit reported by the managed devices.
`nb_device()`	Return the number of managed devices.

add_devices(devices)#

Add one or more devices to the scalar aggregator.

Parameters:: devices (DeviceAccess | list[DeviceAccess]) – Control-system device or devices to manage.

get()#

Read the current values from all managed devices.

nb_device()#

Return the number of managed devices.

readback()#

Read the last available values from all managed devices.

set(value)#

Write scalar values to the managed control-system devices.

Parameters:: value (NDArray[np.float64]) – Values to write, in the same order as the managed devices.

set_and_wait(value)#

Set values and wait for readback confirmation.

This control-system implementation does not currently support waiting for device readback and raises NotImplementedError.

Parameters:: value (NDArray[np.float64]) – Values that would be written to the managed devices.
Raises:: NotImplementedError – Always raised because asynchronous setpoint confirmation is not implemented.

unit()#

Return the unit reported by the managed devices.

class pyaml.control.abstract_impl.CSStrengthScalarAggregator(peer)#

Bases: CSScalarAggregator

Aggregate magnet strengths while avoiding duplicate hardware writes.

Magnet models convert between exposed strengths and hardware setpoints. Shared models, such as those used by virtual magnets from combined-function magnets, are written only once per underlying power supply.

Parameters:: peer (CSScalarAggregator) – Scalar device aggregator containing the hardware devices.

Methods

`add_magnet()`	Register a magnet and its hardware devices with the aggregator.
`set()`	Convert strengths to hardware values and write the setpoints.
`set_and_wait()`	Set magnet strengths and wait for hardware readback.
`get()`	Read the current values from the underlying devices.
`readback()`	Read back magnet strengths from measured hardware values.
`unit()`	Return the units associated with the underlying devices.

add_magnet(magnet, devs)#

Parameters:

magnet (Magnet) – Magnet whose strength is being aggregated.
devs (list[DeviceAccess]) – Hardware devices associated with the magnet model.

get()#

Read the current values from the underlying devices.

readback()#

Read back magnet strengths from measured hardware values.

Returns:: numpy.ndarray – Magnet strengths corresponding to the latest device readbacks.
Return type:: array

set(value)#

Convert strengths to hardware values and write the setpoints.

Parameters:: value (NDArray[np.float64]) – Magnet strengths, ordered according to the registered magnets.

set_and_wait(value)#

Set magnet strengths and wait for hardware readback.

Waiting for readback is not implemented by this aggregator.

Parameters:: value (NDArray[np.float64]) – Magnet strengths to convert and write.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the units associated with the underlying devices.

class pyaml.control.abstract_impl.RBetatronTuneArray(tune_monitor, devs)#

Bases: ReadFloatArray

Expose horizontal and vertical betatron tunes as a read-only array.

Parameters:

tune_monitor (object) – Tune monitor configuration supplying the tune unit.
devs (list[DeviceAccess]) – Devices providing horizontal and vertical tune measurements.

Methods

`get()`	Return horizontal and vertical betatron tunes.
`unit()`	Return the configured betatron-tune unit.

get()#

Return horizontal and vertical betatron tunes.

unit()#

Return the configured betatron-tune unit.

class pyaml.control.abstract_impl.RBpmArray(hDev, vDev)#

Bases: ReadFloatArray

Expose horizontal and vertical BPM positions as an array.

Parameters:

hDev (DeviceAccess) – Device providing the horizontal BPM position.
vDev (DeviceAccess) – Device providing the vertical BPM position.

Methods

`get()`	Return horizontal and vertical BPM positions.
`unit()`	Return the unit reported by the BPM device.

get()#

Return horizontal and vertical BPM positions.

unit()#

Return the unit reported by the BPM device.

class pyaml.control.abstract_impl.RWBpmOffsetArray(hDev, vDev)#

Bases: ReadWriteFloatArray

Expose horizontal and vertical BPM offsets as a writable array.

Parameters:

hDev (DeviceAccess) – Device handle for the horizontal BPM offset.
vDev (DeviceAccess) – Device handle for the vertical BPM offset.

Methods

`get()`	Return horizontal and vertical BPM offsets.
`set()`	Write horizontal and vertical BPM offsets.
`set_and_wait()`	Set BPM offsets and wait for readback confirmation.
`unit()`	Return the unit shared by the BPM offset devices.

get()#

Return horizontal and vertical BPM offsets.

set(value)#

Write horizontal and vertical BPM offsets.

Parameters:: value (NDArray[np.float64]) – Two-element array containing horizontal and vertical offsets.

set_and_wait(value)#

Set BPM offsets and wait for readback confirmation.

Parameters:: value (NDArray[np.float64]) – Two-element array containing horizontal and vertical offsets.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the unit shared by the BPM offset devices.

class pyaml.control.abstract_impl.RWBpmTiltScalar(dev)#

Bases: ReadFloatScalar

Class providing read access to a BPM tilt of a control system

Parameters:: dev (DeviceAccess) – Device handle giving access to the BPM tilt attribute.

Methods

`get()`	Return horizontal and vertical BPM positions.
`set()`	Write the BPM tilt value to the control-system device.
`set_and_wait()`	Set the BPM tilt and wait for readback confirmation.
`unit()`	Return the unit shared by the BPM devices.

get()#

Return horizontal and vertical BPM positions.

set(value)#

Write the BPM tilt value to the control-system device.

Parameters:: value (float) – BPM tilt value in the device’s configured units.

set_and_wait(value)#

Set the BPM tilt and wait for readback confirmation.

Parameters:: value (NDArray[np.float64]) – BPM tilt value to write.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the unit shared by the BPM devices.

class pyaml.control.abstract_impl.RWHardwareArray(model, devs)#

Bases: ReadWriteFloatArray

Expose multiple magnet hardware setpoints as an array interface.

Parameters:

model (MagnetModel) – Magnet model defining the hardware units.
devs (list[DeviceAccess]) – Control-system devices holding the hardware setpoints.

Methods

`get()`	Return the current hardware values in device order.
`set()`	Validate and write hardware values for all devices.
`set_and_wait()`	Set hardware values and wait for readback confirmation.
`unit()`	Return the hardware units defined by the magnet model.

get()#

Return the current hardware values in device order.

set(value)#

Validate and write hardware values for all devices.

Parameters:: value (np.array) – Hardware values ordered to match self.__devs.
Raises:: PyAMLException – If any value is outside its device’s configured range.

set_and_wait(value)#

Set hardware values and wait for readback confirmation.

Parameters:: value (np.array) – Hardware values to write in device order.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the hardware units defined by the magnet model.

class pyaml.control.abstract_impl.RWHardwareScalar(model, dev)#

Bases: ReadWriteFloatScalar

Expose one magnet hardware setpoint as a readable/writable scalar.

Parameters:

model (MagnetModel) – Magnet model used to determine the hardware unit.
dev (DeviceAccess) – Control-system device holding the hardware setpoint.

Methods

`get()`	Return the current hardware setpoint.
`set()`	Validate and write a magnet hardware setpoint.
`set_and_wait()`	Set the hardware value and wait for readback confirmation.
`unit()`	Return the hardware unit defined by the magnet model.
`set_magnet_rigidity()`	Set the beam rigidity used by the magnet model.

get()#

Return the current hardware setpoint.

set(value)#

Validate and write a magnet hardware setpoint.

Parameters:: value (float) – Hardware value to write to the control-system device.
Raises:: PyAMLException – If value is outside the device’s configured range.

set_and_wait(value)#

Set the hardware value and wait for readback confirmation.

Parameters:: value (double) – Hardware value to write.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

set_magnet_rigidity(brho)#

Set the beam rigidity used by the magnet model.

Parameters:: brho (np.double) – Magnetic rigidity in tesla metres.

unit()#

Return the hardware unit defined by the magnet model.

class pyaml.control.abstract_impl.RWRFFrequencyScalar(rf, dev)#

Bases: ReadWriteFloatScalar

Class providing read write access to RF frequency of a control system.

Parameters:

rf (RFPlant) – RF plant whose configuration supplies the frequency unit.
dev (DeviceAccess) – Control-system device holding the RF frequency.

Methods

`get()`	Return the current RF frequency.
`set()`	Write an RF-frequency setpoint to the plant device.
`set_and_wait()`	Set the RF frequency and wait for readback confirmation.
`unit()`	Return the configured RF-frequency unit.

get()#

Return the current RF frequency.

set(value)#

Write an RF-frequency setpoint to the plant device.

Parameters:: value (float) – RF-frequency setpoint in the configured frequency unit.

set_and_wait(value)#

Set the RF frequency and wait for readback confirmation.

Parameters:: value (float) – RF-frequency setpoint.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the configured RF-frequency unit.

class pyaml.control.abstract_impl.RWRFPhaseScalar(transmitter, dev)#

Bases: ReadWriteFloatScalar

Class providing read write access to cavity phase for a transmitter of a control system.

Parameters:

transmitter (RFTransmitter) – RF transmitter whose configuration supplies the phase unit.
dev (DeviceAccess) – Control-system device holding the cavity phase.

Methods

`get()`	Return the current cavity phase.
`set()`	Write a cavity-phase setpoint to the transmitter device.
`set_and_wait()`	Set the cavity phase and wait for readback confirmation.
`unit()`	Return the configured cavity-phase unit.

get()#

Return the current cavity phase.

set(value)#

Write a cavity-phase setpoint to the transmitter device.

Parameters:: value (float) – Cavity-phase setpoint in the configured phase unit.

set_and_wait(value)#

Set the cavity phase and wait for readback confirmation.

Parameters:: value (float) – Cavity-phase setpoint.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the configured cavity-phase unit.

class pyaml.control.abstract_impl.RWRFVoltageScalar(transmitter, dev)#

Bases: ReadWriteFloatScalar

Class providing read write access to cavity voltage for a transmitter of a control system.

Parameters:

transmitter (RFTransmitter) – RF transmitter whose configuration supplies the voltage unit.
dev (DeviceAccess) – Control-system device holding the cavity-voltage value.

Methods

`get()`	Return the current cavity voltage.
`set()`	Write a cavity-voltage setpoint to the transmitter device.
`set_and_wait()`	Set the cavity voltage and wait for readback confirmation.
`unit()`	Return the configured cavity-voltage unit.

get()#

Return the current cavity voltage.

set(value)#

Write a cavity-voltage setpoint to the transmitter device.

Parameters:: value (float) – Cavity-voltage setpoint in the configured voltage unit.

set_and_wait(value)#

Set the cavity voltage and wait for readback confirmation.

Parameters:: value (float) – Cavity-voltage setpoint.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the configured cavity-voltage unit.

class pyaml.control.abstract_impl.RWStrengthArray(model, devs)#

Bases: ReadWriteFloatArray

Expose multiple magnet strengths with hardware-value conversion.

Parameters:

model (MagnetModel) – Magnet model used for strength conversion and units.
devs (list[DeviceAccess]) – Control-system devices corresponding to the model’s hardware values.

Methods

`get()`	Read hardware values and convert them to magnet strengths.
`set()`	Convert strengths to hardware values and write the setpoints.
`set_and_wait()`	Set magnet strengths and wait for readback confirmation.
`unit()`	Return the strength units defined by the magnet model.

get()#

Read hardware values and convert them to magnet strengths.

set(value)#

Convert strengths to hardware values and write the setpoints.

Parameters:: value (np.array) – Magnet strengths ordered to match self.__devs.

set_and_wait(value)#

Set magnet strengths and wait for readback confirmation.

Parameters:: value (np.array) – Magnet strengths to convert and write.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

unit()#

Return the strength units defined by the magnet model.

class pyaml.control.abstract_impl.RWStrengthScalar(model, dev)#

Bases: ReadWriteFloatScalar

Expose one magnet strength with hardware-value conversion.

Parameters:

model (MagnetModel) – Magnet model used for strength conversion and units.
dev (DeviceAccess) – Control-system device holding the corresponding hardware value.

Methods

`get()`	Read the hardware value and convert it to magnet strength.
`set()`	Convert and write a magnet strength setpoint.
`set_and_wait()`	Set a magnet strength and wait for readback confirmation.
`unit()`	Return the strength unit defined by the magnet model.
`set_magnet_rigidity()`	Set the beam rigidity used by the magnet model.

get()#

Read the hardware value and convert it to magnet strength.

set(value)#

Convert and write a magnet strength setpoint.

Parameters:: value (float) – Magnet strength to convert to hardware units and write.
Raises:: PyAMLException – If the converted hardware value is outside the device’s configured range.

set_and_wait(value)#

Set a magnet strength and wait for readback confirmation.

Parameters:: value (float) – Magnet strength to convert and write.
Raises:: NotImplementedError – Always raised because asynchronous confirmation is unsupported.

set_magnet_rigidity(brho)#

Set the beam rigidity used by the magnet model.

Parameters:: brho (np.double) – Magnetic rigidity in tesla metres.

unit()#

Return the strength unit defined by the magnet model.

pyaml.control.abstract_impl.check_range(values, dev_range)#

Check whether values are within given ranges.

Inverted semantics:

True -> all checks pass (everything is within bounds)
False -> at least one check fails (out of range)

dev_range format (flat):

[min1, max1, min2, max2, …]

Broadcasting rules:

Let N = number of values, K = number of ranges (pairs). - N == K : one range per value - N == 1 and K > 1: the single value must satisfy ALL ranges - N > 1 and K == 1: the single range applies to ALL values

pyaml.control.abstract_impl.format_out_of_range_message(values, devs, *, header='Values out of range:')#

Build a user-friendly error message for out-of-range values.

Output example:: Values out of range: 110 A, ‘//host/dev/attr’ [10.0, 109.0] 110 A, ‘//host/dev/attr’ [10.0, 109.0]

Notes

Only failing channels are listed.
Supports scalar/array values and DeviceAccess/DeviceAccessList.
Uses check_range() semantics (inclusive bounds, None => unbounded).