Source code for ska_tango_base.subarray.subarray_component_manager

#
# 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.
"""This module provides an abstract component manager for SKA Tango subarray devices."""

from __future__ import annotations

from typing import Any

from ska_control_model import TaskStatus

from ..obs import ObsDeviceComponentManager
from ..type_hints import TaskCallbackType


[docs] class SubarrayComponentManager(ObsDeviceComponentManager): """ An abstract base class for a component manager for an SKA subarray Tango devices. It supports: * Maintaining a connection to its component * Controlling its component via commands like AssignResources(), Configure(), Scan(), etc. * Monitoring its component, e.g. detect that a scan has completed """
[docs] def __init__( self: SubarrayComponentManager, *args: Any, **kwargs: Any, ) -> None: """ Initialise a new ComponentManager instance. :param args: additional positional arguments :param kwargs: additional keyword arguments """ super().__init__(*args, **kwargs)
[docs] def assign( self: SubarrayComponentManager, task_callback: TaskCallbackType | None, **kwargs: Any, ) -> tuple[TaskStatus, str]: """ Assign resources to the component. :param task_callback: callback to be called when the status of the command changes :param kwargs: keyword arguments. These will be the root keys defined by the command schema. :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'assign' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def release( self: SubarrayComponentManager, task_callback: TaskCallbackType | None, **kwargs: Any, ) -> tuple[TaskStatus, str]: """ Release resources from the component. :param task_callback: callback to be called when the status of the command changes :param kwargs: keyword arguments. These will be the root keys defined by the command schema. :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'release' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def release_all( self: SubarrayComponentManager, task_callback: TaskCallbackType | None ) -> tuple[TaskStatus, str]: """ Release all resources. :param task_callback: callback to be called when the status of the command changes :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'release_all' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def configure( self: SubarrayComponentManager, task_callback: TaskCallbackType | None, **kwargs: Any, ) -> tuple[TaskStatus, str]: """ Configure the component. :param task_callback: callback to be called when the status of the command changes :param kwargs: keyword arguments. These will be the root keys defined by the command schema. :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'configure' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def deconfigure( self: SubarrayComponentManager, task_callback: TaskCallbackType | None ) -> tuple[TaskStatus, str]: """ Deconfigure this component. :param task_callback: callback to be called when the status of the command changes :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'deconfigure' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def scan( self: SubarrayComponentManager, task_callback: TaskCallbackType | None, **kwargs: Any, ) -> tuple[TaskStatus, str]: """ Start scanning. :param task_callback: callback to be called when the status of the command changes :param kwargs: keyword arguments. These will be the root keys defined by the command schema. :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'scan' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def end_scan( self: SubarrayComponentManager, task_callback: TaskCallbackType | None ) -> tuple[TaskStatus, str]: """ End scanning. :param task_callback: callback to be called when the status of the command changes :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'end_scan' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def obsreset( self: SubarrayComponentManager, task_callback: TaskCallbackType | None ) -> tuple[TaskStatus, str]: """ Reset the component to unconfigured but do not release resources. :param task_callback: callback to be called when the status of the command changes :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'obsreset' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
[docs] def restart( self: SubarrayComponentManager, task_callback: TaskCallbackType | None ) -> tuple[TaskStatus, str]: """ Deconfigure and release all resources. :param task_callback: callback to be called when the status of the command changes :raises NotImplementedError: This is an abstract class """ raise NotImplementedError( f"'restart' method must be implemented by '{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )
@property def assigned_resources(self: SubarrayComponentManager) -> list[str]: """ Return the resources assigned to the component. :raises NotImplementedError: the resources assigned to the component """ raise NotImplementedError( "'assigned_resources' property must be implemented by " f"'{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." ) @property def configured_capabilities(self: SubarrayComponentManager) -> list[str]: """ Return the configured capabilities of the component. :raises NotImplementedError: list of strings indicating number of configured instances of each capability type """ raise NotImplementedError( "'configured_capabilities' property must be implemented by " f"'{self.__class__.__name__}'. " "The parent 'SubarrayComponentManager' is an abstract base class." )