Control Layers

Main control layer objects. Exposes basic communication operations, and dispatches to various shims depending on the context.

class squirrel.control_layer.core.ControlLayer(*args, shims: List[str] | None = None, **kwargs)

Control Layer used to communicate with the control system, dispatching to whichever shim is relevant.

Methods

get(...)	Get the value(s) in address.
put(-> ~squirrel.control_layer.status.TaskStatus)	Put value to address If address is a list, value and cb must be lists of equal length
shim_from_pv(address)	Determine the correct shim to use for the provided address.
subscribe(address, cb)	Subscribes a callback (cb) to the provide address (address)

get(address: str | Iterable[str]) → EpicsData | Iterable[EpicsData]

get(address: str) → EpicsData | CommunicationError

get(address: Iterable) → Iterable[EpicsData | CommunicationError]

Get the value(s) in address. If a single pv is provided, will return a single value. If a list of pvs is provided, will get the values for each asynchronously.

Parameters:

addressUnion[str, Iterable[str]]

The PV(s) to get values for.

Returns:

Union[EpicsData, Iterable[EpicsData]]

The requested data

put(address: str | list[str], value: Any | list[Any], cb: Callable | list[Callable] | None = None) → TaskStatus | list[TaskStatus]

put(address: str, value: Any, cb: Callable | None = None) → TaskStatus

put(address: list, value: list, cb: list[Callable] | None = None) → list[TaskStatus]

Put value to address If address is a list, value and cb must be lists of equal length

Parameters:

addressUnion[str, list[str]]

The PV(s) to put values to

valueUnion[Any, list[Any]]

The value(s) to put to the address

cbOptional[Callable], by default None

Callbacks to run on completion of the put task. Callbacks will be called with the associated TaskStatus as its sole argument

Returns:

Union[TaskStatus, list[TaskStatus]]

The TaskStatus object(s) for the put operation

shim_from_pv(address: str) → _BaseShim

Determine the correct shim to use for the provided address. address can optionally hold a protocol defining prefix such as “ca://” or “pva://”. If no prefix is provided, will select the first available shim.

Parameters:

addressstr

a PV address such as “MY:PREFIX:mtr1” or “pva://MY:PREFIX:dt”

Returns:

_BaseShim

The shim held by this ControlLayer for address’s protocol

Raises:

ValueError

If address cannot be recognized or a matching shim cannot be found

subscribe(address: str, cb: Callable)

Subscribes a callback (cb) to the provide address (address)

class squirrel.control_layer.status.TaskStatus(awaitable: Awaitable)

Unified Status object for wrapping task completion information and attaching callbacks. This must be created inside of a coroutine, but can be returned to synchronous scope for examining the task.

Awaiting this status is similar to awaiting the wrapped task.

Largely vendored from bluesky/ophyd-async

Attributes:

done

success

Methods

wait([timeout])	Block until the coroutine finishes.
wrap(f)	Wrap an async function in a TaskStatus.

add_callback
exception

wait(timeout=None) → None

Block until the coroutine finishes. Raises asyncio.TimeoutError if the timeout elapses before the task is completed

To be called in a synchronous context, if the status has not been awaited

Parameters:

timeoutnumber, optional

timeout in seconds, by default None

Raises:

asyncio.TimeoutError

classmethod wrap(f: Callable[[...], Awaitable]) → Callable[[...], TS]

Wrap an async function in a TaskStatus.