VCC All Bands Controller Classes

VCCAllBandsController Class

class ska_mid_cbf_fhs_vcc.vcc_all_bands.vcc_all_bands_device.VCCAllBandsController(*args: Any, **kwargs: Any)[source]

Bases: FhsControllerBaseDevice[VCCAllBandsComponentManager], SKAObsDevice

Tango device class for the VCC All Bands Controller.

Initialise object.

set_local_change_events() → None[source]: Set attribute change events specific to the implementing class.

property component_manager_class: type[VCCAllBandsComponentManager]: The component manager class associated with the VCC controller device.

property long_running_commands: list[tuple[str, str]]

The list of all long-running commands defined on the VCC controller, each a tuple of the form (“TangoCommandName”, “python_method_name”).

Type:: list[tuple[str, str]]

expectedDishId

Read-only Tango attribute specifying the expected dish ID for this VCC.

Returns:: The expected dish ID.
Return type:: str

subarrayID

Read-only Tango attribute specifying the subarray ID currently assigned to this VCC.

Returns:: The assigned subarray ID.
Return type:: int

frequencyBand

Read-only Tango attribute specifying the frequency band in which this VCC is operating.

Returns:: the frequency band (being observed by the current scan, one of [“1”, “2”, “3”, “4”, “5a”, “5b”]).
Return type:: FrequencyBandEnum

inputSampleRate

Read-only Tango attribute specifying the input sample rate of this VCC.

Returns:: The input sample rate.
Return type:: int

frequencyBandOffset

Read-only Tango attribute specifying the frequency band offset(s) configured for this VCC. Contains two offset values, however the second value is only meaningful when operating in band 5. Only the first value is required in all other bands.

Returns:: the frequency band offset(s).
Return type:: list[int]

requestedRFIHeadroom

Read-only Tango attribute specifying the requested headrooms from the most recent call to AutoSetFilterGains.

Returns:: The requested headrooms.
Return type:: list[float]

vccGains

Read-only Tango attribute specifying the currently applied gain multipliers for VCC coarse channels.

Returns:: The VCC coarse channel gain multipliers, in the format [ch0_polX, ch1_polX, …, chN_polX, ch0_polY, ch1_polY, …, chN_polY].
Return type:: list[float]

UpdateSubarrayMembership(subarray_id: int) → tuple[list[ResultCode], list[str]][source]

Tango command to update the subarray currently assigned to this VCC.

Parameters:: subarray_id (int) – The subarray ID to assign.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

AutoSetFilterGains(auto_set_filter_gains_schema: str | None = None) → tuple[list[ResultCode], list[str]][source]

Tango command to start a scan operation.

Parameters:: auto_set_filter_gains_schema (str) – JSON String following the auto set filter gains command schema Requested RFI headroom, in decibels (dB). Must be a list containing either a single value to apply to all frequency slices, or a value per frequency slice to be applied separately.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

init_device() → None[source]: Initialize the Tango device after startup.

create_component_manager() → VCCAllBandsComponentManager[source]

Instantiate the component manager for this device.

Returns:: The instantiated component manager.
Return type:: FhsControllerComponentManagerT

reset_obs_state()[source]: Set the ObsState of the device back to IDLE.

Abort() → tuple[list[ResultCode], list[str]]

Abort any executing long running command(s) and empty the queue.

Abort is itself a long running command that is executed immediately.

Subclasses should override execute_Abort() to change the behaviour of this command.

Returns:: A tuple containing ResultCode.STARTED and the unique ID of the command

AbortCommands() → tuple[list[ResultCode], list[str]]

Empty out long running commands in queue.

Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

CheckLongRunningCommandStatus(argin: str) → str

Check the status of a long running command by ID.

Parameters:: argin – the command id
Returns:: command status

ClearLoggingLevelOverrides() → tuple[list[ResultCode], list[str]]

Tango command to clear all IP block logging level overrides, such that all IP blocks will use the global controller logging level.

Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

class ClearLoggingLevelOverridesCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the ClearLoggingLevelOverrides command.

Initialise a new BaseCommand instance.

Parameters:

logger – the logger to be used by this Command. If not provided, then a default module logger will be used.
validator – an optional validator to use to parse, validate and/or unpack command arguments.

do() → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

Exception – any exception that is raised during the execution of the self.do method.

Returns:

result of command

ConfigureScan(config: str) → tuple[list[ResultCode], list[str]]

Tango command to apply a received scan configuration in preparation for a scan.

Parameters:: config (str) – The configuration JSON string.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

DEVICE_CLASS_DESCRIPTION: ClassVar[str | None]

Description of the device class (optional).

If not specified, the class docstring is used. Available to clients via tango.DeviceProxy.description(). Use as a class variable.

DEVICE_CLASS_INITIAL_STATE: ClassVar[DevState]

Initial state value for all instances of the device (optional).

Use as a class variable.

DEVICE_CLASS_INITIAL_STATUS: ClassVar[str]

Initial status string for all instances of the device (optional).

Use as a class variable.

DebugDevice() → int

Enable remote debugging of this device.

To modify behaviour for this command, modify the do() method of the command class: DebugDeviceCommand.

Returns:: the port the debugger is listening on

EndScan(end_scan_schema: str | None = None) → tuple[list[ResultCode], list[str]]

Tango command to end a scan operation.

Parameters:: end_scan_schema (str) – The end scan schema JSON string.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

GetLongRunningCommandResult(transaction_id: str | None = None) → tuple[list[ResultCode], list[str]]

Tango command to get the result(s) of Long Running Commands

Parameters:: transaction_id (str | None, optional) – The transaction id of the Long Running Command to fetch the result for.
Returns:: The Tango result code and a JSON string containing all requested results.
Return type:: tuple[list[ResultCode], list[str]]

class GetLongRunningCommandResultCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the GetLongRunningCommandResult command.

Initialise a new BaseCommand instance.

Parameters:

logger – the logger to be used by this Command. If not provided, then a default module logger will be used.
validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(transaction_id: str | None = None) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

Exception – any exception that is raised during the execution of the self.do method.

Returns:

result of command

GetStatus(ip_blocks: list[str] | None = None) → tuple[list[ResultCode], list[str]]

Tango command to fetch the status(es) of any specified IP block(s).

Parameters:: ip_blocks (list[str] | None, optional) – The list of IP block names/IDs to fetch the status for.
Returns:: The Tango result code and a JSON string containing all requested statuses.
Return type:: tuple[list[ResultCode], list[str]]

class GetStatusCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the GetStatus command.

Initialise a new BaseCommand instance.

Parameters:

logger – the logger to be used by this Command. If not provided, then a default module logger will be used.
validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(ip_blocks: list[str] | None = None) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

Exception – any exception that is raised during the execution of the self.do method.

Returns:

result of command

GetVersionInfo() → list[str]

Return the version information of the device.

To modify behaviour for this command, modify the do() method of the command class.

Returns:: The result code and the command unique ID

GoToIdle(go_to_idle_schema: str) → tuple[list[ResultCode], list[str]]

Tango command to return the device to IDLE state.

Parameters:: go_to_idle_schema (str) – The go_to_idle schema JSON string.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

GroupDefinitions

Device property.

Each string in the list is a JSON serialised dict defining the group_name, devices and subgroups in the group. A Tango Group object is created for each item in the list, according to the hierarchy defined. This provides easy access to the managed devices in bulk, or individually.

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": []}
   ]} ]

InitCommand = None

LoggingLevelDefault

Device property.

Default logging level at device startup. See LoggingLevel

LoggingTargetsDefault

Device property.

Default logging targets at device startup. See the project readme for details.

ObsReset(transaction_id: str | None = None) → tuple[list[ResultCode], list[str]]

Tango command to reset the observing state.

Parameters:: transaction_id (str) – The Transaction id, can be none
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

Off: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

On: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

ROOT_OBSERVER_PREFIX: ClassVar[str] = '.'

Reset: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

Scan(scan_schema: str) → tuple[list[ResultCode], list[str]]

Tango command to start a scan operation.

Parameters:: scan_schema (str) – The scan schema JSON string.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

SkaLevel

Device property.

Indication of importance of the device in the SKA hierarchy to support drill-down navigation: 1..6, with 1 highest.

Standby: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

TangoClassClass: alias of VCCAllBandsControllerClass

TangoClassName = 'VCCAllBandsController'

UpdateGlobalLoggingLevel(logging_level: str) → tuple[list[ResultCode], list[str]]

Tango command to update the global logging level, impacting the controller and any IP blocks without logging level overrides.

Parameters:: logging_level (str) – The new logging level.
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

class UpdateGlobalLoggingLevelCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the UpdateGlobalLoggingLevel command.

Initialise a new BaseCommand instance.

Parameters:

logger – the logger to be used by this Command. If not provided, then a default module logger will be used.
validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(logging_level: str) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

Exception – any exception that is raised during the execution of the self.do method.

Returns:

result of command

UpdateIPBlockLoggingLevels(input_json: str) → tuple[list[ResultCode], list[str]]

Tango command to assign a specified logging level override to all specified IP block(s).

Parameters:: input_json (str) – A JSON string in the format: {“ip_blocks”: [“SomeIPBlockIDOne”, “SomeIPBlockIDTwo”], “level”: “DEBUG”}
Returns:: The Tango result code and a string message indicating status. The message is for information purpose only.
Return type:: tuple[list[ResultCode], list[str]]

class UpdateIPBlockLoggingLevelsCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the UpdateIPBlockLoggingLevels command.

Initialise a new BaseCommand instance.

Parameters:

logger – the logger to be used by this Command. If not provided, then a default module logger will be used.
validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(input_json: str) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

invoke(*args: Any, **kwargs: Any) → CommandReturnT

Invoke the command.

This is implemented to simply call the do() hook, thus running the user-specified functionality therein.

Parameters:

args – positional args to the component manager method
kwargs – keyword args to the component manager method

Raises:

Exception – any exception that is raised during the execution of the self.do method.

Returns:

result of command

_admin_mode: Signal[ska_control_model.AdminMode]

Signal for the admin mode of the device.

Values are emitted for this signal whenever a client successfully changes to the adminMode attribute.

Typically, subclasses will not need to listen to this signal and should instead override change_control_level().

_commanded_obs_state: CommandedObsStateSignal

Signal for the commanded observation state of the device.

Write to this signal whenever a command is executed which will result in a obs state transition.

_commanded_state: CommandedStateSignal

Signal for the commanded Operational State of the device.

Write to this signal whenever a command is executed which will result in a state transition.

_config_delay_expected: Signal[int]

Signal for the configuration delay expected of the device.

Write to this signal to report the expected configuration delay to Tango clients.

_config_progress: Signal[int]

Signal for the configuration progress of the device.

Write to this signal to report configuration progress to Tango clients.

_control_mode: Signal[ska_control_model.ControlMode]

Signal for the control mode of the device.

Values are emitted for this signal whenever a client successfully changes to the controlMode attribute.

_health_state

Signal for the health state of the device.

Write to this signal to set the reported health state of the Tango device.

_obs_mode: Signal[ska_control_model.ObsMode]

Signal for the observation mode of the device.

Write to this signal to report observation mode of the device to Tango clients.

_obs_state: Signal[ska_control_model.ObsState]

Signal for the observation state of the device.

Write to this signal to report the observation state of the device to Tango clients.

_op_state: OpStateSignal

Signal for the Operational State of the device.

Write to this signal to set the state of the Tango device.

_simulation_mode: Signal[ska_control_model.SimulationMode]

Signal for the simulation mode of the device.

Values are emitted for this signal whenever a client successfully changes to the simulationMode attribute.

_status: Signal[str]

Signal for the status of the device.

Write to this signal to set the status of the Tango device.

_test_mode: Signal[ska_control_model.TestMode]

Signal for the test mode of the device.

Values are emitted for this signal whenever a client successfully changes to the testMode attribute.

_update_admin_mode(admin_mode: AdminMode) → None

Update the admin mode of the device.

This helper method is passed to the admin mode model as a callback, so that the model can trigger actions in the Tango device.

Parameters:: admin_mode – the new admin mode value

add_command(cmd, device_level: bool = True)

Add a new command to the device command list.

Parameters:

cmd (server.command) – the new command to be added to the list
device_level (bool) – Set this flag to true if the command must be added for only this device

Raises:

DevFailed

add_version_info(self: tango._tango.DeviceImpl, key: str, value: str) → None

Method to add information about the module version a device is using

Parameters:

key (str) – Module name
value (str) – Module version, or other relevant information.

Added in version 10.0.0.

adminMode: attribute_from_signal: Admin mode attribute of the device.

allocate_lrc(name: str, *, started_callback: Callable[[...], None] | str | None = None, completed_callback: Callable[[...], None] | str | None = None) → tuple[str, TaskCallbackType]

Allocate a new command_id for a long running command.

The status and progress of the long running command should be updated using the returned task_callback object.

Parameters:

name – The name of the command
started_callback – Callback passed to the command tracker that is called when the command transitions to IN_PROGRESS status. If a str, this will be used to look up a callable on the self. If None the default name of “started_<Cmd>” will be used.
completed_callback – Callback passed to the command tracker that is called when the command transitions to COMPLETED status. If a str, this will be used to look up a callable on the self. If None the default name of “completed_<Cmd>” will be used.

Returns:

(command_id, task_callback)

allow_internal_threads() → Iterator[None]

Allow internal threads to run without allowing new Tango requests.

This is intended to be used from a Tango client request thread to allow internal threads to acquire the Tango monitor, without allowing another Tango request to start.

This is used in the following two places in ska-tango-base:

During init_device() to allow background threads to
cleanup gracefully and be joined.

If we allow other threads to run during init_device() or delete_device(), a client could initiate a new request on our device as the Tango client thread is able to grab the device lock. This should be avoided because the device is only partially initialised. This method avoids this issue by arranging for Tango client threads to be disallowed from grabbing the Tango monitor, while still allowing other internal threads to acquire the monitor.
At the start of every request to allow the signal bus thread to
catch up.

This ensures that Tango clients get a sequentially consistent view of the Tango device. That is, if they call a command which modifies the value of an attribute then read that attribute, they will see the updated value, even if the attribute is using the signal bus (e.g. via attribute_from_signal).

Subclasses of SignalBusMixin should

always_executed_hook() → None: Raise an exception if we have released threads during an Init command.

append_status(self: tango._tango.DeviceImpl, status: str, new_line: bool = False) → None

Appends a string to the device status.

Parameters:

status (str) – the string to be appended to the device status
new_line (bool) – If true, appends a new line character before the string. Default is False

async async_add_attribute(attr, r_meth: Callable | None = None, w_meth: Callable | None = None, is_allo_meth: Callable | None = None)

Add a new attribute to the device attribute list.

Please, note that if you add an attribute to a device at device creation time, this attribute will be added to the device class attribute list. Therefore, all devices belonging to the same class created after this attribute addition will also have this attribute.

If you pass a reference to unbound method for read, write or is_allowed method (e.g. DeviceClass.read_function or self.__class__.read_function), during execution the corresponding bound method (self.read_function) will be used.

Note: Calling the synchronous add_attribute method from a coroutine function in an asyncio server may cause a deadlock. Use await async_add_attribute() instead. However, if overriding the synchronous method initialize_dynamic_attributes, then the synchronous add_attribute method must be used, even in asyncio servers.

param attr:

the new attribute to be added to the list.

type attr:

attribute or Attr or AttrData

param r_meth:

the read method to be called on a read request (if attr is of type server.attribute, then use the fget field in the attr object instead)

type r_meth:

typing.Callable

param w_meth:

the write method to be called on a write request (if attr is writable) (if attr is of type server.attribute, then use the fset field in the attr object instead)

type w_meth:

typing.Callable

param is_allo_meth:

the method that is called to check if it is possible to access the attribute or not (if attr is of type server.attribute, then use the fisallowed field in the attr object instead)

type is_allo_meth:

typing.Callable

raises:

DevFailed

Added in version 10.0.0.

async async_remove_attribute(attr_name: str, free_it: bool = False, clean_db: bool = True)

Remove one attribute from the device attribute list.

Note: Call of synchronous remove_attribute method from a coroutine function in an asyncio server may cause a deadlock. Use await async_remove_attribute() instead. However, if overriding the synchronous method initialize_dynamic_attributes, then the synchronous remove_attribute method must be used, even in asyncio servers.

Parameters:

attr_name (str) – attribute name
free_it (bool) – free Attr object flag. Default False
clean_db (bool) – clean attribute related info in db. Default True

Raises:

DevFailed

Added in version 10.0.0.

bitstream_id

ID of the bitstream used for this device.

Type:: device_property(dtype=str)

bitstream_path

File path to the bitstream used by the device.

Type:: device_property(dtype=str)

bitstream_version

Version of the bitstream used for this device.

Type:: device_property(dtype=str)

bitstreams

JSON string containing the bitstreams currently available to use.

Type:: device_property(dtype=str)

buildState

Read the Build State of the device.

Returns:: the build state of the device

check_command_exists(self: tango._tango.DeviceImpl, cmd_name: str) → None

Check that a command is supported by the device and does not need input value.

The method throws an exception if the command is not defined or needs an input value.

Parameters:: cmd_name (str) – the command name
Raises:: DevFailed
Raises:: API_IncompatibleCmdArgumentType
Raises:: API_CommandNotFound

property command_tracker: CommandTrackerProtocol

Get the command tracker.

Returns:: The initialised CommandTrackerProtocol object.

commandedObsState

Attribute for the last commanded Observation State of the device.

This should be set by subclasses of this interface by writing to _commanded_obs_state.

commandedState: attribute_from_signal

Attribute for the last commanded Operating State of the device.

This should be set by subclasses of this interface by writing to _commanded_state.

communicationState

Read-only Tango attribute specifying the communication state of the device.

Returns:: The communication state of the device.
Return type:: CommunicationStatus

component_disconnected() → None: Perform disconnected action on the OpStateModel.

component_fault() → None: Perform fault action on the OpStateModel.

property component_manager: ComponentManagerT: Get the component manager.

component_no_fault() → None: Perform fault stopped action on the OpStateModel.

component_off() → None: Perform off action on the OpStateModel.

component_on() → None: Perform on action on the OpStateModel.

component_standby() → None: Perform cstandby action on the OpStateModel.

component_unknown() → None: Perform unknown state action on the OpStateModel.

configurationDelayExpected

Configuration delay expected attribute of the device.

This should be set by subclasses of this interface by writing to _config_delay_expected.

configurationProgress

Configuration progress attribute of the device.

This should be set by subclasses of this interface by writing to _config_progress.

controlMode: attribute_from_signal

Control mode attribute of the device.

The control mode of the device is either REMOTE or LOCAL. Tango Device accepts only from a ‘local’ client and ignores commands and queries received from TM or any other ‘remote’ clients. The Local clients has to release LOCAL control before REMOTE clients can take control again.

static convert_submission_result_to_lrc_return(command_id: str, status: TaskStatus, message: str) → tuple[list[ResultCode], list[str]]

Convert a submit result to a DevVarLongStringArray.

Example:

@command
def MyCommand(self) -> DevVarLongStringArrayType:
    command_id, task_callback = self.allocate_lrc("MyCommand")
    status, message = self.submit_lrc_task(
        "MyCommand", self.do_my_command, task_callback
    )
    return self.submit_result_to_lrc_return(
        command_id, status, message
    )

Parameters:

command_id – The command id of the LRC
status – The task status returned by submit
message – The message returned by submit

Returns:

[converted ResultCode], [command_id or message]

create_telemetry_tracer(device_tracer_provider) → opentelemetry.trace.Tracer

Factory method returning a Tracer for telemetry.

The default implementation can be overridden.

Added in version 10.0.0.

create_telemetry_tracer_provider(class_name, device_name) → opentelemetry.trace.TracerProvider

Factory method returning a TracerProvider for telemetry.

The default implementation can be overridden.

Added in version 10.0.0.

debug_stream(msg: str, *args, source: Callable | None = None) → None

Sends the given message to the tango debug stream.

Since PyTango 7.1.3, the same can be achieved with:

print(msg, file=self.log_debug)

Parameters:

msg (str) – the message to be sent to the debug stream
*args – Arguments to format a message string.
source (typing.Callable) – Function that will be inspected for filename and lineno in the log message.

Added in version 9.4.2: added source parameter

dev_state(self: tango._tango.Device_6Impl) → tango._tango.DevState

Get device state.

Default method to get device state. The behaviour of this method depends on the device state. If the device state is ON or ALARM, it reads the attribute(s) with an alarm level defined, check if the read value is above/below the alarm and eventually change the state to ALARM, return the device state. For all the other device states, this method simply returns the state.

This method can be redefined in sub-classes in case of the default behaviour does not fullfill the needs.

Raises:: DevFailed: If it is necessary to read attribute(s) and a problem occurs during the reading

dev_status(self: tango._tango.Device_6Impl) → str

Get device status.

Default method to get device status. It returns the contents of the device dev_status field. If the device state is ALARM, alarm messages are added to the device status.

This method can be redefined in sub-classes in case of the default behaviour does not fullfill the needs.

Raises:: DevFailed: If it is necessary to read attribute(s) and a problem occurs during the reading

device_gitlab_hash

Gitlab hash of the device.

Type:: device_property(dtype=str)

device_id

ID of the device.

Type:: device_property(dtype=int)

device_version_num

Version of the device.

Type:: device_property(dtype=str)

emulation_mode

1 if the device is deployed in emulation mode, 0 otherwise.

Type:: device_property(dtype=int)

emulator_base_url

Base URL of the emulator used by the device.

Type:: device_property(dtype=str)

emulator_id

ID of the emulator used by the device.

Type:: device_property(dtype=str)

error_stream(msg: str, *args, source: Callable | None = None) → None

Sends the given message to the tango error stream.

Since PyTango 7.1.3, the same can be achieved with:

print(msg, file=self.log_error)

Parameters:

msg (str) – the message to be sent to the error stream
*args – Arguments to format a message string.
source (typing.Callable) – Function that will be inspected for filename and lineno in the log message.

Added in version 9.4.2: added source parameter

execute_Off() → tuple[list[ResultCode], list[str]]

Execute the standard Off command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should turn the device off and result in the operational state transitioning to OFF. The command should not return until the transition has occurred.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

execute_On() → tuple[list[ResultCode], list[str]]

Execute the standard On command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should turn the device on and result in the operational state transitioning to ON or ALARM. The command should not return until the transition has occurred.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

execute_Reset() → tuple[list[ResultCode], list[str]]

Execute the standard Reset command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

For a device that directly monitors and controls hardware, this command should put that hardware into a known state, for example by clearing buffers and loading default values into registers, or if necessary even by power-cycling and re-initialising the hardware.

Logical devices should generally implement this command to perform a sensible reset of that logical device. For example, aborting any current activities and clearing internal state.

Reset() generally should not change the power state of the device or its hardware:

If invoking Reset() from STANDBY state, the device would usually be expected to remain in STANDBY.
If invoking Reset() from ON state, the device would usually be expected to remain in ON.
If invoking Reset() from FAULT state, the device would usually be expected to transition to ON or remain in FAULT, depending on whether the reset was successful in clearing then fault.

Reset() generally should not propagate to subservient devices. For example, a subsystem controller device should implement :py:meth`!Reset()` to reset the subsystem as a whole, but that probably should not result in all of the subsystem’s hardware being power-cycled.

Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

execute_Standby() → tuple[list[ResultCode], list[str]]

Execute the standard Standby command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should put the device in standby mode and result in the operational state transitioning to STANDBY. The command should not return until the transition has occurred.

Returns:: A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

property fast_commands: list[tuple[str, FhsFastCommand]]

A list of fast commands defined on this controller, each a tuple of the form (“TangoCommandName”, PythonFastCommandClass).

Type:: list[tuple[str, FhsFastCommand]]

fatal_stream(msg: str, *args, source: Callable | None = None) → None

Sends the given message to the tango fatal stream.

Since PyTango 7.1.3, the same can be achieved with:

print(msg, file=self.log_fatal)

Parameters:

msg (str) – the message to be sent to the fatal stream
*args – Arguments to format a message string.
source (typing.Callable) – Function that will be inspected for filename and lineno in the log message.

Added in version 9.4.2: added source parameter

fill_attr_polling_buffer(attribute_name: str, attr_history_stack: TimedAttrData | list[TimedAttrData]) → None

Fill attribute polling buffer with your own data. E.g.:

def fill_history(self):
    # note is such case quality will ATTR_VALID, and time_stamp will be time.time()
    self.fill_attr_polling_buffer(attribute_name, TimedAttrData(my_new_value))

or:

def fill_history(self):
    data = TimedAttrData(value=my_new_value,
                         quality=AttrQuality.ATTR_WARNING,
                         w_value=my_new_w_value,
                         time_stamp=my_time)

    self.fill_attr_polling_buffer(attribute_name, data)

or:

def fill_history(self):
    data = [TimedAttrData(my_new_value),
            TimedAttrData(error=RuntimeError("Cannot read value")]

    self.fill_attr_polling_buffer(attribute_name, data)

Parameters:

attribute_name (str) – name of the attribute to fill polling buffer
attr_history_stack (TimedAttrData or list[TimedAttrData]) – data to be inserted.

Raises:

DevFailed

Added in version 10.1.0.

fill_cmd_polling_buffer(command_name: str, cmd_history_stack: TimedCmdData | list[TimedCmdData]) → None

Fill command polling buffer with your own data. E.g.:

def fill_history(self):
    # note is such case time_stamp will be set to time.time()
    self.fill_cmd_polling_buffer(command_name, TimedCmdData(my_new_value))

or:

def fill_history(self):
    data = TimedCmdData(value=my_new_value,
                         time_stamp=my_time)

    self.fill_cmd_polling_buffer(command_name, data)

or:

def fill_history(self):
    data = [TimedCmdData(my_new_value),
            TimedCmdData(error=RuntimeError("Cannot read value")]

    self.fill_cmd_polling_buffer(command_name, data)

Parameters:

command_name (str) – name of the command to fill polling buffer
cmd_history_stack (TimedCmdData or list[TimedCmdData]) – data to be inserted

Raises:

DevFailed

Added in version 10.1.0.

get_attr_min_poll_period(self: tango._tango.DeviceImpl) → tango._tango.StdStringVector: Returns the min attribute poll period in milliseconds

Added in version 7.2.0.

get_attr_poll_ring_depth(self: tango._tango.DeviceImpl, attr_name: str) → int

Returns the attribute poll ring depth.

Parameters:: attr_name (str) – the attribute name

Added in version 7.1.2.

get_attribute_config(attr_names) → list[AttributeConfig]

Returns the list of AttributeConfig for the requested names

Parameters:: attr_names (list[str] | str) – sequence of str with attribute names, or single attribute name
Returns:: tango.AttributeConfig for each requested attribute name
Return type:: list[tango.AttributeConfig]

get_attribute_config_2(attr_names: str | list[str]) → list[AttributeConfig_2]

Returns the list of AttributeConfig_2 for the requested names

Parameters:: attr_names (list[str] | str) – sequence of str with attribute names, or single attribute name
Returns:: tango.AttributeConfig_2 for each requested attribute name

get_attribute_config_3(attr_names: str | list[str]) → list[AttributeConfig_3]

Returns the list of AttributeConfig_3 for the requested names

Parameters:: attr_names (list[str] | str) – sequence of str with attribute names, or single attribute name
Returns:: tango.AttributeConfig_3 for each requested attribute name

get_attribute_poll_period(self: tango._tango.DeviceImpl, attr_name: str) → int

Returns the attribute polling period (milliseconds) or 0 if the attribute is not polled.

Parameters:: attr_name (str) – attribute name

Added in version 8.0.0.

get_client_ident(self: tango._tango.DeviceImpl) → tango._tango.ClientAddr

Get client identification.

This method is only useful while handling a command or attribute read/write. I.e., when a method has been invoked by a client. It will return None if the method was not invoked in the context of a client call. E.g., called on startup, or called internally (e.g., from the polling loop).

It can only be used with tango.GreenMode.Synchronous device servers. Other device servers will not have the correct context active at the time the attribute/command handler is running. E.g., for an asyncio device server, the handler is running in the asyncio event loop thread.

Returns:: client identification structure
Return type:: ClientAddr | None

get_cmd_min_poll_period(self: tango._tango.DeviceImpl) → tango._tango.StdStringVector: Returns the min command poll period in milliseconds.

Added in version 7.2.0.

get_cmd_poll_ring_depth(self: tango._tango.DeviceImpl, cmd_name: str) → int

Returns the command poll ring depth.

Parameters:: cmd_name (str) – the command name

Added in version 7.1.2.

get_command_poll_period(self: tango._tango.DeviceImpl, cmd_name: str) → int

Returns the command polling period (milliseconds) or 0 if the command is not polled.

Parameters:: cmd_name (str) – command name

Added in version 8.0.0.

get_dev_idl_version(self: tango._tango.DeviceImpl) → int: Returns the IDL version.

Added in version 7.1.2.

get_dev_state() → DevState

Get the DevState of the device.

Returns:: The DevState of the device.
Return type:: DevState

get_device_attr(self: tango._tango.DeviceImpl) → tango._tango.MultiAttribute: Get device multi attribute object.

get_device_class() → DeviceClass: Get device class singleton.

get_device_properties(ds_class=None)

Utility method that fetches all the device properties from the database and converts them into members of this DeviceImpl.

Parameters:: ds_class (DeviceClass) – the DeviceClass object. Optional. Default value is None meaning that the corresponding DeviceClass object for this DeviceImpl will be used
Raises:: DevFailed:

get_exported_flag(self: tango._tango.DeviceImpl) → bool: Returns the state of the exported flag

Added in version 7.1.2.

get_logger(self: tango._tango.DeviceImpl) → tango._tango.Logger: Returns the Logger object for this device

get_min_poll_period(self: tango._tango.DeviceImpl) → int: Returns the min poll period in milliseconds.

Added in version 7.2.0.

get_name(self: tango._tango.DeviceImpl) → str: Get the device name.

get_non_auto_polled_attr(self: tango._tango.DeviceImpl) → tango._tango.StdStringVector: Returns the list of non automatic polled attributes

Added in version 7.1.2.

get_non_auto_polled_cmd(self: tango._tango.DeviceImpl) → tango._tango.StdStringVector: Returns the list of non automatic polled commands

Added in version 7.1.2.

get_poll_old_factor(self: tango._tango.DeviceImpl) → int: Returns the poll old factor

Added in version 7.1.2.

get_poll_ring_depth(self: tango._tango.DeviceImpl) → int: Returns the poll ring depth

Added in version 7.1.2.

get_polled_attr(self: tango._tango.DeviceImpl) → tango._tango.StdStringVector: Returns the list of polled attributes

Added in version 7.1.2.

get_polled_cmd(self: tango._tango.DeviceImpl) → tango._tango.StdStringVector: Returns the list of polled commands

Added in version 7.1.2.

get_prev_state(self: tango._tango.DeviceImpl) → tango._tango.DevState: Get the device’s previous state.

get_state(self: tango._tango.DeviceImpl) → tango._tango.DevState: Get the device state.

get_status(self: tango._tango.DeviceImpl) → str: Get the device status.

get_telemetry_tracer() → opentelemetry.trace.Tracer: Returns device telemetry tracer, or None if telemetry not supported.

Added in version 10.0.0.

get_version_info(self: tango._tango.DeviceImpl) → dict

Returns a dict with versioning of different modules related to the pytango device.

Example

{: “Build.PyTango.NumPy”: “1.26.4”, “Build.PyTango.Pybind11”: “3.0.1”, “Build.PyTango.Python”: “3.12.2”, “Build.PyTango.cppTango”:”10.0.0”, “NumPy”: “1.26.4”, “PyTango”: “10.0.0.dev0”, “Python”: “3.12.2”, “cppTango”: “10.0.0”, “omniORB”: “4.3.2”, “zmq”: “4.3.5”

}

Added in version 10.0.0.

healthState

Health state attribute of the device.

This should be set by subclasses of this interface by writing to _health_state.

info_stream(msg: str, *args, source: Callable | None = None) → None

Sends the given message to the tango info stream.

Since PyTango 7.1.3, the same can be achieved with:

print(msg, file=self.log_info)

Parameters:

msg (str) – the message to be sent to the info stream
*args – Arguments to format a message string.
source (typing.Callable) – Function that will be inspected for filename and lineno in the log message.

Added in version 9.4.2: added source parameter

init_command_objects() → None: Set up all command objects.

init_fast_command_objects(commands_and_classes: list[tuple]) → None

Set up all fast command objects.

Parameters:: commands_and_classes (list[tuple]) – The list of commands to set up for this device, each a tuple of the form (“TangoCommandName”, PythonFastCommandClass).

init_invoked() → None: Perform ‘init_invoked’ action on the OpStateModel.

init_logger(self: tango._tango.DeviceImpl) → None: Setups logger for the device. Called automatically when device starts.

initialize_dynamic_attributes(): Method executed at initializion phase to create dynamic attributes. Default implementation does nothing. Overwrite when necessary.

Note

This method is only called once when the device server starts, after init_device(), but before the device is marked as exported. If the Init command is executed on the device, the initialize_dynamic_attributes() method will not be called again.

ipBlockAliases

Read-only Tango attribute specifying all possible aliases for each IP block (which can be used e.g. in command arguments).

Returns:: The list of aliases for all IP blocks.
Return type:: str

ipBlockIDs

Read-only Tango attribute specifying the list of IP block IDs the controller can interact with.

Returns:: The list of IP block IDs the controller can interact with.
Return type:: list[str]

ip_blocks

Base64-encoded string containing the deploy-time properties of all IP blocks managed by the controller.

Type:: device_property(dtype=str)

is_Off_allowed = None

is_On_allowed = None

is_Reset_allowed = None

is_Standby_allowed = None

is_attribute_polled(self: tango._tango.DeviceImpl, attr_name: str) → bool

True if the attribute is polled.

Parameters:: attr_name (str) – attribute name

is_command_polled(self: tango._tango.DeviceImpl, cmd_name: str) → bool

True if the command is polled.

Parameters:: cmd_name (str) – attribute name

is_device_locked(self: tango._tango.DeviceImpl) → bool: Returns if this device is locked by a client.

Added in version 7.1.2.

is_polled(self: tango._tango.DeviceImpl) → bool: Returns if it is polled

Added in version 7.1.2.

is_there_subscriber(self: tango._tango.DeviceImpl, attr_name: str, event_type: tango._tango.EventType) → bool

Check if there is subscriber(s) listening for the event.

This method returns a boolean set to true if there are some subscriber(s) listening on the event specified by the two method arguments. Be aware that there is some delay (up to 600 sec) between this method returning false and the last subscriber unsubscription or crash…

The device interface change event is not supported by this method.

Parameters:

attr_name (str) – the attribute name
event_type (EventType) – the event type

property log_debug

property log_error

property log_fatal

property log_info

property log_warn

loggingLevel

Read the logging level of the device.

Initialises to LoggingLevelDefault on startup. See LoggingLevel

Returns:: Logging level of the device.

loggingLevelOverrides

Read-only Tango attribute specifying any overridden IP block logging levels.

Returns:: JSON string containing any overridden IP block logging levels.
Return type:: str

loggingTargets

Read the additional logging targets of the device.

Note that this excludes the handlers provided by the ska_ser_logging library defaults - initialises to LoggingTargetsDefault on startup.

Returns:: Logging level of the device.

logging_level

Logging level of the device.

Type:: device_property(dtype=str)

long_running_command_result_buffer_max_size

Max size of the long running command result buffer.

Type:: device_property(dtype=str)

lrcFinished: Expose a signal as a Tango attribute.

lrcProtocolVersions

Return supported protocol versions.

Returns:: A tuple containing the lower and upper bounds of supported long running command protocol versions.

notify_emission(signal: str, value: Any) → None: Push change events for AttributeSignal objects.

obsMode

Observation mode attribute of the device.

This should be set by subclasses of this interface by writing to _obs_mode.

obsState

Observation state attribute of the device.

This should be set by subclasses of this interface by writing to _obs_state.

property observer_prefix: str: Return the path from the root SharingObserver holding the bus.

poll_attribute(self: tango._tango.DeviceImpl, attr_name: str, period_ms: SupportsInt | SupportsIndex) → None

Add an attribute to the list of polled attributes.

Parameters:

attr_name (str) – attribute name
period_ms (int) – polling period in milliseconds

poll_command(self: tango._tango.DeviceImpl, cmd_name: str, period_ms: SupportsInt | SupportsIndex) → None

Add a command to the list of polled commands.

Parameters:

cmd_name (str) – command name
period_ms (int) – polling period in milliseconds

push_alarm_event(self, attr_name: str, except: Exception)

push_alarm_event(self, attr_name: str, data: Any) → None

push_alarm_event(self, attr_name: str, data: Any, time_stamp: float, quality: AttrQuality) → None

push_alarm_event(self, attr_name: str, str_data: str, data: Any) → None

push_alarm_event(self, attr_name: str, str_data: str, data: Any, time_stamp: float, quality: AttrQuality) → None

Push an alarm event for the given attribute name.

Parameters:

attr_name (str) – attribute name
data (typing.Any) – the data to be sent as attribute event data. Data must be compatible with the attribute type and format. for SPECTRUM and IMAGE attributes, data can be any type of sequence of elements compatible with the attribute type
str_data (str) – special variation for DevEncoded data type. In this case ‘data’ must be a str or an object with the buffer interface.
except (Exception | DevFailed) – Instead of data, you may want to send an exception.
time_stamp (float) – the time stamp
quality (AttrQuality) – the attribute quality factor

Raises:

DevFailed: If the attribute data type is not coherent.

Changed in version 10.1.0: Removed optional ‘dim_x’ and ‘dim_y’ arguments. The dimensions are automatically determined from the data.

push_att_conf_event(self: tango._tango.DeviceImpl, attr: tango._tango.Attribute) → None

Push an attribute configuration event.

Parameters:: attr (Attribute) – the attribute for which the configuration event will be sent.

Added in version 7.2.1.

push_data_ready_event(self: tango._tango.DeviceImpl, attr_name: str, counter: SupportsInt | SupportsIndex) → None

Push a data ready event for the given attribute name.

The method needs the attribute name and a “counter” which will be passed within the event

Parameters:

attr_name (str) – attribute name
counter (int) – the user counter

Raises:

DevFailed: If the attribute name is unknown

push_event(self, attr_name: str, filt_names, filt_vals, except: Exception)

push_event(self, attr_name: str, filt_names, filt_vals, data: Any) → None

push_event(self, attr_name: str, filt_names, filt_vals, str_data: str, data: Any) → None

push_event(self, attr_name: str, filt_names, filt_vals, data: Any, time_stamp: float, quality: AttrQuality) → None

push_event(self, attr_name: str, filt_names, filt_vals, str_data: str, data: Any, time_stamp: float, quality: AttrQuality) → None

Push a user event for the given attribute name.

Parameters:

attr_name (str) – attribute name
filt_names (:py:obj:list[str]) – unused (kept for backwards compatibility) - pass an empty list.
filt_vals (:py:obj:list[float]) – unused (kept for backwards compatibility) - pass an empty list.
data (typing.Any) – the data to be sent as attribute event data. Data must be compatible with the attribute type and format. for SPECTRUM and IMAGE attributes, data can be any type of sequence of elements compatible with the attribute type
str_data (str) – special variation for DevEncoded data type. In this case ‘data’ must be a str or an object with the buffer interface.
except (Exception | DevFailed) – Instead of data, you may want to send an exception.
time_stamp (float) – the time stamp
quality (AttrQuality) – the attribute quality factor

Raises:

DevFailed: If the attribute data type is not coherent.

Changed in version 10.1.0: Removed optional ‘dim_x’ and ‘dim_y’ arguments. The dimensions are automatically determined from the data.

read_adminMode() → AdminMode | tuple[AdminMode, float, AttrQuality]

Read the admin mode of the device.

Subclasses can override this to change the behaviour of the adminMode attribute.

read_attr_hardware(self: tango._tango.Device_6Impl, attr_list: tango._tango.StdLongVector) → None

Read the hardware to return attribute value(s).

Default method to implement an action necessary on a device to read the hardware involved in a read attribute CORBA call. This method must be redefined in sub-classes in order to support attribute reading

Parameters:: attr_list (list[int]) – list of indices in the device object attribute vector of an attribute to be read.
Raises:: DevFailed: This method does not throw exception but a redefined method can

read_commandedObsState() → ObsState | tuple[ObsState, float, AttrQuality]

Read the commanded observation state of the device.

Subclasses can override this to change the behaviour of the commandedObsState attribute.

read_commandedState() → str | tuple[str, float, AttrQuality]

Read the commanded state of the device.

Subclasses can override this to change the behaviour of the commandedState attribute.

read_configurationDelayExpected() → int | tuple[int, float, AttrQuality]

Read the expected configuration delay of the device.

Subclasses can override this to change the behaviour of the configurationDelayExpected attribute.

read_configurationProgress() → int | tuple[int, float, AttrQuality]

Read the configuration progress of the device.

Subclasses can override this to change the behaviour of the configurationProgress attribute.

read_healthState() → HealthState | tuple[HealthState, float, AttrQuality]

Read the health state of the device.

Subclasses can override this to change the behaviour of the healthState attribute.

read_obsMode() → ObsMode | tuple[ObsMode, float, AttrQuality]

Read the observation mode of the device.

Subclasses can override this to change the behaviour of the obsMode attribute.

read_obsState() → ObsState | tuple[ObsState, float, AttrQuality]

Read the observation state of the device.

Subclasses can override this to change the behaviour of the obsState attribute.

register_signal(self: tango._tango.DeviceImpl, signo: SupportsInt | SupportsIndex, own_handler: bool = False) → None

Register this device as device to be informed when signal signo is sent to to the device server process

Parameters:

signo (int) – signal identifier
own_handler (bool) – true if you want the device signal handler to be executed in its own handler instead of being executed by the signal thread. If this parameter is set to true, care should be taken on how the handler is written. A default false value is provided

remove_attribute(attr_name: str, free_it: bool = False, clean_db: bool = True) → None

Remove one attribute from the device attribute list.

Note: Call of synchronous remove_attribute method from a coroutine function in an asyncio server may cause a deadlock. Use await async_remove_attribute() instead. However, if overriding the synchronous method initialize_dynamic_attributes, then the synchronous remove_attribute method must be used, even in asyncio servers.

Parameters:

attr_name (str) – attribute name
free_it (bool) – free Attr object flag. Default False
clean_db (bool) – clean attribute related info in db. Default True

Raises:

DevFailed

Added in version 9.5.0: free_it parameter. clean_db parameter.

remove_command(cmd_name: str, free_it: bool = False, clean_db: bool = True) → None

Remove one command from the device command list.

Parameters:

cmd_name (str) – command name to be removed from the list
free_it (bool) – set to true if the command object must be freed.
clean_db (bool) – Clean command related information (included polling info if the command is polled) from database.

Raises:

DevFailed

classmethod run(args=None, **kwargs) → Any | None

Run the class as a Tango device server.

Parameters:

*args – Arbitrary positional arguments to be passed to the Tango run_server method
**kwargs – Arbitrary keyword arguments to be passed to the Tango run_server method

classmethod run_server(args=None, **kwargs)

Run the class as a device server. It is based on the tango.server.run method.

The difference is that the device class and server name are automatically given.

Parameters:

args (iterable) – args as given in the tango.server.run method without the server name. If None, the sys.argv list is used
kwargs – the other keywords argument are as given in the tango.server.run method.

server_init_hook(): Tango server_init_hook. Called once the device server admin device (DServer) is exported. Default implementation does nothing.

set_alarm_event(self: tango._tango.DeviceImpl, attr_name: str, implemented: bool, detect: bool = True) → None

Set an implemented flag for the attribute to indicate that the server fires alarm events manually, without the polling to be started.

If the detect parameter is set to true, the criteria specified for the alarm event (rel_change and abs_change) are verified and the event is only pushed if a least one of them are fulfilled (change in value compared to previous event exceeds a threshold). If detect is set to false the event is fired without any value checking!

Parameters:

attr_name (str) – attribute name
implemented (bool) – True when the server fires alarm events manually.
detect (bool) – Triggers the verification of the alarm event properties when set to true. Default value is true.

Added in version 10.0.0.

set_archive_event(self: tango._tango.DeviceImpl, attr_name: str, implemented: bool, detect: bool = True) → None

Set an implemented flag for the attribute to indicate that the server fires archive events manually, without the polling to be started.

If the detect parameter is set to true, the criteria specified for the archive event (rel_change and abs_change) are verified and the event is only pushed if a least one of them are fulfilled (change in value compared to previous event exceeds a threshold). If detect is set to false the event is fired without any value checking!

Parameters:

attr_name (str) – attribute name
implemented (bool) – True when the server fires change events manually.
detect (bool) – Triggers the verification of the change event properties when set to true. Default value is true.

set_attribute_config(self: tango._tango.DeviceImpl, new_conf: list[AttributeConfig]) → None

Sets attribute configuration locally and in the Tango database

Parameters:: new_conf (list[tango.AttributeConfig]) – The new attribute(s) configuration. One AttributeConfig structure is needed for each attribute to update

Added in version 10.0.0.

set_attribute_config_3(self: tango._tango.Device_3Impl, new_conf: list[AttributeConfig_3]) → None

Sets attribute configuration locally and in the Tango database

Parameters:: new_conf (list[tango.AttributeConfig_3]) – The new attribute(s) configuration. One AttributeConfig structure is needed for each attribute to update

set_data_ready_event(self: tango._tango.DeviceImpl, attr_name: str, implemented: bool) → None

Set an implemented flag for the attribute to indicate that the server fires data ready events manually.

Parameters:

attr_name (str) – attribute name
implemented (bool) – True when the server fires change events manually.

set_logging_level(value: LoggingLevel) → None

Set the logging level for the device.

Both the Python logger and the Tango logger are updated.

Parameters:: value – Logging level for logger
Raises:: LoggingLevelError – for invalid value

set_logging_targets(targets: list[str]) → None

Set the additional logging targets for the device.

Note that this excludes the handlers provided by the ska_ser_logging library defaults.

Parameters:: targets – Logging targets for logger

property shared_bus: BusProtocol: The shared bus used by this SharingObserver.

signal_handler(self: tango._tango.Device_3Impl, signo: SupportsInt | SupportsIndex) → None

Signal handler.

The method executed when the signal arrived in the device server process. This method is defined as virtual and then, can be redefined following device needs.

Parameters:: signo (int) – the signal number
Raises:: DevFailed: This method does not throw exception but a redefined method can

simulationMode: attribute_from_signal

Simulation mode attribute of the device.

Some devices may implement both modes, while others will have simulators that set simulationMode to True while the real devices always set simulationMode to False.

simulation_mode

1 if the device is deployed in simulation mode, 0 otherwise.

Type:: device_property(dtype=int)

start_logging(self: tango._tango.DeviceImpl) → None: Starts logging

stop_logging(self: tango._tango.DeviceImpl) → None: Stops logging

stop_poll_attribute(self: tango._tango.DeviceImpl, attr_name: str) → None

Remove an attribute from the list of polled attributes.

Parameters:: attr_name (str) – attribute name

stop_poll_command(self: tango._tango.DeviceImpl, cmd_name: str) → None

Remove a command from the list of polled commands.

Parameters:: cmd_name (str) – cmd_name name

stop_polling(*args, **kwargs)

Overloaded function.

stop_polling(self: tango._tango.DeviceImpl) -> None

Stop all polling for a device. If the device is polled, call this method before deleting it.

Added in version 7.1.2.
stop_polling(self: tango._tango.DeviceImpl, with_db_upd: bool) -> None

Stop all polling for a device. If the device is polled, call this method before deleting it.

param with_db_upd:

Is it necessary to update db?

type with_db_upd:

bool

Added in version 7.1.2.

submit_lrc_task(name: str, task: TaskFunctionType, *, args: list[Any] | None = None, kwargs: dict[str, Any] | None = None, task_callback: TaskCallbackType | None = None, fisallowed: Callable[[LRCReqType], bool] | str | None = None) → tuple[TaskStatus, str]

Submit a long running command task to the task_executor.

The f”is_{name}_allowed” method is looked up and, if found, will be called with LRCReqType.EXECUTE_REQ immediately before the task is executed.

Parameters:

task – task to be executed
args – positional arguments to be passed to the task
kwargs – keyword arguments to be passed to the task
fisallowed – override the is_cmd_allowed method used

Returns:

TaskStatus, message

testMode: attribute_from_signal

Test mode attribute of the device.

Either no test mode or an indication of the test mode.

unregister_signal(self: tango._tango.DeviceImpl, signo: SupportsInt | SupportsIndex) → None

Unregister this device as device to be informed when signal signo is sent to to the device server process

Parameters:: signo (int) – signal identifier

versionId

Read the Version Id of the device.

Returns:: the version id of the device

warn_stream(msg: str, *args, source: Callable | None = None) → None

Sends the given message to the tango warn stream.

Since PyTango 7.1.3, the same can be achieved with:

print(msg, file=self.log_warn)

Parameters:

msg (str) – the message to be sent to the warn stream
*args – Arguments to format a message string.
source (typing.Callable) – Function that will be inspected for filename and lineno in the log message.

Added in version 9.4.2: added source parameter

write_adminMode(mode: AdminMode) → None

Set the Admin Mode of the device.

Subclasses can override this to change the behaviour of the adminMode attribute.

Parameters:

value – Admin Mode of the device.

Raises:

ValueError – for unknown adminMode
StateModelError – for a disallowed transition

write_attr_hardware(self: tango._tango.Device_6Impl, attr_list: tango._tango.StdLongVector) → None

Write the hardware for attributes.

Default method to implement an action necessary on a device to write the hardware involved in a write attribute. This method must be redefined in sub-classes in order to support writable attribute

Parameters:: attr_list (list[int]) – list of indices in the device object attribute vector of an attribute to be written.
Raises:: DevFailed: This method does not throw exception but a redefined method can

admin_mode_model: ska_control_model.AdminModeModel: State model used to ensure that admin mode transitions are allowed.

op_state_model: ska_control_model.OpStateModel: State model used to ensure that operational state transitions are allowed.

VCCAllBandsComponentManager Class

class ska_mid_cbf_fhs_vcc.vcc_all_bands.vcc_all_bands_component_manager.VCCAllBandsComponentManager(*args: Any, device, logger: Logger, simulation_mode: SimulationMode = SimulationMode.FALSE, attr_change_callback: Callable[[str, Any], None] | None = None, attr_archive_callback: Callable[[str, Any], None] | None = None, health_state_callback: Callable[[HealthState], None] | None = None, obs_state_action_callback: Callable[[str], None] | None = None, obs_command_running_callback: Callable[[str, bool], None] | None = None, emulation_mode: bool = False, create_log_file: bool = True, long_running_command_result_buffer_max_size=100, **kwargs: Any)[source]

Bases: FhsControllerComponentManagerBase, ObsDeviceComponentManager

Component manager for the VCC All Bands Controller device.

Parameters:

*args (Any) – Any arbitrary positional arguments to pass to the superclass constructor.
device (FhsBaseDevice) – A reference to the controller device instance.
logger (logging.Logger) – A logger instance to be used by this component manager.
simulation_mode (SimulationMode, optional) – Whether the controller is deployed in simulation mode (SimulationMode.TRUE) or not (SimulationMode.FALSE). Default is SimulationMode.FALSE.
attr_change_callback (Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute change event occurs. Default is None.
attr_archive_callback (Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute archive event occurs. Default is None.
health_state_callback (Callable[[HealthState], None] | None, optional) – Callback that is called when a HealthState change occurs. Default is None.
obs_state_action_callback (Callable[[str, bool], None] | None, optional) – Callback that is called when the controller’s observation state changes. Default is None.
obs_command_running_callback (Callable[[str, bool], None] | None, optional) – Callback that is called when an observing command starts running. Default is None.
emulation_mode (bool, optional) – Whether the controller is deployed in emulation mode or not. Default is False.
create_log_file (bool, optional) – Whether or not to create a log file for this controller. Default is True.
**kwargs (Any) – Any arbitrary keyword arguments to pass to the superclass init method.

subarray_id: int

The ID of the subarray assigned to this VCC.

Type:: int

expected_dish_id: str | None

The ID of the dish this VCC is expecting data from.

Type:: int

frequency_band: FrequencyBandEnum

The frequency band in which this VCC is operating.

Type:: FrequencyBandEnum

frequency_band_offset: list[int]

The frequency band offset(s) configured for this VCC. Contains two offset values, however the second value is only meaningful when operating in band 5. Only the first value is required in all other bands.

Type:: list[int]

input_sample_rate: int

The input sample rate of this VCC.

Type:: int

vcc_gains: list[float]

The currently applied gain multipliers for VCC coarse channels.

Type:: list[float]

last_requested_headrooms: list[float]

The requested headrooms from the most recent call to AutoSetFilterGains.

Type:: list[float]

ethernet_200g: FtileEthernetManager

The IP block manager for the F-tile Ethernet block.

Type:: FtileEthernetManager

b123_vcc: B123VccOsppfbChannelizerManager

The IP block manager for the B123 Channelizer.

Type:: B123VccOsppfbChannelizerManager

frequency_slice_selection: FrequencySliceSelectionManager

The IP block manager for the Frequency Slice Selection block.

Type:: FrequencySliceSelectionManager

packet_validation: PacketValidationManager

The IP block manager for the Packet Validation block.

Type:: PacketValidationManager

wideband_frequency_shifter: WidebandFrequencyShifterManager

The IP block manager for the Wideband Frequency Shifter.

Type:: WidebandFrequencyShifterManager

wideband_input_buffer: WidebandInputBufferManager

The IP block manager for the Wideband Input Buffer.

Type:: WidebandInputBufferManager

vcc_stream_merges: dict[int, VCCStreamMergeManager]

Dictionary containing the IP block managers for the two VCC Stream Merge blocks, mapped by index (1 or 2).

Type:: dict[int, VCCStreamMergeManager]

wideband_power_meters: dict[VCCBandGroup | int, WidebandPowerMeterManager]

Dictionary containing the IP block managers for all Wideband Power Meters, mapped by either band group (B123, etc) or FS index (1 to 26).

Type:: dict[VCCBandGroup | int, WidebandPowerMeterManager]

property config_schema: dict[str, Any]: The ConfigureScan input JSON schema for the VCC All Bands Controller.

property config_dataclass: type[VCCAllBandsConfigureScanConfig]: The ConfigureScan input dataclass for the VCC All Bands Controller.

obs_state

The current observation state of this controller.

Type:: ObsState

update_subarray_membership(argin: int, task_callback: Callable | None = None) → tuple[TaskStatus, str][source]

Submit the task to start running the UpdateSubarrayMembership command implementation.

Parameters:

argin (int) – The subarray ID.
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]

is_allowed(error_msg: str, obs_states: list[ObsState]) → bool[source]

Determine whether the current ObsState is in the provided list of states, and log a warning message if not.

Returns:: True if the current ObsState is in the list provided, False otherwise.
Return type:: bool

is_go_to_idle_allowed() → bool[source]

Determine whether the GoToIdle command is allowed from the current ObsState.

Returns:: True if the GoToIdle command is allowed, False otherwise.
Return type:: bool

is_obs_reset_allowed() → bool[source]

Determine whether the ObsReset command is allowed from the current ObsState.

Returns:: True if the ObsReset command is allowed, False otherwise.
Return type:: bool

go_to_idle(argin: str, task_callback: Callable | None = None) → tuple[TaskStatus, str][source]

Submit the task to start running the GoToIdle command implementation.

Parameters:

argin (str) – The go_to_idle schema JSON string from the command’s input argument.
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]

obs_reset(argin: str | None = None, task_callback: Callable | None = None) → tuple[TaskStatus, str][source]

Submit the task to start running the ObsReset command implementation.

Parameters:

argin (str) – The Transaction id from the command’s input argument, can be none
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]

abort_commands(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str][source]

Run a task to abort all commands and stop any started IP blocks.

Parameters:: task_callback (Optional[Callable], optional) – A callback to pass to the superclass to be run when the task status changes. Default is None.
Returns:: The status of the task and an informative message string.
Return type:: tuple[TaskStatus, str]

_update_communication_state(communication_state: CommunicationStatus) → None

Handle a change in communication status.

This is a helper method for use by subclasses.

Parameters:: communication_state – the new communication status of the component manager.

_update_component_state(**kwargs: Any) → None

Handle a change in component state.

This is a helper method for use by subclasses.

Parameters:: kwargs – key/values for state

abort(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Abort activities on the device.

Parameters:: task_callback – callback to be called whenever the status of the task changes.
Returns:: tuple of TaskStatus & message

clear_logging_level_overrides() → tuple[ResultCode, str]

Clear all IP block logging level overrides. This is the implementation for the ClearLoggingLevelOverrides command.

Returns:: The Tango result code, and an informative message string.
Return type:: tuple[ResultCode, str]

property communication_state: CommunicationStatus

Return the communication status of this component manager.

Returns:: status of the communication channel with the component.

property component_state: dict[str, Any]

Return the state of this component manager’s component.

Returns:: state of the component.

configure_scan(argin: str, task_callback: Callable | None = None) → tuple[TaskStatus, str]

Submit the task to start running the ConfigureScan command implementation.

Parameters:

argin (str) – The configuration JSON string from the command’s input argument.
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]

end_scan(argin: str | None = None, task_callback: Callable | None = None) → tuple[TaskStatus, str]

Submit the task to start running the EndScan command implementation.

Parameters:

argin (str) – The end scan schema JSON string from the command’s input argument.
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]

property faulty: bool | None

Whether this component manager is currently experiencing a fault.

Type:: bool

get_device_health_state() → HealthState

Get the health state of the device.

Returns:: HealthState The health state of the device.

get_long_running_command_result(transaction_id: str | None = None) → tuple[ResultCode, dict[str, LongRunningCommandResult | None]]

Fetch the result(s) for the LRC with the specified transaction_id. If the provided transaction_id is None, fetch all the stored results. This is the implementation for the GetLongRunningCommandResult command.

Parameters:: transaction_id (str) – The transaction_id of the LRC.
Returns:: The Tango result code, and a dictionary containing the result(s) of the LRCs.
Return type:: tuple[ResultCode, dict[str, LongRunningCommandResult | None]]

get_status(ip_blocks: list[str]) → tuple[ResultCode, dict[str, dict]]

Fetch the status(es) from the specified IP block(s). This is the implementation for the GetStatus command.

Parameters:: ip_blocks (list[str]) – The list of IP blocks for which to fetch the status.
Returns:: The Tango result code, and a dictionary containing all of the requested IP block statuses.
Return type:: tuple[ResultCode, dict[str, dict]]

log_critical(msg: str, transaction_id: str | None = None) → None

log_debug(msg: str, transaction_id: str | None = None) → None

log_error(msg: str, transaction_id: str | None = None) → None

log_info(msg: str, transaction_id: str | None = None) → None

log_warning(msg: str, transaction_id: str | None = None) → None

property max_executing_tasks: int

Get the max number of tasks that can be executing at once.

Returns:: max number of simultaneously executing tasks.

off(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Turn the component off.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

on(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Turn the component on.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

reset(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Reset the component (from fault state).

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

scan(argin: str, task_callback: Callable | None = None) → tuple[TaskStatus, str]

Submit the task to start running the Scan command implementation.

Parameters:

argin (str) – The scan schema JSON string from the command’s input argument.
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]

set_fault_and_failed() → None: Set the component state to faulty and update its health state to FAILED. This is to be called when an exception occurs in the component manager.

standby(task_callback: TaskCallbackType | None = None) → tuple[TaskStatus, str]

Put the component into low-power standby mode.

Parameters:: task_callback – callback to be called when the status of the command changes
Raises:: NotImplementedError – Not implemented it’s an abstract class

start_communicating() → None: Establish communication with the component, then start monitoring.

stop_communicating() → None: Close communication with the component, then stop monitoring.

task_abort_event_is_set(command_name: str, task_callback: Callable, task_abort_event: Event) → bool

Helper method for checking task abort event during command thread.

Parameters:

command_name (str) – name of command for result message
task_callback (Callable) – command tracker update_command_info callback
task_abort_event (Event) – task executor abort event

Returns:

True if abort event is set, otherwise False

Return type:

bool

update_device_health_state(health_state: HealthState) → None

Handle a health state change. This is a helper method for use by subclasses.

Parameters:: health_state (HealthState) – The new health state of the component manager.

update_global_logging_level(logging_level: str) → tuple[ResultCode, str]

Update the controller’s global logging level. This is the implementation for the UpdateGlobalLoggingLevel command.

Parameters:: logging_level (str) – The new logging level.
Returns:: The Tango result code, and an informative message string.
Return type:: tuple[ResultCode, str]

update_ip_block_logging_levels(input_json: str) → tuple[ResultCode, str]

Assign a specified logging level override to all specified IP block(s). This is the implementation for the UpdateIPBlockLoggingLevels command.

Parameters:: input_json (str) – A JSON string in the format: {“ip_blocks”: [“SomeIPBlockIDOne”, “SomeIPBlockIDTwo”], “level”: “DEBUG”}
Returns:: The Tango result code, and an informative message string.
Return type:: tuple[ResultCode, str]

ip_block_list: list[str]

List of IP block IDs managed by this controller.

Type:: list[str]

ip_block_aliases: dict[str, list[str]]

Mapping of IP block aliases to their base IDs.

Type:: dict[str, list[str]]

ip_block_manager_map: dict[str, BaseIPBlockManager]

Mapping of IP block IDs and aliases to their respective manager classes.

Type:: dict[str, BaseIPBlockManager]

transaction_ids_per_command: dict[CommandType, str]

Dictionary to store the latest transaction ids for each command

Type:: dict[CommandType, str]

device

Reference to the controller device instance.

Type:: FhsControllerBaseDevice

simulation_mode

Whether the controller is deployed in simulation mode.

Type:: SimulationMode

emulation_mode

Whether the controller is deployed in emulation mode.

Type:: bool

long_running_command_result_buffer

Buffer to store the result of long running commands

Type:: LongRunningCommandResultBuffer

health_monitor

The health state monitor for this controller.

Type:: FhsHealthMonitor

state

The current DevState of the device.

Type:: DevState

auto_set_filter_gains(argin: str, task_callback: Callable | None = None) → tuple[TaskStatus, str][source]

Submit the task to start running the AutoSetFilterGains command implementation.

Parameters:

argin (str) – JSON String following the auto set filter gains command schema Requested RFI headroom, in decibels (dB). Must be a list containing either a single value to apply to all frequency slices, or a value per frequency slice to be applied separately.
task_callback (Optional[Callable], optional) – A callback to run when the task status changes. Default is None.

Returns:

The status of the task and an informative message string.

Return type:

tuple[TaskStatus, str]