Type Hints

Common types used through out ska-tango-base.

class ska_tango_base.type_hints.BusProtocol

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

register_observer(observer: ObserverProtocol) → None

Register the observer to be notified is emitted for any signal.

emit(signal: str, value: Any, *, store: bool = False) → None

Emit a new value for the signal.

get_last_emitted_value(signal: str) → Any

Get the last emission that was stored for the given signal.

delete_last_emitted_value(signal: str) → None

Delete the last emission that was stored for the given signal.

wait_for_thread(timeout: float | None = 5.0) → None

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

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 ska_tango_base.type_hints.CommandTrackerProtocol

All things to do with commands.

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

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

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

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 ska_tango_base.type_hints.LRCCallbackType

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

Call the callback.

__init__(*args, **kwargs)

class ska_tango_base.type_hints.LRCSubscriptionsProtocol

LRC event subscriptions that is returned by invoke_lrc.

Unsubscribes from all events when the instance is deleted.

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

Unsubscribe from LRC attributes’ events.

__init__(*args, **kwargs)

class ska_tango_base.type_hints.ObserverProtocol

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

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 ska_tango_base.type_hints.ProgressCallbackType

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

Call the callback.

__init__(*args, **kwargs)

class ska_tango_base.type_hints.SharingObserverProtocol

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 ska_tango_base.type_hints.TaskCallbackType

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

Call the callback.

__init__(*args, **kwargs)

class ska_tango_base.type_hints.TaskFunctionType

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).

Note

Due to limitations of the python 3.10 type system, this protocol is not used for type checking. However, this protocol describes the interface assumed by ska-tango-base.

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

Run the task.

__init__(*args, **kwargs)

class ska_tango_base.type_hints.SimpleTaskFunctionType

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.

Note

Due to limitations of the python 3.10 type system, this protocol is not used for type checking. However, this protocol describes the interface assumed by ska-tango-base.

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

Run the task.

__init__(*args, **kwargs)

class ska_tango_base.type_hints.TaskExecutorProtocol

Protocol for a TaskExecutor.

property max_executing_tasks: int

Get the maximum number of simultaneously executing tasks.

property max_queued_tasks: int

Get the maximum task queue size.

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

Submit a new task.

Parameters:

• func – the task function to be executed.

• args – positional arguments to the task function

• kwargs – keyword arguments to the task function

• is_cmd_allowed – sanity check for task execution

• 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]

Tell this executor to abort execution.

Parameters:

task_callback – callback for abort complete

Returns:

tuple of task status & message

shutdown() → None

Permanently shutdown the executor.

__init__(*args, **kwargs)

ska_tango_base.type_hints.EventDataType: TypeAlias = tango._tango.EventData | tango._tango.AttrConfEventData | tango._tango.DataReadyEventData | tango._tango.DevIntrChangeEventData

Tango event data.

ska_tango_base.type_hints.EventCallbackCallable

A Tango event callback using the __call__ operator.

class ska_tango_base.type_hints.EventCallbackPushEventProtocol

A Tango event callback using the push_event method.

__init__(*args, **kwargs)

push_event(event: EventData | AttrConfEventData | DataReadyEventData | DevIntrChangeEventData) → None

Process the event data.

ska_tango_base.type_hints.EventCallbackType: TypeAlias = collections.abc.Callable[[tango._tango.EventData | tango._tango.AttrConfEventData | tango._tango.DataReadyEventData | tango._tango.DevIntrChangeEventData], None] | ska_tango_base.type_hints.EventCallbackPushEventProtocol

An object that can be used to process Tango events.

class ska_tango_base.type_hints.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 ska_tango_base.type_hints.DevVarLongStringArrayType

A type hint for the DevVarLongStringArray Tango data type.

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

class ska_tango_base.type_hints.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