Skip to content

redsun.device.sharedbox

BoxDeviceConnector 

BoxDeviceConnector(
    prefix: str = "", *, declared_only: bool = False
)

Bases: DeviceConnector

Connect a device to the boxes of a service.

The top device attaches the index and the boxes of the service and fills itself and its children from what the index describes. A child gets its part of the description from its parent.

Parameters:

Name Type Description Default
declared_only bool

Fill only the signals, commands and children the device declares, and skip the rest of what the service serves. Declared children follow the same rule, and a declared DeviceVector takes every entry the service serves.

 False

create_children_from_annotations 

create_children_from_annotations(device: Device) -> None

Create the signals, commands and children the device declares.

connect_mock async 

connect_mock(
    device: Device, mock: DeviceMock[Device]
) -> None

Connect with mock backends, needing no service.

connect_real async 

connect_real(
    device: Device, timeout: float, force_reconnect: bool
) -> None

Attach the boxes of the service, fill the device from them and connect it.

BoxSignalBackend 

BoxSignalBackend(datatype: type[SignalDatatypeT] | None)

Bases: SignalBackend[SignalDatatypeT]

A signal backed by one attribute of a controller box.

source 

source(name: str, read: bool) -> str

Return the location of the attribute.

connect async 

connect(timeout: float) -> None

Check the declared datatype against the description.

Raises:

Type Description
TypeError

The declared datatype is not what the service serves.

put async 

put(value: SignalDatatypeT | None) -> None

Write value and wait until the service has put it.

Raises:

Type Description
TypeError

value is None.
BoxPutError

The service refused the value or its put raised.

get_datakey async 

get_datakey(source: str) -> DataKey

Return the data key, with the units, precision, limits and choices served.

get_reading async 

get_reading() -> Reading[SignalDatatypeT]

Return the current value.

get_value async 

get_value() -> SignalDatatypeT

Return the current value.

get_setpoint async 

get_setpoint() -> SignalDatatypeT

Return the last value written.

set_callback 

set_callback(
    callback: Callback[Reading[SignalDatatypeT]] | None,
) -> None

Call callback with the current reading, then with each new one.

BoxCommandBackend 

BoxCommandBackend()

Bases: CommandBackend[[], None]

A command backed by the request fields of a controller box.

source 

source(name: str) -> str

Return the location of the command.

connect async 

connect(timeout: float) -> None

Do nothing: the box is attached with the device.

execute async 

execute() -> None

Run the command and wait until the service has finished it.

Raises:

Type Description
BoxPutError

The command raised in the service.

BoxPutError 

BoxPutError(attribute: str, message: str)

Bases: Exception

A service refused a write or a command, or it raised while running it.

attribute instance-attribute 

attribute: str = attribute

Name of the attribute or command the service refused.

stream_name 

stream_name(prefix: str, part: str) -> str

Return the name of a stream the service publishes outside the controller boxes.

Raises:

Type Description
ValueError

If part is fastcs, which the controller boxes use, or holds a character other than a letter, a digit, _ or -.