ska_oso_scripting.engineering.low.utils

ska_oso_scripting.engineering.low.utils.aiv_utils

class ska_oso_scripting.engineering.low.utils.aiv_utils.PASDDevices(station, pasdbus, fndh, smartboxes)

fndh: Alias for field number 2

pasdbus: Alias for field number 1

smartboxes: Alias for field number 3

station: Alias for field number 0

class ska_oso_scripting.engineering.low.utils.aiv_utils.MCCSDevices(controller, stations, subarrays, subarray_beams, station_beams)

controller: Alias for field number 0

station_beams: Alias for field number 4

stations: Alias for field number 1

subarray_beams: Alias for field number 3

subarrays: Alias for field number 2

class ska_oso_scripting.engineering.low.utils.aiv_utils.CSPDevices(controller, subarrays)

controller: Alias for field number 0

subarrays: Alias for field number 1

class ska_oso_scripting.engineering.low.utils.aiv_utils.SPSDevices(station, subracks, tpms, daqs)

daqs: Alias for field number 3

station: Alias for field number 0

subracks: Alias for field number 1

tpms: Alias for field number 2

class ska_oso_scripting.engineering.low.utils.aiv_utils.CBFDevices(controller, allocator, connector, processors, delaypoly, cnics, subarrays)

allocator: Alias for field number 1

cnics: Alias for field number 5

connector: Alias for field number 2

controller: Alias for field number 0

delaypoly: Alias for field number 4

processors: Alias for field number 3

subarrays: Alias for field number 6

class ska_oso_scripting.engineering.low.utils.aiv_utils.SDPDevices(controller, queue_connectors, subarrays)

controller: Alias for field number 0

queue_connectors: Alias for field number 1

subarrays: Alias for field number 2

class ska_oso_scripting.engineering.low.utils.aiv_utils.TMCDevices(central_node, csp_master_leafnode, csp_subarray_leafnodes, mccs_master_leafnode, mccs_subarray_leafnodes, sdp_master_leafnode, sdp_subarray_leafnodes, tmc_subarray_nodes)

central_node: Alias for field number 0

csp_master_leafnode: Alias for field number 1

csp_subarray_leafnodes: Alias for field number 2

mccs_master_leafnode: Alias for field number 3

mccs_subarray_leafnodes: Alias for field number 4

sdp_master_leafnode: Alias for field number 5

sdp_subarray_leafnodes: Alias for field number 6

tmc_subarray_nodes: Alias for field number 7

ska_oso_scripting.engineering.low.utils.aiv_utils.get_tmc_devices() → TMCDevices[source]

Returns TMC Tango devices. Of which the fields are:

central_node csp_master_leafnode csp_subarray_leafnodes mccs_master_leafnode mccs_subarray_leafnodes sdp_master_leafnode sdp_subarray_leafnodes tmc_subarray_nodes

These devices are derived by following Tango properties from the TMC central node. This device has properties containing the TRLs of each of the subsystem controller leaf nodes, and to the TMC subarray nodes.

The TMC subarray nodes have a property containing the TRL of the corresponding subarray leaf node device for each subsystem.

Therefore, it is possible for TMC devices to exist that are not returned by this function, if they are not referenced directly or transitively via the Tango properties of the TMC central node.

Returns:: TMCDevice namedtuple containing central node, controller leaf node, subarray node and subarray leaf node devices.

ska_oso_scripting.engineering.low.utils.aiv_utils.get_mccs_devices() → MCCSDevices[source]: Return stations, subarray beams, station beams, and subarrays.

ska_oso_scripting.engineering.low.utils.aiv_utils.get_cbf_devices() → CBFDevices[source]: CBF Devices. :return: CBFDevices named tuple

ska_oso_scripting.engineering.low.utils.aiv_utils.get_sdp_devices() → SDPDevices[source]: SDP Devices. :return: A named tuple instance holding SDP devices.

ska_oso_scripting.engineering.low.utils.aiv_utils.get_mccs_tangodb() → tango.Database[source]

Get the Database object for the Tango DB containing MCCS devices.

Returns:: a tango.Database object representing the MCCS Tango database

ska_oso_scripting.engineering.low.utils.aiv_utils.get_mccs_device(trl: str, timeout_ms=10000) → tango.DeviceProxy[source]: Grab the Tango device using a TRL, using TANGO_HOST_MCCS. :params trl: the fully qualified domain name :return device: The Tango device

ska_oso_scripting.engineering.low.utils.aiv_utils.get_stations()[source]: Returns a list of MccsStation devices. :param host: the Tango database host :return: A list of MccsStation devices

ska_oso_scripting.engineering.low.utils.aiv_utils.split_host_port(endpoint, default_port=4660)[source]: Split a host:port string into a tuple of (host, port).

ska_oso_scripting.engineering.low.utils.aiv_utils.load_tmdata(*args, **kwargs) → ska_telmodel_client.TMData

Wrap TMData with backoff to handle flaky networks.

Arguments are as for TMData.__init__().

Returns:: A TMData object.

ska_oso_scripting.engineering.low.utils.aiv_utils.get_telescope_name() → str: Returns the “telescope” field from the low-layout.json file.

ska_oso_scripting.engineering.low.utils.aiv_utils.station_label_from_id(station_id: int) → str[source]: Returns the station label for a given station ID.

ska_oso_scripting.engineering.low.utils.constants

Constants used within the workarounds module.

ska_oso_scripting.engineering.low.utils.tango_utils

ska_oso_scripting.engineering.low.utils.tango_utils.single_prop(dev: tango.DeviceProxy, prop_name: str) → str[source]

Return the first value of a device property.

Properties are always lists but many properties only expect a single value. This convenience method makes accessing those less verbose.

ska_oso_scripting.engineering.low.utils.tango_utils.get_device(trl: str, tango_host=None, timeout_ms=10000) → tango.DeviceProxy[source]: Grab the Tango device from the Tango Resource Locator (TRL) :params trl: the fully qualified domain name :return device: The Tango device

ska_oso_scripting.engineering.low.utils.tango_utils.restart_devices(devs: list[tango.DeviceProxy], force_restartserver=False)[source]

Restarts devs without affecting other devices.

If dev is the only device in its device server, restarts the device server using DServer.RestartServer(). Otherwise, restarts just this device via Init().

ska_oso_scripting.engineering.low.utils.tango_utils.member(dev: tango.DeviceProxy) → str[source]

Return the member part of the device TRL.

For example, if the TRL is “low-mccs/tile/s8-1-tpm01”, then the member part is “s8-1-tpm01”.

Returns:: the member part of the device TRL.

ska_oso_scripting.engineering.low.utils.tango_utils.wait_for(devs: DeviceProxy | list[DeviceProxy], attr: str, desired_value, failed_value=None, timeout: float = 60.0, quiet: bool = False) → None[source]

Block until attr of each of devs has a certain value or matches a predicate.

Parameters:

devs – a DeviceProxy or a list of DeviceProxy
attr – the name of the attribute to wait for
desired_value – an attribute value to wait for, or a custom predicate which may accept either one argument (the value) or two (the value, and the device)
failed_value – the value of the attribute which if reached to add the device in the failed list and not wait for any more updates on it. When every other device has reached either the desired value or this value, raise an appropriate exception. Also accepts a custom predicate.
timeout – maximum time to wait for each device to meet the condition
quiet – don’t log anything

Returns:

None

ska_oso_scripting.engineering.low.utils.tango_utils.format_trls(devs)[source]

Returns a compact representation of device names by factoring out a common prefix.

>>> format_trls(["low-mccs/tile/s8-1-tpm15", "low-mccs/tile/s8-1-tpm16"])
'low-mccs/tile/s8-1-(tpm15|tpm16)'

>>> format_trls(["low-mccs/subarray/01", "low-sdp/subarray/01"])
'low-(mccs|sdp)/subarray/01'

>>> format_trls(["low-mccs/station/s8-1", "low-mccs/spsstation/s8-1"])
'low-mccs/(spsstation|station)/s8-1'

>>> format_trls(["domain/family-a/1", "domain/family-a/2", "domain/family-b/1"])
'domain/family-a/1, domain/family-a/2, domain/family-b/1'

>>> format_trls([])
''

ska_oso_scripting.engineering.low.utils.tango_utils.fq_dev_name(dev: tango.DeviceProxy) → str[source]: Given a DeviceProxy, return a fully-qualified device name.

ska_oso_scripting.engineering.low.utils.tango_utils.convert_devicetree_str_to_deviceproxy(device_tree: dict) → dict[source]

Process a device_tree dictionary, replacing any TRL strings with device proxy instances.

This function recursively traverses a nested dictionary structure, replacing any string TRLs with DeviceProxy instances.

Parameters:: device_tree – A nested dictionary where string keys/values are TRLs
Returns:: A new nested dictionary with DeviceProxy instances replacing TRL strings

ska_oso_scripting.engineering.low.utils.versions

Functions for handling version dependencies between SKA Low components.

ska_oso_scripting.engineering.low.utils.versions.select_sdp_vis_receive_version(sdp_version: str, version_specifier: str = '') → str[source]

Use the supplied sdp_version to determine the best version of vis-receive.

version_specifier is a PEP 440-compatible version specifier that optionally restricts which vis-receive versions are eligible for selection.