Tango control subpackage (ska_tangoctl.tango_control)

Read and display Tango stuff.

class ska_tangoctl.tango_control.DispAction(disp_action: int)[source]

Control the format here.

__init__(disp_action: int)[source]

Set up the values.

Parameters:

disp_action – format flag

__repr__() str[source]

Set up the string thing.

Returns:

string thing

__str__() str[source]

Set up the string thing used for file extensions.

Returns:

string thing

__weakref__

list of weak references to the object (if defined)

check(disp_action: Any) bool[source]

Check the form.

Parameters:

disp_action – format flag

Returns:

true or false

property format: int

Get the format.

Returns:

current format

property indent: int

Get the indentation.

Returns:

indentation for JSON

show() str[source]

Print the setup.

Returns:

details about current format and output flags

class ska_tangoctl.tango_control.TangoControl(logger: Logger)[source]

Connect to Tango environment and retrieve information.

__del__() None[source]

Destructor.

__init__(logger: Logger)[source]

Set it up.

Parameters:

logger – logging handle

check_tango() int[source]

Check Tango host address.

Returns:

error condition

get_tango_classes() dict[source]

Read tango classes.

Returns:

dictionary with devices

list_classes() int[source]

Get device classes.

Returns:

error condition

list_devices() int[source]

List Tango devices.

Returns:

error condition

read_config() None[source]

Read configuration.

read_input_file() None[source]

Read instructions from JSON file.

read_input_files(json_dir: str) int[source]

Read info from JSON script files.

Parameters:

json_dir – directory with script files

Returns:

error condition

reset() None[source]

Reset it to defaults.

run_info() int[source]

Read information on Tango devices.

Returns:

error condition

set_logging_level() int[source]

Set logging level for a device.

Change a device’s logging level, where: - 0=OFF - 1=FATAL - 2=ERROR - 3=WARNING - 4=INFO - 5=DEBUG

Returns:

error condition

set_output() None[source]

Open output file.

set_value() int[source]

Set value for a Tango device.

Returns:

error condition

unset_output() None[source]

Close output file.

class ska_tangoctl.tango_control.TangoJsonReader(logger: Logger, indent: int, quiet_mode: bool, kube_namespace: str | None, devsdict: dict, outf: Any)[source]

Read JSON and print as markdown or text.

__del__() None[source]

Destructor.

__init__(logger: Logger, indent: int, quiet_mode: bool, kube_namespace: str | None, devsdict: dict, outf: Any)[source]

Rock and roll.

Parameters:
  • logger – logging handle

  • indent – indentation for JSON and YAML

  • quiet_mode – flag for displaying progress bar

  • kube_namespace – Kubernetes namespace

  • devsdict – dictionary with device data

  • outf – output file stream

__weakref__

list of weak references to the object (if defined)

print_html_large(html_body: bool) None[source]

Print the whole thing.

Parameters:

html_body – print HTML header and footer

print_html_small(html_body: bool) None[source]

Print text in short form.

Parameters:

html_body – Flag to print HTML header and footer

print_markdown_large() None[source]

Print the whole thing.

print_txt_large() None[source]

Print the whole thing.

print_txt_medium() None[source]

Print the thing in medium size.

print_txt_small() None[source]

Print text in short form.

class ska_tangoctl.tango_control.TangoScript(logger: Logger, input_file: int | str | bytes | PathLike[str] | PathLike[bytes], device_name: str | None, dry_run: bool)[source]

The classy Tango.

__init__(logger: Logger, input_file: int | str | bytes | PathLike[str] | PathLike[bytes], device_name: str | None, dry_run: bool)[source]

Read actions from file.

Raises:

Exception – error condition

Parameters:
  • logger – logging handle

  • input_file – input file

  • device_name – device name

  • dry_run – flag for dry run

__weakref__

list of weak references to the object (if defined)

read_write_attribute(attr_thing: str | None, attr_read: int | float | str | None, attr_write: int | float | str | None) int[source]

Read or write Tango attribute.

Parameters:
  • attr_thing – attribute name

  • attr_read – read value

  • attr_write – write value

Returns:

error condition

run() int[source]

Run the thing.

Returns:

error condition

run_command(cmd_thing: str, cmd_args: Any) int[source]

Run Tango command.

Parameters:
  • cmd_thing – command name

  • cmd_args – command arguments

Returns:

error condition

class ska_tangoctl.tango_control.TangoctlDevice(logger: Logger, disp_action: DispAction, outf: Any, timeout_millis: int | None, dev_status: dict, device: str, list_items: dict, block_items: dict, tgo_attrib: str | None, tgo_cmd: str | None, tgo_prop: str | None, k8s_ctx: str | None, domain_name: str | None, indent: int = 0)[source]

Compile a dictionary for a Tango device.

__del__() None[source]

Destructor.

__init__(logger: Logger, disp_action: DispAction, outf: Any, timeout_millis: int | None, dev_status: dict, device: str, list_items: dict, block_items: dict, tgo_attrib: str | None, tgo_cmd: str | None, tgo_prop: str | None, k8s_ctx: str | None, domain_name: str | None, indent: int = 0)[source]

Iniltialise the thing.

Parameters:
  • logger – logging handle

  • disp_action – display settings

  • outf – output file pointer

  • timeout_millis – Tango device timeout in milliseconds

  • dev_status – flag to read status

  • device – device name

  • list_items – attributes, commands or properties in list output

  • block_items – attributes, commands or properties not to be shown in list output

  • tgo_attrib – attribute filter

  • tgo_cmd – command filter

  • tgo_prop – property filter

  • k8s_ctx – Kubernetes context

  • domain_name – Kubernetes domain name

  • indent – indentation for JSON and YAML

Raises:

Exception – could not open device

__weakref__

list of weak references to the object (if defined)

check_for_attribute(tgo_attrib: str | None) list[source]

Filter by attribute name.

Parameters:

tgo_attrib – attribute name

Returns:

list of device names matched

check_for_command(tgo_cmd: str | None) list[source]

Filter by command name.

Parameters:

tgo_cmd – command name

Returns:

list of device names matched

check_for_property(tgo_prop: str | None) list[source]

Filter by command name.

Parameters:

tgo_prop – property name

Returns:

list of device names matched

device_run_cmd(ns_name: str, pod_name: str, pod_cmd: str) dict[source]

Run a command in specified pod.

Parameters:
  • ns_name – namespace

  • pod_name – pod name

  • pod_cmd – command to run

Returns:

dictionary with output information

make_json_large() dict[source]

Convert internal values to JSON.

Returns:

dictionary

make_json_medium() dict[source]

Convert internal values to medium size JSON.

Returns:

dictionary

make_json_small() dict[source]

Convert internal values to JSON.

Returns:

dictionary

print_html_large(html_body: bool) None[source]

Print full HTML report.

Parameters:

html_body – Flag to print HTML header and footer

print_html_small(html_body: bool) None[source]

Print shortened HTML report.

Parameters:

html_body – Flag to print HTML header and footer

print_list(eol: str = '\n') None[source]

Print data.

Parameters:

eol – printed at the end

print_list_attribute(lwid: int, show_val: bool = True) None[source]

Print list of devices with attribute.

Parameters:
  • lwid – line width

  • show_val – print value

print_list_command(lwid: int, show_val: bool = True) None[source]

Print list of devices with command.

Parameters:
  • lwid – line width

  • show_val – print value

print_list_property(lwid: int, show_val: bool = True) None[source]

Print list of devices with property.

Parameters:
  • lwid – line width

  • show_val – print value

read_attribute_config(attrib: str) tuple[source]

Read configuration of attribute.

Parameters:

attrib – attribute name

Returns:

tuple with configuration, poll period and error message

read_attribute_value() None[source]

Read device attributes.

read_command_value(run_commands: list, run_commands_name: list) None[source]

Read device commands.

Parameters:
  • run_commands – commands safe to run without parameters

  • run_commands_name – commands safe to run with device name as parameter

read_config() None[source]

Read additional data as configured in JSON file.

State, adminMode and versionId are specific to devices

read_config_all() None[source]

Read attribute and command configuration.

read_info(device: str, info: str, log_it: bool = False) Any[source]

Read device name and database info.

Parameters:
  • device – device name

  • info – what to read

  • log_it – log warning when read fails

Returns:

String with information read from Tango

read_pod(ns_name: str | None) int[source]

Read info about pod running this device.

Parameters:

ns_name – namespace

Returns:

error condition

read_procs(ns_name: str | None) int[source]

Read processes running on host.

Parameters:

ns_name – namespace

Returns:

error condition

read_property_value() None[source]

Read device properties.

write_attribute_value(attrib: str, value: str) int[source]

Set value of attribute.

Parameters:
  • attrib – attribute name

  • value – attribute value

Returns:

error condition

class ska_tangoctl.tango_control.TangoctlDevices(logger: Logger, tango_host: str | None, outf: Any, timeout_millis: int | None, dev_status: dict, cfg_data: dict, tgo_name: str | None, uniq_cls: bool, disp_action: DispAction, k8s_ctx: str | None, k8s_cluster: str | None, k8s_ns: str | None, domain_name: str | None, tgo_attrib: str | None = None, tgo_cmd: str | None = None, tgo_prop: str | None = None, tgo_class: str | None = None, dev_count: int = 0)[source]

Compile a dictionary of available Tango devices.

__del__() None[source]

Desctructor.

__init__(logger: Logger, tango_host: str | None, outf: Any, timeout_millis: int | None, dev_status: dict, cfg_data: dict, tgo_name: str | None, uniq_cls: bool, disp_action: DispAction, k8s_ctx: str | None, k8s_cluster: str | None, k8s_ns: str | None, domain_name: str | None, tgo_attrib: str | None = None, tgo_cmd: str | None = None, tgo_prop: str | None = None, tgo_class: str | None = None, dev_count: int = 0)[source]

Get a dictionary of devices.

Parameters:
  • logger – logging handle

  • tango_host – Tango database host

  • outf – output file pointer

  • timeout_millis – Tango device timeout in milliseconds

  • dev_status – dictionary with status stuff

  • cfg_data – configuration data in JSON format

  • tgo_name – filter device name

  • uniq_cls – only read one device per class

  • disp_action – output format

  • k8s_ctx – K8S context

  • k8s_cluster – K8S cluster

  • k8s_ns – K8S namespace

  • domain_name – K8S domain name

  • tgo_attrib – filter attribute name

  • tgo_cmd – filter command name

  • tgo_prop – filter property name

  • tgo_class – filter class name

  • dev_count – number of Tango device to read (for testing)

Raises:

Exception – when database connect fails

__weakref__

list of weak references to the object (if defined)

get_classes() dict[source]

Print list of device names.

Returns:

dictionary with class and device names

make_devices_json_large() dict[source]

Read device data.

Returns:

dictionary

make_devices_json_medium() dict[source]

Read device data.

Returns:

dictionary

make_devices_json_small() dict[source]

Read device data.

Returns:

dictionary

print_classes() None[source]

Print list of device names.

print_html() None[source]

Print in HTML format.

print_json() None[source]

Print in JSON format.

print_json_large() None[source]

Print in JSON format.

print_json_medium() None[source]

Print in JSON medium format.

print_json_small() None[source]

Print in shortened JSON format.

print_json_table() None[source]

Print in JSON format.

print_markdown() None[source]

Print in JSON format.

print_names_list() None[source]

Print list of device names.

print_txt(heading: str | None = None) None[source]

Print in text format.

Parameters:

heading – to be printed on the top

print_txt_heading(eol: str = '\n') int[source]

Print heading for list of devices.

Parameters:

eol – printed at the end

Returns:

width of characters printed

print_txt_large() None[source]

Print devices as text.

print_txt_list(heading: str | None = None) None[source]

Print list of devices.

Parameters:

heading – print at the top

print_txt_list_attributes(show_val: bool = True) None[source]

Print list of devices as plain text.

Parameters:

show_val – print value

print_txt_list_commands(show_val: bool = True) None[source]

Print list of device commands.

Parameters:

show_val – print value

print_txt_list_properties(show_val: bool = True) None[source]

Print list of device properties.

Parameters:

show_val – print value

print_txt_medium() None[source]

Print devices as text.

print_txt_small() None[source]

Print devices as text.

print_yaml() None[source]

Print in YAML format.

print_yaml_large() None[source]

Print in YAML large format.

print_yaml_medium() None[source]

Print in YAML medium format.

print_yaml_small() None[source]

Print in YAML small format.

read_attribute_names() dict[source]

Read device data.

Returns:

dictionary of devices

read_attribute_values() None[source]

Read device attribute values.

read_command_names() dict[source]

Read device data.

Returns:

dictionary of devices

read_command_values() None[source]

Read device commands.

read_configs() None[source]

Read additional data.

read_configs_all() None[source]

Read additional data.

read_device_hosts() list[source]

Compile a list of hosts.

Returns:

list of dictionaries with host information

read_device_values() None[source]

Read device values.

read_devices() None[source]

Read all devices.

read_devices_nodb() None[source]

Read a single device without database connection.

read_logs() None[source]

Read device logs.

read_pods() None[source]

Read device pods.

read_procs() None[source]

Read device processes.

read_property_names() dict[source]

Read device data.

Returns:

dictionary of devices

read_property_values() None[source]

Read device properties.

class ska_tangoctl.tango_control.TestTangoDevice(logger: Logger, device_name: str)[source]

Test a Tango device.

__init__(logger: Logger, device_name: str)[source]

Get going.

Parameters:
  • logger – logging handle

  • device_name – Tango device name

__weakref__

list of weak references to the object (if defined)

admin_mode_off() None[source]

Turn admin mode off.

admin_mode_on() None[source]

Turn admin mode on.

device_off() int[source]

Turn this device off.

Returns:

error condition

device_on() int[source]

Turn this device on.

Returns:

error condition

device_standby() int[source]

Set this device to standby mode.

Returns:

error condition

device_status() int | None[source]

Print device status.

Returns:

device state

get_admin_mode() int | None[source]

Read attribute for admin mode.

Returns:

attribute value

get_simulation_mode() int | None[source]

Read attribute for simulation mode.

Returns:

attribute value

read_device_attributes() None[source]

Read all attributes of this device.

run_test(dev_admin: int | None, dev_off: bool, dev_on: bool, dev_ping: bool, dev_sim: int | None, dev_standby: bool, dev_status: dict, show_attrib: bool, show_command: bool, show_prop: bool, tgo_attrib: str | None, tgo_name: str | None) int[source]

Run tests on Tango devices.

Parameters:
  • dev_admin – check admin mode

  • dev_off – turn device on

  • dev_on – turn device off

  • dev_ping – ping device

  • dev_sim – simulation flag

  • dev_standby – place device in standby

  • dev_status – set device status

  • show_attrib – test device attributes

  • show_command – test device commands

  • show_prop – test device properties

  • tgo_attrib – name of attribute

  • tgo_name – device name

Returns:

error condition

set_admin_mode(admin_mode: int) int[source]

Change admin mode.

Parameters:

admin_mode – new value

Returns:

error condition

set_simulation_mode(dev_sim: int | None) int | None[source]

Set attribute for simulation mode.

Parameters:

dev_sim – attribute value

Returns:

error condition

test_admin_mode(dev_admin: int | None) int[source]

Test admin mode.

Parameters:

dev_admin – new value

Returns:

error condition

test_all(show_attrib: bool) int[source]

Test everything that device can do.

Parameters:

show_attrib – flag for attributes display.

Returns:

error condition

test_device_attributes(show: bool = False) None[source]

Display number and names of attributes.

Parameters:

show – flag to print names

test_device_commands(show: bool = False) None[source]

Display number and names of commands.

Parameters:

show – flag to print names

test_device_properties(show: bool = False) None[source]

Display number and names of properties.

Parameters:

show – flag to print names

test_off(dev_sim: int | None) int[source]

Test that device can be turned off.

Parameters:

dev_sim – flag for hardware simulation.

Returns:

error condition

test_on(dev_sim: int | None) int[source]

Test that device can be turned on.

Parameters:

dev_sim – flag for hardware simulation.

Returns:

error condition

test_ping() int[source]

Check that device is online.

Returns:

error condition

test_simulation_mode(dev_sim: int | None) int[source]

Test that device hardware simulation can be set.

Parameters:

dev_sim – flag for hardware simulation.

Returns:

error condition

test_standby(dev_sim: int | None) int[source]

Test that device can be placed into standby mode.

Parameters:

dev_sim – flag for hardware simulation.

Returns:

error condition

test_status() int[source]

Test that device status can be read.

Returns:

error condition

test_subscribe(attrib: str) int[source]

Test subscribed to event.

Parameters:

attrib – attribute name

Returns:

error condition

ska_tangoctl.tango_control.check_tango(tango_fqdn: str, tango_port: int = 10000) int[source]

Check Tango host address.

Parameters:
  • tango_fqdn – fully qualified domain name

  • tango_port – port number

Returns:

error condition

ska_tangoctl.tango_control.show_obs_state(obs_stat: int) None[source]

Display Python enumerated type for observing state.

Parameters:

obs_stat – observing state numeric value