#
# 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."""
import ska_control_model as scm
import tango
import tango.server
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 ..software_bus import Signal, attribute_from_signal
from ..type_hints import DevVarLongStringArrayType, ReadAttrType
from ._obs_interface import ObsInterface
__all__ = ["SubarrayInterface", "AbstractSubarrayInterface"]
[docs]
class AbstractSubarrayInterface(AbstractLRCMixin, ObsInterface):
"""
Provides the Tango interface for an SKA subarray.
This abstract class combines :py:class:`~.future._obs_interface.ObsInterface` and
:py:class:`~.long_running_commands.mixin.AbstractLRCMixin` that requires a concrete
:py:class:`~.type_hints.TaskExecutorProtocol` to be provided via ``task_executor``.
In most cases you should use :py:class:`SubarrayInterface` instead, which bundles
this class with :py:class:`~.LRCMixin` and a
:py:class:`~.TaskExecutorComponentManager`-backed task executor. Use this abstract
class only when you need to provide your own ``task_executor`` implementation (e.g.
via :py:class:`~.ComponentManagerLRCMixin`).
"""
# -----------------
# Device Properties
# -----------------
CapabilityTypes = tango.server.device_property(
dtype=("str",),
doc="Types of capability supported by this subarray.",
)
SubID = tango.server.device_property(
dtype="str",
doc="Unique identifier for this subarray.",
)
# ----------
# 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`.
"""
@tango.server.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."
)
@tango.server.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: tango.AttReqType) -> bool:
return self.is_activationTime_allowed(attr)
[docs]
def is_activationTime_allowed(self, attr: tango.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
# --------
# 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 [scm.ObsState.EMPTY, scm.ObsState.IDLE]:
raise StateModelError(
"AssignResources command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_AssignResources(self, argin: str) -> DevVarLongStringArrayType:
"""
Execute the standard AssignResources command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make AssignResources a long running command.
- Before calling this command, the device must be in obsState **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._update_obs_state(command_in_progress="AssignResources")
[docs]
def completed_AssignResources(self) -> None:
"""AssignResources command completed callback."""
self._update_obs_state(command_in_progress=None)
[docs]
@tango.DebugIt()
@tango.server.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 [scm.ObsState.EMPTY, scm.ObsState.IDLE]:
raise StateModelError(
"ReleaseResources command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_ReleaseResources(self, argin: str) -> DevVarLongStringArrayType:
"""
Execute the standard ReleaseResources command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make ReleaseResources a long running command.
- Before calling this command, the device must be in obsState **IDLE**.
- While this command is running, the device should have obsState **RESOURCING**.
- Once this command has completed successfully, the device should have
obsState **IDLE** or **EMPTY** depending on remaining resources.
- 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._update_obs_state(command_in_progress="ReleaseResources")
[docs]
def completed_ReleaseResources(self) -> None:
"""ReleaseResources command completed callback."""
self._update_obs_state(command_in_progress=None)
[docs]
@tango.DebugIt()
@tango.server.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. "ReleaseAllResources not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [scm.ObsState.EMPTY, scm.ObsState.IDLE]:
raise StateModelError(
"ReleaseAllResources command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_ReleaseAllResources(self) -> DevVarLongStringArrayType:
"""
Execute the standard ReleaseAllResources command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make ReleaseAllResources a long running command.
- Before calling this command, the device must be in obsState **IDLE**.
- 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._update_obs_state(command_in_progress="ReleaseAllResources")
[docs]
def completed_ReleaseAllResources(self) -> None:
"""ReleaseAllResources command completed callback."""
self._update_obs_state(command_in_progress=None)
[docs]
@tango.DebugIt()
@tango.server.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()
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. "Scan not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state != scm.ObsState.READY:
raise StateModelError(
"Scan command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_Scan(self, argin: str) -> DevVarLongStringArrayType:
"""
Execute the standard Scan command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Scan a long running command.
- Before calling this command, the device must be in obsState **READY**.
- 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."
)
[docs]
@tango.DebugIt()
@tango.server.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. "EndScan not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state != scm.ObsState.SCANNING:
raise StateModelError(
"EndScan command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_EndScan(self) -> DevVarLongStringArrayType:
"""
Execute the standard EndScan command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make EndScan a long running command.
- Before calling this command, the device must be in obsState **SCANNING**.
- 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."
)
[docs]
@tango.DebugIt()
@tango.server.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.
: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_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. "End not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [scm.ObsState.IDLE, scm.ObsState.READY]:
raise StateModelError(
f"End command not permitted in observation state {self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_End(self) -> DevVarLongStringArrayType:
"""
Execute the standard End command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make End a long running command.
- Before calling this command, the device must be in obsState **READY**.
- 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."
)
[docs]
@tango.DebugIt()
@tango.server.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_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. "Restart not allowed when the device is in ON state".
# So let's raise an exception ourselves.
if self._obs_state not in [scm.ObsState.FAULT, scm.ObsState.ABORTED]:
raise StateModelError(
"Restart command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def execute_Restart(self) -> DevVarLongStringArrayType:
"""
Execute the standard Restart command.
This method must be overridden by all subclasses. The
:py:func:`~.long_running_commands.decorators.submit_lrc_task`
decorator can be used to make Restart a long running command.
- Before calling this command, the device must be in obsState **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._update_obs_state(
command_in_progress="Restart", obsfault=False, aborted=False
)
[docs]
def completed_Restart(self) -> None:
"""Restart command completed callback."""
self._update_obs_state(
command_in_progress=None,
scanning=False,
resourced=False,
configured=False,
)
[docs]
@tango.DebugIt()
@tango.server.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 [
scm.ObsState.RESOURCING,
scm.ObsState.IDLE,
scm.ObsState.CONFIGURING,
scm.ObsState.READY,
scm.ObsState.SCANNING,
scm.ObsState.RESETTING,
]:
raise StateModelError(
"Abort command not permitted in observation state "
f"{self._obs_state.name}"
)
return self.get_state() is tango.DevState.ON
[docs]
def started_Abort(self) -> None:
"""Abort command started callback."""
self._update_obs_state(command_in_progress="Abort")
self._commanded_obs_state = scm.ObsState.ABORTED
[docs]
def completed_Abort(self) -> None:
"""Abort command completed callback."""
self._update_obs_state(command_in_progress=None, aborted=True)
[docs]
class SubarrayInterface(LRCMixin, AbstractSubarrayInterface):
"""
Provides the Tango interface for an SKA subarray device.
This class extends :py:class:`~.future._obs_interface.ObsInterface`
with the standard subarray command set and long running command (LRC) support
via :py:class:`~.long_running_commands.mixin.LRCMixin`, which supplies a
:py:class:`~.executor.TaskExecutor`-backed ``task_executor`` and the associated
LRC attributes.
Subclasses must override the ``execute_*`` methods for each command and apply
the :py:func:`~.long_running_commands.decorators.submit_lrc_task` decorator to make
them long running commands.
Observation state is driven by writing to the signals inherited from
:py:class:`~.future._obs_interface.ObsInterface`, primarily via
:py:meth:`~.ObsStateEmitMixin._update_obs_state`.
The ``started_*`` and ``completed_*`` callbacks on
:py:class:`AbstractSubarrayInterface` manage the ``command_in_progress`` portion of
the observation state automatically when using
:py:func:`~.long_running_commands.decorators.submit_lrc_task`.
Use :py:class:`AbstractSubarrayInterface` directly only when you need to supply your
own ``task_executor`` implementation (e.g. via
:py:class:`~.future._component_manager_mixins.ComponentManagerLRCMixin`).
"""