#
# This file is part of the SKA Tango Base project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
"""Interfaces for SKA controller devices."""
import tango
import tango.server
from ..software_bus import (
Signal,
attribute_from_signal,
)
from ..type_hints import ReadAttrType
from ..utils import (
convert_dict_to_list,
validate_capability_types,
validate_input_sizes,
)
from ._base_interface import BaseInterface
__all__ = ["ControllerInterface"]
[docs]
class ControllerInterface(BaseInterface):
"""Provides the Tango interface for an SKA controller devices."""
# -----------------
# Device Properties
# -----------------
# List of maximum number of instances per capability type provided by this Element;
# CORRELATOR=512, PSS-BEAMS=4, PST-BEAMS=6, VLBI-BEAMS=4 or for DSH it can be:
# BAND-1=1, BAND-2=1, BAND3=0, BAND-4=0, BAND-5=0 (if only bands 1&2 is
# installed).
MaxCapabilities: list[str] = tango.server.device_property(
dtype=("str",),
doc=(
"List of maximum number of instances per capability type provided by this "
"element."
),
)
# ----------
# Attributes
# ----------
_max_capabilities = Signal[dict[str, int]](stored=True)
"""Signal for the maximum capabilities of the device.
Write to this signal to signal a change in the maximum capabilities of
the device.
:meta public:
"""
maxCapabilities = attribute_from_signal(
_max_capabilities,
dtype=("str",),
to_tango=convert_dict_to_list,
max_dim_x=20,
doc=(
"Maximum number of instances of each capability type,"
" e.g. 'CORRELATOR:512', 'PSS-BEAMS:4'."
),
)
"""
The maximum capabilities of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_max_capabilities`.
"""
def __read_maxCapabilities(self) -> ReadAttrType[list[str]]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_maxCapabilities()
maxCapabilities.read(__read_maxCapabilities)
[docs]
def read_maxCapabilities(self) -> ReadAttrType[list[str]]:
"""
Read the max capabilities of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`maxCapabilities` attribute.
"""
return self.__class__.maxCapabilities.do_read(self)
def __is_maxCapabilities_allowed(self, attr: tango.AttReqType) -> bool:
return self.is_maxCapabilities_allowed(attr)
[docs]
def is_maxCapabilities_allowed(self, attr: tango.AttReqType) -> bool:
"""
Check if the maxCapabilities can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
maxCapabilities.fisallowed = __is_maxCapabilities_allowed
_available_capabilities = Signal[dict[str, int]](stored=True)
"""Signal for the available capabilities of the device.
Write to this signal to report a change in the available capabilities of
the device.
:meta public:
"""
availableCapabilities = attribute_from_signal(
_available_capabilities,
dtype=("str",),
to_tango=convert_dict_to_list,
max_dim_x=20,
doc=(
"A list of available number of instances of each capability type, "
"e.g. 'CORRELATOR:512', 'PSS-BEAMS:4'."
),
)
"""
The available capabilities of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_available_capabilities`.
"""
def __read_availableCapabilities(self) -> ReadAttrType[list[str]]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_availableCapabilities()
availableCapabilities.read(__read_availableCapabilities)
[docs]
def read_availableCapabilities(self) -> ReadAttrType[list[str]]:
"""
Read the available capabilities of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`availableCapabilities` attribute.
"""
return self.__class__.availableCapabilities.do_read(self)
def __is_availableCapabilities_allowed(self, attr: tango.AttReqType) -> bool:
return self.is_availableCapabilities_allowed(attr)
[docs]
def is_availableCapabilities_allowed(self, attr: tango.AttReqType) -> bool:
"""
Check if the availableCapabilities can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
availableCapabilities.fisallowed = __is_availableCapabilities_allowed
# ---------
# Lifecycle
# ---------
[docs]
def on_new_shared_bus(self) -> None:
"""Initialise the device."""
super().on_new_shared_bus()
capabilities: dict[str, int] = {}
if self.MaxCapabilities:
for max_capability in self.MaxCapabilities:
(
capability_type,
max_capability_instances,
) = max_capability.split(":")
capabilities[capability_type] = int(max_capability_instances)
self._max_capabilities = capabilities
self._available_capabilities = capabilities.copy()
[docs]
@tango.server.command(
dtype_in="DevVarLongStringArray",
doc_in=(
"Check if provided capabilities can be achieved by the resource(s).\n\n"
":parameter: [No. of instances][Capability types]"
),
dtype_out=bool,
doc_out="[ResultCode][Command ID]",
)
def IsCapabilityAchievable(self, argin: tuple[list[int], list[str]]) -> bool:
"""
Check if provided capabilities can be achieved by the resource(s).
Subclasses should override :py:meth:`execute_IsCapabilityAchievable()`
to change the behaviour of this command.
:param argin: An array consisting pair of
* [nrInstances]: DevLong. Number of instances of the capability.
* [Capability types]: DevString. Type of capability.
:return: True or False
"""
return self.execute_IsCapabilityAchievable(argin)
[docs]
def execute_IsCapabilityAchievable(
self, argin: tuple[list[int], list[str]]
) -> bool:
"""
Execute the IsCapabilityAchievable command.
Subclasses should override this to change the behaviour of the
IsCapabilityAchievable command.
"""
command_name = "IsCapabilityAchievable"
capabilities_instances, capability_types = argin
validate_input_sizes(command_name, argin)
validate_capability_types(
command_name,
capability_types,
list(self._max_capabilities.keys()),
)
for capability_type, capability_instances in zip(
capability_types, capabilities_instances
):
if (
not self._available_capabilities[capability_type]
>= capability_instances
):
return False
return True