Utils

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

exception_manager(cls: type[Exception], callback: Callable[[], None] | None = None) → Generator[None, None, None]

Return a context manager that manages exceptions.

Parameters:

• cls – class type

• callback – a callback

Yields:

return a context manager

get_dev_info(domain_name: str, device_server_name: str, device_ref: str) → DbDevInfo

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

dp_set_property(device_name: str, property_name: str, property_value: Any) → None

Use a DeviceProxy to set a device property.

Parameters:

• device_name – tango device name

• property_name – tango property name

• property_value – tango property value

get_device_group_and_id(device_name: str) → list[str]

Return the group and id part of a device name.

Parameters:

device_name – tango device name

Returns:

group & id part of tango device name

convert_api_value(param_dict: dict[str, str]) → tuple[str, Any]

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)

coerce_value(value: Any) → Any

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

Parameters:

value – a tango DevState

Returns:

DevState as a string

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

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

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

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

get_tango_device_type_id(tango_address: str) → list[str]

Return the type id of a TANGO device.

Parameters:

tango_address – tango device address

Returns:

the type id of the device

get_groups_from_json(json_definitions: list[str]) → dict[str, Any]

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]

validate_capability_types(command_name: str, requested_capabilities: list[str], valid_capabilities: list[str]) → None

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.

validate_input_sizes(command_name: str, argin: tuple[list[int], list[str]]) → None

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

convert_dict_to_list(dictionary: dict[Any, Any]) → list[str]

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

Parameters:

dictionary – a dictionary to be converted

Returns:

a list of key/value strings

for_testing_only(func: ~typing.Callable[[...], ~typing.Any], _testing_check: ~typing.Callable[[], bool] = <function <lambda>>) → Callable[[...], Any]

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

generate_command_id(command_name: str) → str

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

P = ~P

Parameter specification variable.

class T

Type variable.

alias of TypeVar(‘T’)

deprecate_kwarg(name: str, detail: str | None = None) → Callable[[Callable[[P], T]], Callable[[P], T]]

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