# -*- coding: utf-8 -*-
#
# This file is part of the CspSubElementSubarray project
#
# INAF - SKA Telescope
#
# Distributed under the terms of the GPL license.
# See LICENSE.txt for more info.
""" CSP SubElement Subarray
CSP SubElementbsubarray functionality is modeled via a TANGO Device Class
"""
# 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(CspSubElementSubarray.additionnal_import) ENABLED START #
from ska.base import SKASubarray
from ska.base.control_model import HealthState, AdminMode, ObsState, ObsMode
from csp_lmc_common.utils.cspcommons import CmdExecState
from csp_lmc_common.utils.decorators import AdminModeCheck, ObsStateCheck, SubarrayRejectCmd
from .decorators import IsSubarrayCommandAllowed
from . import release
# PROTECTED REGION END # // CspSubElementSubarray.additionnal_import
__all__ = ["CspSubElementSubarray", "main"]
class CspSubElementSubarray(SKASubarray):
"""
CSP SubElementbsubarray functionality is modeled via a TANGO Device Class
**Properties:**
- Device Property
Master
- The TANGO address of the CSP SubElement Master.
- Type:'DevString'
"""
# PROTECTED REGION ID(CspSubElementSubarray.class_variable) ENABLED START #
# PROTECTED REGION END # // CspSubElementSubarray.class_variable
# -----------------
# Device Properties
# -----------------
Master = device_property(
dtype='DevString',
)
# ----------
# Attributes
# ----------
scanID = attribute(
dtype='DevULong64',
access=AttrWriteType.READ_WRITE,
doc="The scan identification number sent as argument of the Scan command.",
)
validScanConfiguration = attribute(
dtype='DevString',
label="Valid Scan Configuration",
doc="Store the last valid scan configuration.",
)
goToIdleDurationExpected = attribute(
dtype='DevUShort',
access=AttrWriteType.READ_WRITE,
label="GoToIdle command duration expected",
doc="The duration expected (in sec) for the EndSB command.",
)
goToIdleDurationMeasured = attribute(
dtype='DevDouble',
label="GoToIdle command duration measured",
doc="The duration measured (in sec) for the EndSB command.",
)
numOfDevCompletedTask = attribute(
dtype=(('DevString',),),
max_dim_x=2, max_dim_y=10,
label="Number of devices that completed the task",
doc="Number of devices that completed the task",
)
goToIdleCmdProgress = attribute(
dtype='DevUShort',
label="GoToIdle command progress percentage",
polling_period=1500,
abs_change=5,
doc="The progress percentage for the EndSB command.",
)
endScanCmdProgress = attribute(
dtype='DevUShort',
label="EndScan command progress percentage",
polling_period=1500,
abs_change=5,
doc="The progress percentage for the EndScan command.",
)
scanCmdProgress = attribute(
dtype='DevUShort',
label="Scan percentage progress",
polling_period=1500,
abs_change=5,
doc="The progress percentage for the Scan command.",
)
timeoutExpiredFlag = attribute(
dtype='DevBoolean',
label="Subarray command execution timeout flag",
polling_period=1000,
doc="The timeout flag for a Subarray command.",
)
failureRaisedFlag = attribute(
dtype='DevBoolean',
label="Subarray failure flag",
polling_period=1000,
doc="The failure flag for a Subarray command.",
)
failureMessage = attribute(
dtype=(('DevString',),),
max_dim_x=2, max_dim_y=10,
label="The failure message",
doc="The failure message",
)
configureScanCmdProgress = attribute(
dtype='DevUShort',
label="ConfigureScan command execution progress",
doc="The progress percentage for the ConfigureScan command.",
)
configureScanDurationMeasured = attribute(
dtype='DevDouble',
label="ConfigureScan execution time",
doc="The measured execution time for the ConfigureScan command",
)
listOfDevCompletedTask = attribute(
dtype=(('DevString',),),
max_dim_x=2, max_dim_y=10,
label="List of devices that completed the task",
doc="List of devices that completed the task",
)
# ---------------
# General methods
# ---------------
def init_device(self):
"""Initialise the attributes and properties of the CspSubElementSubarray."""
SKASubarray.init_device(self)
self.set_state(tango.DevState.INIT)
# PROTECTED REGION ID(CspSubElementSubarray.init_device) ENABLED START #
# _scanID: the scan identification number send as argument of the Scan
# command
# Implemented as unsigned long64.
self._scanID = 0
# _cmd_execution_state: implement the execution state of a long-running
# command for the whole CSP Sub-element subarray. Setting this attribute
# prevents 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('gotoidle', 'configurescan'..)
# 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('gotoidle', 'configurescan'..)
# values: the duration (in sec)
self._cmd_duration_expected = defaultdict(lambda: 30)
# _config_delay_expected: inherited attribute from SKAObsDevice class
self._config_delay_expected = 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('gotoidle', 'configurescan'..)
# values: the duration (in sec)
self._cmd_duration_measured = defaultdict(lambda: 0)
# _timeout_expired: report the timeout flag
# True/False
self._timeout_expired = False
# _failure_raised: report the failure flag
# True/False
self._failure_raised = False
# _failure_message: report the failure message
# Implemented as Python default dictionary
# keys: command name in lower case('gotoidle', 'configurescan'..)
# values: the message
self._failure_message = defaultdict(lambda: '')
# _list_dev_completed_task: for each long-running command report the list of subordinate
# components that completed the task
# Implemented as Python default dictionary
# keys: command name in lower case('gotoidle', 'configurescan'..)
# values: the list of components
self._list_dev_completed_task = defaultdict(lambda: [])
# _num_dev_completed_task: for each long-running command report the number
# of subordinate components that completed the task
# Implemented as Python default dictionary
# keys: command name in lower case('gotoidle', 'configurescan'..)
# values: the number of components completed the task (stored as string).
self._num_dev_completed_task = defaultdict(lambda:'0')
# _valid_scan_coniguration: store the last valid scan configuration (json format)
# Impemented as a string.
self._valid_scan_configuration = ''
# PROTECTED REGION END # // CspSubElementSubarray.init_device
def always_executed_hook(self):
"""Method always executed before any TANGO command is executed."""
# PROTECTED REGION ID(CspSubElementSubarray.always_executed_hook) ENABLED START #
# PROTECTED REGION END # // CspSubElementSubarray.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(CspSubElementSubarray.delete_device) ENABLED START #
# PROTECTED REGION END # // CspSubElementSubarray.delete_device
# ------------------
# Attributes methods
# ------------------
def read_scanID(self):
# PROTECTED REGION ID(CspSubElementSubarray.scanID_read) ENABLED START #
"""Return the scanID attribute."""
return self._scanID
# PROTECTED REGION END # // CspSubElementSubarray.scanID_read
def write_scanID(self, value):
# PROTECTED REGION ID(CspSubElementSubarray.scanID_write) ENABLED START #
"""Set the scanID attribute."""
self._scanID = value
# PROTECTED REGION END # // CspSubElementSubarray.scanID_write
def read_validScanConfiguration(self):
# PROTECTED REGION ID(CspSubElementSubarray.validScanConfiguration_read) ENABLED START #
"""
Return the validScanConfiguration attribute.
*validScanConfiguration* attribute stores the last scan configuration
programmed for the sub-element.
"""
return self._valid_scan_configuration
# PROTECTED REGION END # // CspSubElementSubarray.validScanConfiguration_read
def read_goToIdleDurationExpected(self):
# PROTECTED REGION ID(CspSubElementSubarray.goToIdleDurationExpected_read) ENABLED START #
"""
Return the goToIdleDurationExpected attribute.
*goToIdleDurationExpected* attribute reports the time expected (in sec) for the
execution of the *GoToIdle* command.
The value for this attribute can be configured via the associate *write* method.
"""
return self._cmd_duration_expected['gotoidle']
# PROTECTED REGION END # // CspSubElementSubarray.goToIdleDurationExpected_read
def write_goToIdleDurationExpected(self, value):
# PROTECTED REGION ID(CspSubElementSubarray.goToIdleDurationExpected_write) ENABLED START #
"""
Set the goToIdleDurationExpected attribute.
Configure the time expected (in sec) for the execution of the *GoToIdle* command.
"""
self._cmd_duration_expected['gotoidle'] = value
# PROTECTED REGION END # // CspSubElementSubarray.goToIdleDurationExpected_write
def read_goToIdleDurationMeasured(self):
# PROTECTED REGION ID(CspSubElementSubarray.goToIdleDurationMeasured_read) ENABLED START #
"""Return the goToIdleDurationMeasured attribute.
*goToIdleDurationMeasured* attribute reports the effective execution time for the
*GoToIdle* command.
"""
return self._cmd_duration_measured['gotoidle']
# PROTECTED REGION END # // CspSubElementSubarray.goToIdleDurationMeasured_read
def read_numOfDevCompletedTask(self):
# PROTECTED REGION ID(CspSubElementSubarray.numOfDevCompletedTask_read) ENABLED START #
"""
Return the numOfDevCompletedTask attribute.
The TANGO attribute *numOfDevCompletedTask* is an Image type attribute of strings.
It is represented as a matrix of *(N_rows, N_cols)* where:
- N_rows = image max_dim_y
- N_cols = image max_dim_x
Examples:
*server side*
self._num_dev_completed_task = {'gotoidle':'3', 'configurescan':'10', ...}
image_attr = [[gotoidle','3'], ['configurescan','10']...]
*client side*
>>> num_of_dev_completed_task = proxy.numOfDevCompletedTask
>>> num_of_dev_completed_task
((gotoidle','3'), ('configurescan','10')...)
>>> num_of_dev_completed_task[0][0]
'gotoidle'
>>> num_of_dev_completed_task[0][1]
'3'
num_of_dev_completed_task is a Python nested tuple
- row_index addresses the command name
- col_index addresses the number (string) of completed tasks
"""
image_attr = [[cmd_name, dev_num] for cmd_name, dev_num in self._num_dev_completed_task.items()]
return image_attr
# PROTECTED REGION END # // CspSubElementSubarray.numOfDevCompletedTask_read
def read_goToIdleCmdProgress(self):
# PROTECTED REGION ID(CspSubElementSubarray.goToIdleCmdProgress_read) ENABLED START #
"""
Return the goToIdleCmdProgress attribute.
*goToIdleCmdProgress* attribute stores the execution percentage of the
*GoToIdle* command.
The CSP Subarray device can subscribe to this attribute to monitor and report
the command execution progress.
The attribute assumes values in the range [0, 100].
"""
return self._cmd_progress['gotoidle']
# PROTECTED REGION END # // CspSubElementSubarray.goToIdleCmdProgress_read
def read_endScanCmdProgress(self):
# PROTECTED REGION ID(CspSubElementSubarray.endScanCmdProgress_read) ENABLED START #
"""
Return the endScanCmdProgress attribute.
*endScanCmdProgres* attribute stores the execution percentage of the
*EndScan* command.
The CSP Subarray device can subscribe to this attribute to monitor and report
the command execution progress.
The attribute assumes values in the range [0, 100].
"""
return self._cmd_progress['endscan']
# PROTECTED REGION END # // CspSubElementSubarray.endScanCmdProgress_read
def read_scanCmdProgress(self):
# PROTECTED REGION ID(CspSubElementSubarray.scanCmdProgress_read) ENABLED START #
"""
Return the scanCmdProgress attribute.
*scanCmdProgress* attribute stores the execution percentage of the
*Scan* command.
The CSP Subarray device can subscribe to this attribute to monitor and report
the command execution progress.
The attribute assumes values in the range [0, 100].
"""
return self._cmd_progress['scan']
# PROTECTED REGION END # // CspSubElementSubarray.scanCmdProgress_read
def read_timeoutExpiredFlag(self):
# PROTECTED REGION ID(CspSubElementSubarray.timeoutExpiredFlag_read) ENABLED START #
"""
Return the timeoutExpiredFlag attribute.
*timeoutExpiredFlag* attribute is of type DevBoolean.
Its value is True when a timeout condition occurs during the execution of
a command.
The Csp Subarray device can subscribe to this attribute to monitor a timeout
condition.
"""
return self._timeout_expired
# PROTECTED REGION END # // CspSubElementSubarray.timeoutExpiredFlag_read
def read_failureRaisedFlag(self):
# PROTECTED REGION ID(CspSubElementSubarray.failureRaisedFlag_read) ENABLED START #
"""
Return the failureRaisedFlag attribute.
*failureRaisedFlag* attribute is of type DevBoolean.
Its value is True when a failure condition occurs during the execution of
a command.
The Sub-element AlarmHandler, as well the Csp Subarray can subscribe to this attribute
to monitor failures during command execution.
"""
return self._failure_raised
# PROTECTED REGION END # // CspSubElementSubarray.failureRaisedFlag_read
def read_failureMessage(self):
# PROTECTED REGION ID(CspSubElementSubarray.failureMessage_read) ENABLED START #
"""
Return the failureMessage attribute.
*failureMessage* attribute reports the failure message associated to each failed commad.
Example:
*server side*
self._failure_message = {'gotoidle': 'Device XXX can't execute the command when in SCANNING', ..}
image_attr = [['gotoidle', 'Device XXX can't execute the command when in SCANNING'],...]
*client side*
>>> failure_message = proxy.failureMessage
>>> failureMessage
(('gotoidle', 'Device XXX can't execute the command when in SCANNING'),...))
>>> failureMessage[0][0]
'gotoidle'
>>> failureMessage[0][1]
'Device XXX can't execute the command when in SCANNING'
- row_index addresses the command name
- col_index addresses the number (string) of completed tasks
"""
image_attr = [[cmd_name, message] for cmd_name, message in self._failure_message.items()]
return image_attr
# PROTECTED REGION END # // CspSubElementSubarray.failureMessage_read
def read_configureScanCmdProgress(self):
# PROTECTED REGION ID(CspSubElementSubarray.configureScanProgress_read) ENABLED START #
"""
Return the configureScanProgress attribute.
*configureScanProgress* attribute stores the execution percentage of the
*ConfigureScan* command.
The CSP Subarray device can subscribe to this attribute to monitor and report
the command execution progress.
The attribute assumes values in the range [0, 100].
"""
return self._cmd_progress['configurescan']
# PROTECTED REGION END # // CspSubElementSubarray.configureScanProgress_read
def read_configureScanDurationMeasured(self):
# PROTECTED REGION ID(CspSubElementSubarray.configureScanDurationMeasured_read) ENABLED START #
"""
Return the configureScanDurationMeasured attribute.
*configureScanDurationMeasured* attribute reports the effective execution time for the
*ConfigureScan* command.
"""
return self._cmd_duration_measured['configurescan']
# PROTECTED REGION END # // CspSubElementSubarray.configureScanDurationMeasured_read
def read_listOfDevCompletedTask(self):
# PROTECTED REGION ID(CspSubElementSubarray.listOfDevCompletedTask_read) ENABLED START #
"""Return the listOfDevCompletedTask attribute.
build the content of the TANGO attribute listOfDevCompletedTask
(type Image of DevString) as a nested list.
x element: the command name
y element: the list of devices that completed the task
Example:
*server side*
self._list_dev_completed_task = {'gotoidle': 'device, device2, device3', 'configurescan': 'device1, device2'}
image_attr = [['gotoidle',''], ['device, device2, deviceconfigurescan':''device1, device2']]*client side*
*client side*
>>> list_dev_completed_task = proxy.listOfDevCompletedTask
>>> list_dev_completed_task
(('gotoidle', 'Device XXX can't execute the command when in SCANNING'),...))
>>> list_dev_completed_task[0][0]
'gotoidle'
>>> list_dev_completed_task[0][1]
'device, device2, device3''
- row_index addresses the command name
- col_index addresses the list of devices that completed tasks
"""
image_attr = [[cmd_name, dev_list] for cmd_name, dev_list in self._list_dev_completed_task.items()]
return image_attr
# PROTECTED REGION END # // CspSubElementSubarray.listOfDevCompletedTask_read
# --------
# Commands
# --------
@command(
)
@DebugIt()
def Abort(self):
# PROTECTED REGION ID(CspSubElementSubarray.Abort) ENABLED START #
"""
Stop abruptly the execution of a command.
After execution, the subarray obsState moves to ABORTED.
:return: None
"""
pass
# PROTECTED REGION END # // CspSubElementSubarray.Abort
@command(
)
@DebugIt()
@IsSubarrayCommandAllowed()
@ObsStateCheck('endscan')
def EndScan(self):
# PROTECTED REGION ID(CspSubElementSubarray.EndScan) ENABLED START #
"""
Stop the execution of the current scan.
The command is processed when the subarray is in ON State and its obsState value
is SCANNING.
On success, the device obsState moves to READY.
The *EndScan* class method behavior is modified by two Python decorators:
*IsSubarrayCommandAllowed* and *ObsStateCheck*.
These decorators respectively verify if the subarray is the required State and observing
state.
:return: None
:raises: tango.DevFailed exception when the subarray is not in the proper State or
obsState value.
"""
pass
# PROTECTED REGION END # // CspSubElementSubarray.EndScan
@command(
dtype_in='DevVarStringArray',
)
@DebugIt()
@IsSubarrayCommandAllowed()
@ObsStateCheck('scan')
def Scan(self, argin):
# PROTECTED REGION ID(CspSubElementSubarray.Scan) ENABLED START #
"""
Start the execution of the scan.
The command can be issued only when the subarray State is ON and its obsState
is READY.
On success, the subarray obsState moves to SCANNING.
The *Scan* class method behavior is modified by two Python decorators:
*IsSubarrayCommandAllowed* and *ObsStateCheck*.
These decorators respectively verify if the subarray is the required State and observing
state.
:param argin: the scan id (integer number)
:type: DevVarStringArray
:return: None
:raises: tango.DevFailed exception when the subarray is not in the proper State or
obsState value.
"""
pass
# PROTECTED REGION END # // CspSubElementSubarray.Scan
@command(
dtype_in='DevString',
doc_in="A Json-encoded string with the scan configuration.",
)
@DebugIt()
@IsSubarrayCommandAllowed()
@ObsStateCheck('configscan')
def ConfigureScan(self, argin):
# PROTECTED REGION ID(CspSubElementSubarray.ConfigureScan) ENABLED START #
"""
Configure a complete scan for the subarray.
The command is processed when the subarray is in ON State and its obsState value
is IDLE or READY (re-configuration).
The subarray obsState moves to CONFIGURING during the configuration process and then to
READY when the command ends with success.
The *ConfigureScan* class method behavior is modified by two Python decorators:
*IsSubarrayCommandAllowed* and *ObsStateCheck*.
These decorators respectively verify if the subarray is the required State and observing
state.
:param argin: a Json-encoded string with the scan configuration
:type: DevString
:return: None
:raises: tango.DevFailed exception when the subarray is not in the proper State or
observing state value.
"""
pass
# PROTECTED REGION END # // CspSubElementSubarray.ConfigureScan
@command(
)
@DebugIt()
@IsSubarrayCommandAllowed()
@ObsStateCheck('gotoidle')
@SubarrayRejectCmd('ConfigureScan', 'Scan')
def GoToIdle(self):
# PROTECTED REGION ID(CspSubElementSubarray.GoToIdle) ENABLED START #
"""
Set the Subarray obsState to IDLE.
The *GoToIdle* class method behavior is modified by three Python decorators:
*IsSubarrayCommandAllowed* and *ObsStateCheck* decorators respectively verify
if the subarray is the required State and observing state.
*SubarrayRejectCmd* decorator checks if there is another command is already running.
In this the case, the current command is rejected throwing an exception.
:return: None
:raises: tango.DevFailed exception when the subarray:
- is not in the proper State or observing state value
- is running the *ConfigureScan* or *Scan* command
"""
pass
# PROTECTED REGION END # // CspSubElementSubarray.GoToIdle
# ----------
# Run server
# ----------
[docs]def main(args=None, **kwargs):
"""Main function of the CspSubElementSubarray module."""
# PROTECTED REGION ID(CspSubElementSubarray.main) ENABLED START #
return run((CspSubElementSubarray,), args=args, **kwargs)
# PROTECTED REGION END # // CspSubElementSubarray.main
if __name__ == '__main__':
main()