Type Hints

Common types used through out ska-tango-base.

class BusProtocol[source]

A bus which emits signals and stores the most recently emitted value.

register_observer(observer: ObserverProtocol) → None[source]: Register the observer to be notified is emitted for any signal.

emit(signal: str, value: Any) → None[source]: Emit a new value for the signal.

wait_for_thread(timeout: float | None = 5.0) → None[source]

Synchronise the calling thread with the background thread.

When this function returns, all emissions sequenced before this call will have been processed by the background thread.

wait_for_signal_value(signal: str, conditional: Callable[[Any], bool] | None = None, timeout: float | None = None) → bool[source]

Wait for a value to be emitted for signal where conditional returns True.

Whenever a value is emitted for the signal, conditional(old_value, new_value) is called. The thread calling this method will block until conditional returns True or the timeout is reached. If conditional is None, this method will unblock on any value emitted for the signal.

__init__(*args, **kwargs)

class CommandTrackerProtocol[source]

All things to do with commands.

new_command(command_name: str, completed_callback: Callable[[], None] | None = None, started_callback: Callable[[], None] | None = None) → str[source]

Create a new command.

Parameters:

command_name – the command name
completed_callback – an optional callback for command completion
started_callback – an optional callback for when a command starts

Returns:

a unique command id

update_command_info(command_id: str, status: TaskStatus | None = None, progress: int | None = None, result: tuple[ResultCode, str] | None = None, exception: Exception | None = None) → None[source]

Update status information on the command.

Parameters:

command_id – the unique command id
status – the status of the asynchronous task
progress – the progress of the asynchronous task
result – the result of the completed asynchronous task
exception – any exception caught in the running task

get_command_status(command_id: str) → TaskStatus[source]

Return the current status of a running command.

Parameters:: command_id – the unique command id
Returns:: a status of the asynchronous task.

property commands_in_queue: list[tuple[str, str]]

Return a list of commands in the queue.

Returns:: a list of (command_id, command_name) tuples, ordered by when invoked.

property command_statuses: list[tuple[str, TaskStatus]]

Return a list of command statuses.

Returns:: a list of (command_id, status) tuples, ordered by when invoked.

property command_progresses: list[tuple[str, int | str]]

Return a list of command progresses for commands in the queue.

Returns:: a list of (command_id, progress) tuples, ordered by when invoked.

property command_result: tuple[str, JSONData] | None

Return the result of the most recently completed command.

Returns:: a (command_id, result) tuple. If no command has completed yet, then None.

property command_exception: tuple[str, Exception] | None

Return the most recent exception, if any.

Returns:: a (command_id, exception) tuple. If no command has raised an uncaught exception, then None.

__init__(*args, **kwargs)

class LRCCallbackType[source]

Expected LRC callback signature.

The callback will be called with some combination of the following arguments:

status: TaskStatus

progress: int

result: Any

error: tuple[DevError]

Each of the above arguments is optional and the callback must check which are present by testing them for None. The callback cannot assume that only one argument will be provided per call.

It must accept a generic **kwargs parameter for forwards compatibility.

__call__(**kwargs: Any) → None[source]: Call the callback.

__init__(*args, **kwargs)

class LRCSubscriptionsProtocol[source]

LRC event subscriptions that is returned by invoke_lrc.

Unsubscribes from all events when the instance is deleted.

__init__(*args, **kwargs)

property command_id: str

The command ID.

Returns:: the command ID.

property protocol_version: int

The LRC client-server protocol version used.

Returns:: the protocol version.

unsubscribe_lrc_events() → None[source]: Unsubscribe from LRC attributes’ events.

class ObserverProtocol[source]

An object which observes signals being emitted on a bus.

The observer must be registered with the bus first with BusProtocol.register_observer().

notify_emission(signal: str, value: Any) → None[source]

Notify the observer of an emission asynchronously.

While this function is being called a logger is available at current_signal_bus_logger.get() which should be used (sparingly) for reporting problems.

__init__(*args, **kwargs)

class ProgressCallbackType[source]

Structural subtyping protocol for a progress_callback.

A progress_callback should be called with a single integer indicating the progress of the task.

__call__(progress: int) → None[source]: Call the callback.

__init__(*args, **kwargs)

class SharingObserverProtocol[source]

An object with access to an shared bus.

shared_bus: BusProtocol: Bus to emit signals on.

property observer_prefix: str: Path from the root of the bus to this object.

__init__(*args, **kwargs)

class TaskCallbackType[source]

Structural subtyping protocol for a task_callback.

A task_callback will be called with some combination of the following arguments:

status: TaskStatus of the task.

progress: int progress of the task.

result: JSON serialisable result of the task.

exception: Exception raised from the task.

Each of the above arguments is optional and the callback must check which are present by testing them for None. The callback cannot assume that only one argument will be provided per call.

__call__(status: TaskStatus | None = None, progress: int | None = None, result: JSONData | None = None, exception: Exception | None = None) → None[source]: Call the callback.

__init__(*args, **kwargs)

class TaskFunctionType[source]

Perform some task, managing the task’s status.

The task should call task_callback(status=IN_PROGRESS) when it starts and task_callback(status=COMPLETED, result=<JSONData>) when it finishes normally.

It can periodically call task_callback(progress=<int>) to update clients to its progress. Also, the task must periodically check the task_abort_event and promptly abort if it is set, returning once cleanup has been completed. To acknowledge that the task has been aborted, the task should call task_callback(status=ABORTED, result=<JSONData).

Parameters:

task_callback – callback to notify when the task makes progress and of status changes
task_abort_event – set whenever the task is requested to abort

__call__(*args: ~typing.~P, task_callback: ~ska_tango_base.type_hints.TaskCallbackType | None, task_abort_event: ~threading.Event | None, **kwargs: ~typing.~P) → None[source]: Run the task.

__init__(*args, **kwargs)

class SimpleTaskFunctionType[source]

Perform some task without managing the task’s status.

This task periodically calls progress_callback with a value from 0-100 indicating the percentage completion of the task, and will periodically check the task_abort_event. If the task_abort_event is set the task will abort itself and raise TaskAborted.

With the exception of TaskAborted, this task does not raise an exception.

Parameters:

progress_callback – callback to notify when the task makes progress
task_abort_event – set whenever the task is requested to abort

Raises:

TaskAborted – when task_abort_event is set

Returns:

a JSON encodable result indicating the success or failure of the task

__call__(*args: ~typing.~P, progress_callback: ~ska_tango_base.type_hints.ProgressCallbackType, task_abort_event: ~threading.Event, **kwargs: ~typing.~P) → JSONData[source]: Run the task.

__init__(*args, **kwargs)

class TaskExecutorProtocol[source]

Protocol for a TaskExecutor.

property max_executing_tasks: int: Get the maximum number of simultaneously executing tasks.

property max_queued_tasks: int: Get the task queue size.

submit(func: TaskFunctionType, args: Any | None = None, kwargs: Any | None = None, is_cmd_allowed: Callable[[], bool] | None = None, task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Submit a new task.

Parameters:

func – the function to be executed.
args – positional arguments to the function
kwargs – keyword arguments to the function
is_cmd_allowed – sanity check for func
task_callback – the callback to be called when the status or progress of the task execution changes

Returns:

(TaskStatus, message)

abort(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Tell this executor to abort execution.

Parameters:: task_callback – callback for abort complete
Returns:: tuple of task status & message

shutdown() → None[source]: Permanently shutdown the executor.

__init__(*args, **kwargs)

class JSONData

A type hint for any JSON-encodable data.

JSONData = (
   None
   | bool
   | int
   | float
   | str
   | list["JSONData"]  # A list can contain more JSON-encodable data
   | dict[str, "JSONData"]  # A dict must have str keys and JSON-encodable data
   | tuple["JSONData", ...]  # A tuple can contain more JSON-encodable data
)

class DevVarLongStringArrayType

A type hint for the DevVarLongStringArray Tango data type.

DevVarLongStringArrayType = tuple[list[ResultCode], list[str]]

class ReadAttrType

A type hint for attribute read methods which may return a Tango data triple.

AttrT = TypeVar("AttrT")
ReadAttrType = AttrT | tuple[AttrT, float, AttrQuality]

Usage:

class MyDevice(Device):
   @attribute(dtype="DevLong")
   def myAttr(self) -> ReadAttrType[int]:
      return 0, time.now(), AttrQuality.ATTR_WARNING