Utils

General utilities that may be useful to SKA devices and clients.

ska_tango_base.utils.exception_manager(cls: type[Exception], callback: Callable[[], None] | None = None) → Generator[None, None, None][source]

Return a context manager that manages exceptions.

Parameters:

cls – class type
callback – a callback

Yields:

return a context manager

ska_tango_base.utils.get_dev_info(domain_name: str, device_server_name: str, device_ref: str) → DbDevInfo[source]

Get device info.

Parameters:

domain_name – tango domain name
device_server_name – tango device server name
device_ref – tango device reference

Returns:

database device info instance

ska_tango_base.utils.dp_set_property(device_name: str, property_name: str, property_value: Any) → None[source]

Use a DeviceProxy to set a device property.

Parameters:

device_name – tango device name
property_name – tango property name
property_value – tango property value

ska_tango_base.utils.get_device_group_and_id(device_name: str) → list[str][source]

Return the group and id part of a device name.

Parameters:: device_name – tango device name
Returns:: group & id part of tango device name

ska_tango_base.utils.convert_api_value(param_dict: dict[str, str]) → tuple[str, Any][source]

Validate tango command parameters which are passed via json.

Parameters:

param_dict – parameters

Raises:

TypeError – invalid type
ValueError – value not of specified type

Returns:

tuple(name, value)

ska_tango_base.utils.coerce_value(value: Any) → Any[source]

Coerce tango.DevState values to string, leaving other values alone.

Parameters:: value – a tango DevState
Returns:: DevState as a string

ska_tango_base.utils.get_dp_attribute(device_proxy: DeviceProxy, dp_attribute: tango.server.attribute, with_value: bool = False, with_context: bool = False) → dict[str, Any][source]

Get an attribute from a DeviceProxy.

:param device_proxy:a tango device proxy :param dp_attribute: Attribute :param with_value: default False :param with_context: default False

Returns:: dictionary of attribute info

ska_tango_base.utils.get_dp_command(device_name: str, dp_command: tango.server.command, with_context: bool = False) → dict[str, Any][source]

Get a command from a DeviceProxy.

Parameters:

device_name – tango device name
dp_command – tango command
with_context – default False

Returns:

dictionary of command info

ska_tango_base.utils.get_tango_device_type_id(tango_address: str) → list[str][source]

Return the type id of a TANGO device.

Parameters:: tango_address – tango device address
Returns:: the type id of the device

ska_tango_base.utils.get_groups_from_json(json_definitions: list[str]) → dict[str, Any][source]

Return a dict of tango.Group objects matching the JSON definitions.

Extracts the definitions of groups of devices and builds up matching tango.Group objects. Some minimal validation is done - if the definition contains nothing then None is returned, otherwise an exception will be raised on error.

This function will NOT attempt to verify that the devices exist in the Tango database, nor that they are running.

The definitions would typically be provided by the Tango device property “GroupDefinitions”, available in the SKABaseDevice. The property is an array of strings. Thus a sequence is expected for this function.

Each string in the list is a JSON serialised dict defining the “group_name”, “devices” and “subgroups” in the group. The tango.Group() created enables easy access to the managed devices in bulk, or individually. Empty and whitespace-only strings will be ignored.

The general format of the list is as follows, with optional “devices” and “subgroups” keys:

[
    {"group_name": "<name>", "devices": ["<dev name>", ...]},
    {
        "group_name": "<name>",
        "devices": ["<dev name>", "<dev name>", ...],
        "subgroups" : [{<nested group>}, {<nested group>}, ...]
    },
    ...
]

For example, a hierarchy of racks, servers and switches:

[
    {
        "group_name": "servers",
        "devices": [
            "elt/server/1",
            "elt/server/2",
            "elt/server/3",
            "elt/server/4",
        ],
    },
    {
        "group_name": "switches",
        "devices": ["elt/switch/A", "elt/switch/B"],
    },
    {
        "group_name": "pdus",
        "devices": ["elt/pdu/rackA", "elt/pdu/rackB"],
    },
    {
        "group_name": "racks",
        "subgroups": [
            {
                "group_name": "rackA",
                "devices": [
                    "elt/server/1",
                    "elt/server/2",
                    "elt/switch/A",
                    "elt/pdu/rackA",
                ],
            },
            {
                "group_name": "rackB",
                "devices": [
                    "elt/server/3",
                    "elt/server/4",
                    "elt/switch/B",
                    "elt/pdu/rackB",
                ],
                "subgroups": [],
            },
        ],
    },
]

Parameters:: json_definitions – Sequence of strings, each one a JSON dict with keys “group_name”, and one or both of: “devices” and “subgroups”, recursively defining the hierarchy.
Returns:: A dictionary, the keys of which are the names of the groups, in the following form: {“<group name 1>”: <tango.Group>, “<group name 2>”: <tango.Group>, …}. Will be an empty dict if no groups were specified.
Raises:: GroupDefinitionsError – arising from GroupDefinitionsError - If error parsing JSON string. - If missing keys in the JSON definition. - If invalid device name. - If invalid groups included. - If a group has multiple parent groups. - If a device is included multiple time in a hierarchy. E.g. g1:[a,b] g2:[a,c] g3:[g1,g2]

ska_tango_base.utils.validate_capability_types(command_name: str, requested_capabilities: list[str], valid_capabilities: list[str]) → None[source]

Check the validity of the capability types passed to the specified command.

Parameters:

command_name – The name of the command to be executed.
requested_capabilities – A list of strings representing capability types.
valid_capabilities – A list of strings representing capability types.

ska_tango_base.utils.validate_input_sizes(command_name: str, argin: tuple[list[int], list[str]]) → None[source]

Check the validity of the input parameters passed to the specified command.

Parameters:

command_name – The name of the command which is to be executed.
argin – A tuple of two lists

ska_tango_base.utils.convert_dict_to_list(dictionary: dict[Any, Any]) → list[str][source]

Convert a dictionary to a list of “key:value” strings.

Parameters:: dictionary – a dictionary to be converted
Returns:: a list of key/value strings

ska_tango_base.utils.for_testing_only(func: ~collections.abc.Callable[[...], ~typing.Any], _testing_check: ~collections.abc.Callable[[], bool] = <function <lambda>>) → Callable[[...], Any][source]

Return a function that warns if called outside of testing, then calls a function.

This is intended to be used as a decorator that marks a function as available for testing purposes only. If the decorated function is called outside of testing, a warning is raised.

@for_testing_only
def _straight_to_state(self, state): ...

Parameters:

func – function to be wrapped
_testing_check – True if testing

Returns:

the wrapper function

ska_tango_base.utils.generate_command_id(command_name: str) → str[source]

Generate a unique command ID for a given command name.

Parameters:: command_name – name of the command for which an ID is to be generated.
Returns:: a unique command ID string

ska_tango_base.utils.P = ~P: Parameter specification variable.

class ska_tango_base.utils.T

Type variable.

alias of TypeVar(‘T’)

ska_tango_base.utils.deprecate_kwarg(name: str, detail: str | None = None) → Callable[[Callable[[P], T]], Callable[[P], T]][source]

Deprecate a keyword argument.

If the decorated function is passed the keyword argument a DeprecationWarning is emitted.

Parameters:

name – name of keyword argument
detail – sentence to add to the warning

Returns:

decorator