#
# 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 subarray devices."""
from ska_control_model import ObsState, ObsStateModel
from tango import AttReqType, DebugIt, DevState
from tango.server import attribute, command, device_property
from .._interface_helpers import get_request_type
from ..faults import StateModelError
from ..long_running_commands import LRCMixin, LRCReqType
from ..long_running_commands.mixin import AbstractLRCMixin
from ..obs import ObsInterface
from ..software_bus import (
Signal,
attribute_from_signal,
)
from ..type_hints import DevVarLongStringArrayType, ReadAttrType
__all__ = ["SubarrayInterface", "AbstractSubarrayInterface"]
[docs]
class AbstractSubarrayInterface(AbstractLRCMixin, ObsInterface):
"""
Provides the Tango interface for an SKA subarray.
The long running command implementation is abstract and a task_executor
must be provided.
"""
# --------------
# ObsState model
# --------------
obs_state_model: ObsStateModel
"""The Observing State Machine for the subarray."""
def _init_state_model(self) -> None:
"""Set up the state model for the device."""
super()._init_state_model()
self.obs_state_model = ObsStateModel(
logger=self.logger, callback=self._update_obs_state
)
[docs]
def component_scanning(self) -> None:
"""Perform scanning action on the ``ObsStateModel``."""
self.obs_state_model.perform_action("component_scanning")
[docs]
def component_not_scanning(self) -> None:
"""Perform not scanning action on the ``ObsStateModel``."""
self.obs_state_model.perform_action("component_not_scanning")
[docs]
def component_resourced(self) -> None:
"""Perform resourced action on the ``ObsStateModel``."""
self.obs_state_model.perform_action("component_resourced")
[docs]
def component_unresourced(self) -> None:
"""Perform unresourced action on the ``ObsStateModel``."""
self.obs_state_model.perform_action("component_unresourced")
[docs]
def component_obsfault(self) -> None:
"""Perform fault action on the ``ObsStateModel``."""
self.obs_state_model.perform_action("component_obsfault")
# ----------
# Attributes
# ----------
_activation_time = Signal[float](stored=True)
"""Signal for the activation time of the device.
Write to this signal to report the time of activation to Tango clients.
:meta public:
"""
activationTime = attribute_from_signal(
_activation_time,
dtype="double",
unit="s",
standard_unit="s",
display_unit="s",
description="The time of activation in seconds since Unix epoch.",
)
"""
Activation time attribute of the device.
This should be set by subclasses of this interface by writing to
:py:attr:`_activation_time`.
"""
@attribute(
dtype=("str",),
max_dim_x=512,
description="The resources assigned to the device.",
)
def assignedResources(self) -> list[str]:
"""
Read the assigned resources of the device.
Subclasses of this interface should override :py:meth:`read_assignedResources`.
"""
return self.read_assignedResources()
[docs]
def read_assignedResources(self) -> list[str]:
"""
Read the resources assigned to the device.
Subclasses of this interface must provide this method.
"""
raise NotImplementedError(
"'read_assignedResources' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
@attribute(
dtype=("str",),
max_dim_x=10,
description="""The configured capabilities of the device.
For example, ["Correlators:512", "PssBeams:4", "PstBeams:4", "VlbiBeams:0"].
""",
)
def configuredCapabilities(self) -> list[str]:
"""
Read the configured capabilities of the device.
Subclasses of this interface should override
:py:meth:`read_configuredCapabilities`.
"""
return self.read_configuredCapabilities()
def __read_activationTime(self) -> ReadAttrType[float]:
"""Dispatch to read method to allow subclasses to override."""
return self.read_activationTime()
activationTime.read(__read_activationTime)
[docs]
def read_activationTime(self) -> ReadAttrType[float]:
"""
Read the activation time of the device.
Subclasses can override this to change the behaviour of the
:py:obj:`activationTime` attribute.
"""
return self.__class__.activationTime.do_read(self)
def __is_activationTime_allowed(self, attr: AttReqType) -> bool:
return self.is_activationTime_allowed(attr)
[docs]
def is_activationTime_allowed(self, attr: AttReqType) -> bool:
"""
Check if the activationTime can be read currently.
This can be overridden by subclasses to restrict when clients
can access the attribute.
"""
return True
activationTime.fisallowed = __is_activationTime_allowed
# -----------------
# Device Properties
# -----------------
CapabilityTypes = device_property(
dtype=("str",),
doc="Types of capability supported by this subarray.",
)
SubID = device_property(
dtype="str",
doc="Unique identifier for this subarray.",
)
# --------
# Commands
# --------
def __is_AssignResources_allowed(self) -> bool:
return self.is_AssignResources_allowed()
[docs]
def is_AssignResources_allowed(
self, request_type: LRCReqType | None = None
) -> bool:
"""
Return whether :py:meth:`!AssignResources()` may be executed or enqueued.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "AssignResources") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "AssignResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [ObsState.EMPTY, ObsState.IDLE]:
raise StateModelError(
"AssignResources command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_AssignResources(self, argin: str) -> DevVarLongStringArrayType:
"""
Execute the standard AssignResources command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- When this command is started, the commandedObsState should be
**EMPTY** or **IDLE**.
- While this command is running, the device should have obsState **RESOURCING**.
- Once this command has completed successfully, the device should have
obsState **IDLE**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
:param argin: the resources to be assigned
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_AssignResources' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
[docs]
def started_AssignResources(self) -> None:
"""AssignResources command started callback."""
self.obs_state_model.perform_action("assign_invoked")
[docs]
def completed_AssignResources(self) -> None:
"""AssignResources command completed callback."""
self.obs_state_model.perform_action("assign_completed")
@DebugIt()
@command(
dtype_in="DevString",
doc_in=(
"Assign resources to this subarray.\n\n"
":parameter: The resources to be assigned"
),
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_AssignResources_allowed",
)
def AssignResources(self, argin: str) -> DevVarLongStringArrayType:
"""
Assign resources to this subarray.
Subclasses should override the :py:attr:`execute_AssignResources` method
to change the behaviour of this command.
:param argin: the resources to be assigned
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_AssignResources(argin)
AssignResources.__check_is_long_running__ = "execute_AssignResources"
def __is_ReleaseResources_allowed(self) -> bool:
return self.is_ReleaseResources_allowed()
[docs]
def is_ReleaseResources_allowed(
self, request_type: LRCReqType | None = None
) -> bool:
"""
Return whether :py:meth:`!ReleaseResources()` may be executed or enqueued.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "ReleaseResources") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ReleaseResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [ObsState.EMPTY, ObsState.IDLE]:
raise StateModelError(
"ReleaseResources command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_ReleaseResources(self, argin: str) -> DevVarLongStringArrayType:
"""
Execute the standard ReleaseResources command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- When this command is started, the commandedObsState should be **IDLE**.
- While this command is running, the device should have obsState **RESOURCING**.
- Once this command has completed successfully, the device should have
obsState **IDLE**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
:param argin: the resources to be released
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_ReleaseResources' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
[docs]
def started_ReleaseResources(self) -> None:
"""ReleaseResources command started callback."""
self.obs_state_model.perform_action("release_invoked")
[docs]
def completed_ReleaseResources(self) -> None:
"""ReleaseResources command completed callback."""
self.obs_state_model.perform_action("release_completed")
@DebugIt()
@command(
dtype_in="DevString",
doc_in=(
"Delta release of assigned resources from this subarray.\n\n"
":parameter: The resources to be released"
),
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_ReleaseResources_allowed",
)
def ReleaseResources(self, argin: str) -> DevVarLongStringArrayType:
"""
Delta removal of assigned resources.
Subclasses should override the :py:attr:`execute_ReleaseResources` method
to change the behaviour of this command.
:param argin: the resources to be released
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_ReleaseResources(argin)
ReleaseResources.__check_is_long_running__ = "execute_ReleaseResources"
def __is_ReleaseAllResources_allowed(self) -> bool:
return self.is_ReleaseAllResources_allowed()
[docs]
def is_ReleaseAllResources_allowed(
self, request_type: LRCReqType | None = None
) -> bool:
"""
Return whether the :py:meth:`!ReleaseAllResources()` command may be called.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "ReleaseAllResources") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ReleaseResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [ObsState.EMPTY, ObsState.IDLE]:
raise StateModelError(
"ReleaseAllResources command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_ReleaseAllResources(self) -> DevVarLongStringArrayType:
"""
Execute the standard ReleaseAllResources command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- When this command is started, the commandedObsState should be **EMPTY**.
- While this command is running, the device should have obsState **RESOURCING**.
- Once this command has completed successfully, the device should have
obsState **EMPTY**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_ReleaseAllResources' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
[docs]
def started_ReleaseAllResources(self) -> None:
"""ReleaseAllResources command started callback."""
self.obs_state_model.perform_action("release_invoked")
[docs]
def completed_ReleaseAllResources(self) -> None:
"""ReleaseAllResources command completed callback."""
self.obs_state_model.perform_action("release_completed")
@DebugIt()
@command(
doc_in="Release all resources to tear down to an empty subarray.",
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_ReleaseAllResources_allowed",
)
def ReleaseAllResources(
self,
) -> DevVarLongStringArrayType:
"""
Remove all resources to tear down to an empty subarray.
Subclasses should override the :py:attr:`execute_ReleaseAllResources` method
to change the behaviour of this command.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_ReleaseAllResources()
ReleaseAllResources.__check_is_long_running__ = "execute_ReleaseAllResources"
def __is_Configure_allowed(self) -> bool:
return self.is_Configure_allowed()
@DebugIt()
@command(
dtype_in="DevString",
doc_in=(
"Configure the capabilities of this subarray.\n\n"
":parameter: The scan configuration specification"
),
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_Configure_allowed",
)
def Configure(self, argin: str) -> DevVarLongStringArrayType:
"""
Configure the capabilities of this subarray.
Subclasses should override the :py:attr:`execute_Configure` method
to change the behaviour of this command.
:param argin: the scan configuration specification
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_Configure(argin)
Configure.__check_is_long_running__ = "execute_Configure"
def __is_Scan_allowed(self) -> bool:
return self.is_Scan_allowed()
[docs]
def is_Scan_allowed(self, request_type: LRCReqType | None = None) -> bool:
"""
Return whether the :py:meth:`!Scan()` command may be called currently.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "Scan") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ReleaseResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state != ObsState.READY:
raise StateModelError(
"Scan command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_Scan(self, argin: str) -> DevVarLongStringArrayType:
"""
Execute the standard Scan command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- Once this command has completed successfully, the device should have
obsState **SCANNING**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
:param argin: the per-scan configuration
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_Scan' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
@DebugIt()
@command(
dtype_in="DevString",
doc_in=(
"Start scanning with this configured subarray.\n\n"
":parameter: The per-scan configuration"
),
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_Scan_allowed",
)
def Scan(self, argin: str) -> DevVarLongStringArrayType:
"""
Start scanning.
Subclasses should override the :py:attr:`execute_Scan` method
to change the behaviour of this command.
:param argin: the per-scan configuration
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_Scan(argin)
Scan.__check_is_long_running__ = "execute_Scan"
def __is_EndScan_allowed(self) -> bool:
return self.is_EndScan_allowed()
[docs]
def is_EndScan_allowed(self, request_type: LRCReqType | None = None) -> bool:
"""
Return whether the :py:meth:`!EndScan()` command may be called currently.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "EndScan") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ReleaseResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state != ObsState.SCANNING:
raise StateModelError(
"EndScan command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_EndScan(self) -> DevVarLongStringArrayType:
"""
Execute the standard EndScan command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- Once this command has completed successfully, the device should have
obsState **READY**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_EndScan' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
@DebugIt()
@command(
doc_in="End the current scan with this subarray.",
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_EndScan_allowed",
)
def EndScan(self) -> DevVarLongStringArrayType:
"""
End the scan.
Subclasses should override the :py:attr:`execute_EndScan` method
to change the behaviour of this command.
:raises NotImplementedError:
"""
return self.execute_EndScan()
EndScan.__check_is_long_running__ = "execute_EndScan"
def __is_End_allowed(self) -> bool:
return self.is_End_allowed()
[docs]
def is_End_allowed(self, request_type: LRCReqType | None = None) -> bool:
"""
Return whether the :py:meth:`!End()` command may be called currently.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "End") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ReleaseResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [ObsState.IDLE, ObsState.READY]:
raise StateModelError(
f"End command not permitted in observation state {self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_End(self) -> DevVarLongStringArrayType:
"""
Execute the standard End command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- When this command is started, the commandedObsState should be
**IDLE**.
- Once this command has completed successfully, the device should have
obsState **IDLE**.
- If this command does not complete successfully, the device should have
obsState **FAULT**
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_End' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
@DebugIt()
@command(
doc_in="End the scan block of this subarray.",
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_End_allowed",
)
def End(self) -> DevVarLongStringArrayType:
"""
End the scan block.
Subclasses should override the :py:attr:`execute_End` method
to change the behaviour of this command.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_End()
End.__check_is_long_running__ = "execute_End"
def __is_ObsReset_allowed(self) -> bool:
return self.is_ObsReset_allowed()
[docs]
def is_ObsReset_allowed(self, request_type: LRCReqType | None = None) -> bool:
"""
Return whether :py:meth:`!ObsReset()` may be executed or enqueued.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "ObsReset") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ObsReset not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [ObsState.FAULT, ObsState.ABORTED]:
raise StateModelError(
"ObsReset command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_ObsReset(self) -> DevVarLongStringArrayType:
"""
Execute the standard ObsReset command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- When this command is started, the commandedObsState should be **IDLE**.
- While this command is running, the device should have obsState
**RESETTING**.
- Once this command has completed successfully, the device should have
obsState **IDLE**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
.. note::
:py:meth:`!ObsReset()` is not to be implemented for AA 0.5 or AA 1.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_ObsReset' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
[docs]
def started_ObsReset(self) -> None:
"""ObsReset command started callback."""
self.obs_state_model.perform_action("obsreset_invoked")
[docs]
def completed_ObsReset(self) -> None:
"""ObsReset command completed callback."""
self.obs_state_model.perform_action("obsreset_completed")
@DebugIt()
@command(
doc_in="Reset the current observation process.",
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_ObsReset_allowed",
)
def ObsReset(self) -> DevVarLongStringArrayType:
"""
Reset the current observation process.
Subclasses should override the :py:attr:`execute_ObsReset` method
to change the behaviour of this command.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
return self.execute_ObsReset()
ObsReset.__check_is_long_running__ = "execute_ObsReset"
def __is_Restart_allowed(self) -> bool:
return self.is_Restart_allowed()
[docs]
def is_Restart_allowed(self, request_type: LRCReqType | None = None) -> bool:
"""
Return whether the :py:meth:`!Restart()` command may be called currently.
This method can be overridden by subclasses to change when this command
is allowed.
:param request_type: ENQUEUE_REQ when the LRC is enqueued by the Tango command
and EXECUTE_REQ when the LRC is about to be executed by the executor.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
if (
request_type is None
and get_request_type(self, "Restart") == LRCReqType.ENQUEUE_REQ
):
return True
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "ObsReset not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [ObsState.FAULT, ObsState.ABORTED]:
raise StateModelError(
"Restart command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def execute_Restart(self) -> DevVarLongStringArrayType:
"""
Execute the standard Restart command.
This method must be overridden by all subclasses. The
:py:func:`~ska_tango_base.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Reset a long running command.
- When this command is started, the commandedObsState should be **FAULT**
or **ABORTED**.
- While this command is running, the device should have obsState
**RESTARTING**.
- Once this command has completed successfully, the device should have
obsState **EMPTY**.
- If this command does not complete successfully, the device should have
obsState **FAULT**.
:return: A tuple containing a return code and a string
message indicating status or a command ID. The message
is for information purpose only.
"""
raise NotImplementedError(
"'execute_Restart' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SubarrayInterface' is an abstract base class."
)
[docs]
def started_Restart(self) -> None:
"""Restart command started callback."""
self.obs_state_model.perform_action("restart_invoked")
[docs]
def completed_Restart(self) -> None:
"""Restart command completed callback."""
self.obs_state_model.perform_action("restart_completed")
@DebugIt()
@command(
doc_in="Restart the subarray by deconfiguring and releasing all resources.",
dtype_out="DevVarLongStringArray",
doc_out="[ResultCode][Status message or command ID]",
fisallowed="_AbstractSubarrayInterface__is_Restart_allowed",
)
def Restart(self) -> DevVarLongStringArrayType:
"""
Restart the subarray. That is, deconfigure and release all resources.
Subclasses should override the :py:attr:`execute_Restart` method
to change the behaviour of this command.
:return: A tuple containing a result code and the unique ID of the command
"""
return self.execute_Restart()
Restart.__check_is_long_running__ = "execute_Restart"
[docs]
def is_Abort_allowed(self) -> bool:
"""
Return whether the :py:meth:`!Abort()` command may be called currently.
This method can be overridden by subclasses to change when this command
is allowed.
:raises StateModelError: command not permitted in observation state
:return: whether the command may be called in the current device
state
"""
# If we return False here, Tango will raise an exception that incorrectly blames
# refusal on device state.
# e.g. "Abort not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [
ObsState.RESOURCING,
ObsState.IDLE,
ObsState.CONFIGURING,
ObsState.READY,
ObsState.SCANNING,
ObsState.RESETTING,
]:
raise StateModelError(
"Abort command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() in [DevState.ON, DevState.ALARM]
[docs]
def started_Abort(self) -> None:
"""Abort command started callback."""
self.obs_state_model.perform_action("abort_invoked")
[docs]
def completed_Abort(self) -> None:
"""Abort command completed callback."""
self.obs_state_model.perform_action("abort_completed")
[docs]
class SubarrayInterface(LRCMixin, AbstractSubarrayInterface):
"""Provides the Tango interface for an SKA subarray."""