# -*- coding: utf-8 -*-
#
# This file is part of the CspSubElementMaster project
#
# INAF-SKA Telescope
#
# Distributed under the terms of the GPL license.
# See LICENSE.txt for more info.
""" CSP.LMC Sub-element Master
A base class for the Master of a SKA Sub-element.
"""
# PyTango imports
import tango
from tango import DebugIt
from tango.server import run
from tango.server import Device
from tango.server import attribute, command
from tango.server import device_property
from tango import AttrQuality, DispLevel, DevState
from tango import AttrWriteType, PipeWriteType
from collections import defaultdict
# Additional import
# PROTECTED REGION ID(CspSubElementMaster.additionnal_import) ENABLED START #
from ska.base import SKAMaster
from ska.base.control_model import HealthState, AdminMode, LoggingLevel
from csp_lmc_common.utils.cspcommons import CmdExecState
from csp_lmc_common.utils.decorators import AdminModeCheck
from .decorators import IsMasterCommandAllowed
from . import release
# PROTECTED REGION END # // CspSubElementMaster.additionnal_import
__all__ = ["CspSubElementMaster", "main"]
class CspSubElementMaster(SKAMaster):
"""
A base class for the Master of a SKA Sub-element.
**Properties:**
- Device Property
Racks
- The list with the FQDNs of the sub-element racks devices.
- Type:'DevVarStringArray'
"""
# PROTECTED REGION ID(CspSubElementMaster.class_variable) ENABLED START #
# PROTECTED REGION END # // CspSubElementMaster.class_variable
# -----------------
# Device Properties
# -----------------
Racks = device_property(
dtype='DevVarStringArray',
)
# ----------
# Attributes
# ----------
numOfDevCompletedTask = attribute(
dtype='DevUShort',
label="Number of devices that completed the task",
doc="Number of devices that completed the task",
)
onCmdFailure = attribute(
dtype='DevBoolean',
label="CBF command failure flag",
#polling_period=1000,
doc="Failure flag set when the On command fails with error(s).",
)
onFailureMessage = attribute(
dtype='DevString',
label="On execution failure message",
doc="Failure message when the On command fails with error(s).",
)
offCmdFailure = attribute(
dtype='DevBoolean',
label="Off execution failure flag",
doc="Failure flag set when the Off command fails with error(s).",
)
offFailureMessage = attribute(
dtype='DevString',
label="Off execution failure message",
doc="Failure message when the Off command fails with error(s).",
)
standbyCmdFailure = attribute(
dtype='DevBoolean',
label="Standby execution failure message",
doc="Failure flag set when the Standby command fails with error(s).",
)
standbyFailureMessage = attribute(
dtype='DevString',
label="Standby execution failure message",
doc="Failure message when the Standby command fails with error(s).",
)
onCommandProgress = attribute(
dtype='DevUShort',
label="Progress percentage for the On command",
abs_change=10,
max_value=100,
min_value=0,
doc=("Percentage progress implemented for commands that result in state/mode"
" transitions for a large number of components and/or are executed in "
"stages (e.g power up, power down)"),
)
offCommandProgress = attribute(
dtype='DevUShort',
label="Progress percentage for the Off command",
abs_change=10,
max_value=100,
min_value=0,
doc=("Percentage progress implemented for commands that result in state/mode transitions"
" for a large number of components and/or are executed in stages"
" (e.g power up, power down)"),
)
standbyCommandProgress = attribute(
dtype='DevUShort',
label="Progress percentage for the Standby command",
abs_change=10,
max_value=100,
min_value=0,
doc=("Percentage progress implemented for commands that result in state/mode"
" transitions for a large number of components and/or are executed in"
" stages (e.g power up, power down)"),
)
onCmdDurationExpected = attribute(
dtype='DevUShort',
access=AttrWriteType.READ_WRITE,
label="Expected duration (sec) of the On command execution",
abs_change=0,
max_value=100,
min_value=0,
memorized=True,
doc="Set/Report the duration expected for the On command execution",
)
offCmdDurationExpected = attribute(
dtype='DevUShort',
access=AttrWriteType.READ_WRITE,
label="Expected duration (sec) of the Off command",
abs_change=0,
max_value=100,
min_value=0,
memorized=True,
doc="Set/Report the duration expected for the Off command execution",
)
standbyCmdDurationExpected = attribute(
dtype='DevUShort',
access=AttrWriteType.READ_WRITE,
label="Expected duration (sec) of the Standby command",
abs_change=0,
max_value=100,
min_value=0,
memorized=True,
doc="Set/Report the duration expected for the Standby command",
)
onCmdDurationMeasured = attribute(
dtype='DevUShort',
label="Measured duration (sec) of the On command execution",
abs_change=0,
max_value=100,
min_value=0,
doc="Report the measured duration of the On command execution",
)
offCmdDurationMeasured = attribute(
dtype='DevUShort',
label="Measured duration (sec) of the Off command",
abs_change=0,
max_value=100,
min_value=0,
doc="Report the measured duration of the Off command execution",
)
standbyCmdDurationMeasured = attribute(
dtype='DevUShort',
label="Measured duration (sec) of the Standby command",
abs_change=0,
max_value=100,
min_value=0,
doc="Report the measured duration of the Standby command",
)
onCmdTimeoutExpired = attribute(
dtype='DevBoolean',
label="On execution timeout flag",
doc="Signal the occurence of a timeout during the execution of the on command.",
)
offCmdTimeoutExpired = attribute(
dtype='DevBoolean',
label="Off execution timeout flag",
doc="Signal the occurence of a timeout during the execution of the Off command.",
)
standbyCmdTimeoutExpired = attribute(
dtype='DevBoolean',
label="Standby execution timeout flag.",
doc="Signal the occurence of a timeout during the execution of the Standby command.",
)
listOfDevCompletedTask = attribute(
dtype=('DevString',),
max_dim_x=100,
label="List of devices that completed the task",
doc="List of devices that completed the task",
)
listOfComponents = attribute(
dtype=('DevString',),
max_dim_x=100,
label="List of sub-element components",
doc="The list o the FQDNs of the sub-element components.",
)
# ---------------
# General methods
# ---------------
def init_device(self):
"""Initialises the attributes and properties of the CspSubElementMaster."""
SKAMaster.init_device(self)
self.set_state(tango.DevState.INIT)
# PROTECTED REGION ID(CspSubElementMaster.init_device) ENABLED START #
# _cmd_execution_state: implement the execution state of a long-running
# command for the whole CSP sub-element. Setting this attribute prevent the execution
# of the same command while it is already running.
# Implemented as a Python default dictionary:
# keys: command name
# values:command state
self._cmd_execution_state = defaultdict(lambda: CmdExecState.IDLE)
# _cmd_progress: report the execution progress of a long-running command
# Implemented as a Python dictionary:
# keys: command name in lower case ('on', 'off'..)
# values: the percentage
self._cmd_progress = defaultdict(lambda: 0)
# _cmd_duration_expected: store the duration (in sec.) configured for
# a long-running command
# Implemented as Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: the duration (in sec)
self._cmd_duration_expected = defaultdict(lambda: 30)
# _cmd_duration_measured: report the measured duration (in sec.) for
# a long-running command
# Implemented as Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: the duration (in sec)
self._cmd_duration_measured = defaultdict(lambda: 0)
# _timeout_expired: report the timeout flag
# Implemented as a Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: True/False
self._timeout_expired = defaultdict(lambda: False)
# _failure_raised: report the failure flag
# Implemented as a Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: True/False
self._failure_raised = defaultdict(lambda: False)
# _failure_message: report the failure message
# Implemented as a Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: the message
self._failure_message = defaultdict(lambda: '')
# _list_dev_completed_task: for each long-running command report the list
# of subordinate sub-element components that completed the task
# Implemented as a Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: the list of components
self._list_dev_completed_task = defaultdict(lambda: [])
# _list_of_components: report the list of subordinate
# sub-element components FQDNs.
# Implemented as a Python list
self._list_of_components = []
# _num_dev_completed_task: for each long-running command report the number
# of subordinate components that completed the task
# Implemented as a Python default dictionary
# keys: command name in lower case ('on', 'off',..)
# values: the number of components
self._num_dev_completed_task = defaultdict(lambda:0)
# the last executed command
self._last_executed_command = "none"
# PROTECTED REGION END # // CspSubElementMaster.init_device
def always_executed_hook(self):
"""Method always executed before any TANGO command is executed."""
# PROTECTED REGION ID(CspSubElementMaster.always_executed_hook) ENABLED START #
# PROTECTED REGION END # // CspSubElementMaster.always_executed_hook
def delete_device(self):
"""Hook to delete resources allocated in init_device.
This method allows for any memory or other resources allocated in the
init_device method to be released. This method is called by the device
destructor and by the device Init command.
"""
# PROTECTED REGION ID(CspSubElementMaster.delete_device) ENABLED START #
# PROTECTED REGION END # // CspSubElementMaster.delete_device
# ------------------
# Attributes methods
# ------------------
def read_numOfDevCompletedTask(self):
# PROTECTED REGION ID(CspSubElementMaster.numOfDevCompletedTask_read) ENABLED START #
"""Return the numOfDevCompletedTask attribute."""
return self._num_dev_completed_task[self._last_executed_command]
# PROTECTED REGION END # // CspSubElementMaster.numOfDevCompletedTask_read
def read_onCmdFailure(self):
# PROTECTED REGION ID(CspSubElementMaster.onCmdFailure_read) ENABLED START #
"""Return the onCmdFailure attribute."""
return self._failure_raised['on']
# PROTECTED REGION END # // CspSubElementMaster.onCmdFailure_read
def read_onFailureMessage(self):
# PROTECTED REGION ID(CspSubElementMaster.onFailureMessage_read) ENABLED START #
"""Return the onFailureMessage attribute."""
return self._failure_message['on']
# PROTECTED REGION END # // CspSubElementMaster.onFailureMessage_read
def read_offCmdFailure(self):
# PROTECTED REGION ID(CspSubElementMaster.offCmdFailure_read) ENABLED START #
"""Return the offCmdFailure attribute."""
return self._failure_raised['off']
# PROTECTED REGION END # // CspSubElementMaster.offCmdFailure_read
def read_offFailureMessage(self):
# PROTECTED REGION ID(CspSubElementMaster.offFailureMessage_read) ENABLED START #
"""Return the offFailureMessage attribute."""
return self._failure_message['off']
# PROTECTED REGION END # // CspSubElementMaster.offFailureMessage_read
def read_standbyCmdFailure(self):
# PROTECTED REGION ID(CspSubElementMaster.standbyCmdFailure_read) ENABLED START #
"""Return the standbyCmdFailure attribute."""
return self._failure_raised['standby']
# PROTECTED REGION END # // CspSubElementMaster.standbyCmdFailure_read
def read_standbyFailureMessage(self):
# PROTECTED REGION ID(CspSubElementMaster.standbyFailureMessage_read) ENABLED START #
"""Return the standbyFailureMessage attribute."""
return self._failure_message['standby']
# PROTECTED REGION END # // CspSubElementMaster.standbyFailureMessage_read
def read_onCommandProgress(self):
# PROTECTED REGION ID(CspSubElementMaster.onCommandProgress_read) ENABLED START #
"""Return the onCommandProgress attribute."""
return self._cmd_progress['on']
# PROTECTED REGION END # // CspSubElementMaster.onCommandProgress_read
def read_offCommandProgress(self):
# PROTECTED REGION ID(CspSubElementMaster.offCommandProgress_read) ENABLED START #
"""Return the offCommandProgress attribute."""
return self._cmd_progress['off']
# PROTECTED REGION END # // CspSubElementMaster.offCommandProgress_read
def read_standbyCommandProgress(self):
# PROTECTED REGION ID(CspSubElementMaster.standbyCommandProgress_read) ENABLED START #
"""Return the standbyCommandProgress attribute."""
return self._cmd_progress['standby']
# PROTECTED REGION END # // CspSubElementMaster.standbyCommandProgress_read
def read_onCmdDurationExpected(self):
# PROTECTED REGION ID(CspSubElementMaster.onCmdDurationExpected_read) ENABLED START #
"""Return the onCmdDurationExpected attribute."""
return self._cmd_duration_expected['on']
# PROTECTED REGION END # // CspSubElementMaster.onCmdDurationExpected_read
def write_onCmdDurationExpected(self, value):
# PROTECTED REGION ID(CspSubElementMaster.onCmdDurationExpected_write) ENABLED START #
"""Set the onCmdDurationExpected attribute."""
self._cmd_duration_expected['on'] = value
# PROTECTED REGION END # // CspSubElementMaster.onCmdDurationExpected_write
def read_offCmdDurationExpected(self):
# PROTECTED REGION ID(CspSubElementMaster.offCmdDurationExpected_read) ENABLED START #
"""Return the offCmdDurationExpected attribute."""
return self._cmd_duration_expected['off']
# PROTECTED REGION END # // CspSubElementMaster.offCmdDurationExpected_read
def write_offCmdDurationExpected(self, value):
# PROTECTED REGION ID(CspSubElementMaster.offCmdDurationExpected_write) ENABLED START #
"""Set the offCmdDurationExpected attribute."""
self._cmd_duration_expected['off'] = value
# PROTECTED REGION END # // CspSubElementMaster.offCmdDurationExpected_write
def read_standbyCmdDurationExpected(self):
# PROTECTED REGION ID(CspSubElementMaster.standbyCmdDurationExpected_read) ENABLED START #
"""Return the standbyCmdDurationExpected attribute."""
return self._cmd_duration_expected['standby']
# PROTECTED REGION END # // CspSubElementMaster.standbyCmdDurationExpected_read
def write_standbyCmdDurationExpected(self, value):
# PROTECTED REGION ID(CspSubElementMaster.standbyCmdDurationExpected_write) ENABLED START #
"""Set the standbyCmdDurationExpected attribute."""
self._cmd_duration_expected['standby'] = value
# PROTECTED REGION END # // CspSubElementMaster.standbyCmdDurationExpected_write
def read_onCmdDurationMeasured(self):
# PROTECTED REGION ID(CspSubElementMaster.onCmdDurationMeasured_read) ENABLED START #
"""Return the onCmdDurationMeasured attribute."""
return self._cmd_duration_measured['on']
# PROTECTED REGION END # // CspSubElementMaster.onCmdDurationMeasured_read
def read_offCmdDurationMeasured(self):
# PROTECTED REGION ID(CspSubElementMaster.offCmdDurationMeasured_read) ENABLED START #
"""Return the offCmdDurationMeasured attribute."""
return self._cmd_duration_measured['off']
# PROTECTED REGION END # // CspSubElementMaster.offCmdDurationMeasured_read
def read_standbyCmdDurationMeasured(self):
# PROTECTED REGION ID(CspSubElementMaster.standbyCmdDurationMeasured_read) ENABLED START #
"""Return the standbyCmdDurationMeasured attribute."""
return self._cmd_duration_measured['standby']
# PROTECTED REGION END # // CspSubElementMaster.standbyCmdDurationMeasured_read
def read_onCmdTimeoutExpired(self):
# PROTECTED REGION ID(CspSubElementMaster.onCmdTimeoutExpired_read) ENABLED START #
"""Return the onCmdTimeoutExpired attribute."""
return self._timeout_expired['on']
# PROTECTED REGION END # // CspSubElementMaster.onCmdTimeoutExpired_read
def read_offCmdTimeoutExpired(self):
# PROTECTED REGION ID(CspSubElementMaster.offCmdTimeoutExpired_read) ENABLED START #
"""Return the offCmdTimeoutExpired attribute."""
return self._timeout_expired['off']
# PROTECTED REGION END # // CspSubElementMaster.offCmdTimeoutExpired_read
def read_standbyCmdTimeoutExpired(self):
# PROTECTED REGION ID(CspSubElementMaster.standbyCmdTimeoutExpired_read) ENABLED START #
"""Return the standbyCmdTimeoutExpired attribute."""
return self._timeout_expired['standby']
# PROTECTED REGION END # // CspSubElementMaster.standbyCmdTimeoutExpired_read
def read_listOfDevCompletedTask(self):
# PROTECTED REGION ID(CspSubElementMaster.listOfDevCompletedTask_read) ENABLED START #
"""Return the listOfDevCompletedTask attribute."""
return self._list_dev_completed_task[self._last_executed_command]
# PROTECTED REGION END # // CspSubElementMaster.listOfDevCompletedTask_read
def read_listOfComponents(self):
# PROTECTED REGION ID(CspSubElementMaster.listOfComponents_read) ENABLED START #
"""Return the listOfComponents attribute."""
return self._list_of_components
# PROTECTED REGION END # // CspSubElementMaster.listOfComponents_read
# --------
# Commands
# --------
@command(
dtype_in='DevVarStringArray',
doc_in="The list of sub-element components FQDNs to switch-on or an empty list to switch-on the whole "
"CSP Sub-element."
" "
"If the array length is 0, the command applies to the whole CSP Sub-Element. If the "
"array length is > 1, each array element specifies the FQDN of the"
"CSP SubElement component to switch ON.",
)
@DebugIt()
@IsMasterCommandAllowed()
@AdminModeCheck('On')
def On(self, argin):
# PROTECTED REGION ID(CspSubElementMaster.On) ENABLED START #
"""
Switch-on the CSP sub-element components specified by the input argument. If no argument is
specified, the command is issued on all the CSP sub-element components.
The command is executed if the *AdminMode* is ONLINE or *MAINTENANCE*.
If the AdminMode is *OFFLINE*, *NOT-FITTED* or *RESERVED*, the method throws an
exception.
The CSP sub-element components can be organized as tango-groups or controlled
by some tango device drivers which actas as 'cache'. We call these devices
'racks', even if they can contro a different phisical arrangement.
:param argin: The list of sub-element components FQDNs: if the array length is 0 (empty input \
list), the command applies to the whole CSP Sub-Element. If the array length is > 1, \
each array element specifies the FQDN of the CSP SubElement component to switch ON.
:type: 'DevVarStringArray'
:return: None
"""
pass
# PROTECTED REGION END # // CspSubElementMaster.On
@command(
dtype_in='DevVarStringArray',
doc_in="If the array length is 0, the command applies to the whole"
"CSP Sub-element."
"If the array length is > 1, each array element specifies the FQDN of the"
"CSP SubElement component to switch OFF.",
)
@DebugIt()
@IsMasterCommandAllowed()
@AdminModeCheck('Off')
def Off(self, argin):
# PROTECTED REGION ID(CspSubElementMaster.Off) ENABLED START #
"""
Switch-off the CSP sub-element components specified by the input argument.
If no argument is specified, the command is issued to all the CSP
sub-element components.
The CSP sub-element components can be organized as tango-groups or controlled
by some tango device drivers which actas as 'cache'. We call these devices
'racks', even if they can contro a different phisical arrangement.
:param argin: The list of sub-element components FQDNs: if the array length is 0 (no list \
specified), the command applies to the whole CSP sub-element. If the array length \
is > 1, each array element specifies the FQDN of a CSP SubElement component to switch \
off.
:type: 'DevVarStringArray'
:return: None
"""
pass
# PROTECTED REGION END # // CspSubElementMaster.Off
@command(
dtype_in='DevVarStringArray',
doc_in="If the array length is 0, the command applies to the whole"
"CSP sub-element."
"If the array length is > 1, each array element specifies the FQDN of the"
"CSP SubElement icomponent to put in STANDBY mode.",
)
@DebugIt()
@IsMasterCommandAllowed()
@AdminModeCheck('Standby')
def Standby(self, argin):
# PROTECTED REGION ID(CspSubElementMaster.Standby) ENABLED START #
"""
Transit the CSP Sub-element or one or more CSP SubElement components from ON/OFF to
STANDBY.
:param argin: The list of sub-element components FQDNs: if the array length is 0 (no list \
specified), the command applies to the whole CSP sub-element. If the array length \
is > 1, each array element specifies the FQDN of a CSP SubElement component to put \
in STANDBY mode.
:type: 'DevVarStringArray'
:return: None
"""
pass
# PROTECTED REGION END # // CspSubElementMaster.Standby
@command(
)
@DebugIt()
def Upgrade(self):
# PROTECTED REGION ID(CspSubElementMaster.Upgrade) ENABLED START #
"""
:return: None
"""
pass
# PROTECTED REGION END # // CspSubElementMaster.Upgrade
# ----------
# Run server
# ----------
[docs]def main(args=None, **kwargs):
"""Main function of the CspSubElementMaster module."""
# PROTECTED REGION ID(CspSubElementMaster.main) ENABLED START #
return run((CspSubElementMaster,), args=args, **kwargs)
# PROTECTED REGION END # // CspSubElementMaster.main
if __name__ == '__main__':
main()