#
# 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.
"""
SKAController.
Controller device
"""
from __future__ import annotations
import logging
from collections.abc import Callable
from typing import Any, TypeVar, cast
from ska_control_model import ResultCode
from ..base import BaseComponentManager, BaseInterface, SKABaseDevice
from ..commands import DeviceInitCommand, FastCommand
from ..type_hints import DevVarLongStringArrayType
from .controller_interface import ControllerInterface
__all__ = ["ControllerComponentManager", "SKAController"]
[docs]
class ControllerComponentManager(BaseComponentManager):
"""A stub for an controller component manager."""
# TODO
ComponentManagerT = TypeVar("ComponentManagerT", bound=ControllerComponentManager)
[docs]
class SKAController(SKABaseDevice[ComponentManagerT], ControllerInterface):
"""
A generic controller device for SKA.
:class:`~ska_tango_base.controller.controller_device.SKAController` inherits from
the :class:`~ska_tango_base.base.base_device.SKABaseDevice` and
:class:`~ska_tango_base.controller.controller_interface.ControllerInterface`, and
expects a component manager to be provided by implementing the
:meth:`~ska_tango_base.base.base_device.SKABaseDevice.create_component_manager`
method.
"""
Reset: Callable[[BaseInterface], DevVarLongStringArrayType] | None
Standby: Callable[[BaseInterface], DevVarLongStringArrayType] | None
Off: Callable[[BaseInterface], DevVarLongStringArrayType] | None
On: Callable[[BaseInterface], DevVarLongStringArrayType] | None
[docs]
def init_command_objects(self: SKAController[ComponentManagerT]) -> None:
"""Set up the command objects."""
super().init_command_objects()
self._register_default_command_object(
"IsCapabilityAchievable",
self.IsCapabilityAchievableCommand(self, self.logger),
)
[docs]
class InitCommand(DeviceInitCommand):
"""A class for the SKAController's init_device() "command"."""
[docs]
def do(
self: SKAController.InitCommand,
*args: Any,
**kwargs: Any,
) -> tuple[ResultCode, str]:
"""
Stateless hook for device initialisation.
:param args: positional arguments to the command. This command does
not take any, so this should be empty.
:param kwargs: keyword arguments to the command. This command does
not take any, so this should be empty.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
message = "SKAController Init command completed OK"
self.logger.info(message)
self._completed()
return (ResultCode.OK, message)
[docs]
def create_component_manager(
self: SKAController[ComponentManagerT],
) -> ComponentManagerT:
"""
Create and return a component manager for this device.
:raises NotImplementedError: because it is not implemented.
"""
raise NotImplementedError(
"'create_component_manager' method must be implemented by "
f"'{self.__class__.__name__}'. "
"The parent 'SKAController' is an abstract base class."
)
# --------
# Commands
# --------
# TODO(WOM-725): Work out how this should be moved to the interface.
[docs]
class IsCapabilityAchievableCommand(FastCommand[bool]):
"""A class for the SKAController's IsCapabilityAchievable() command."""
[docs]
def __init__(
self: SKAController.IsCapabilityAchievableCommand,
device: SKAController[ComponentManagerT],
logger: logging.Logger | None = None,
):
"""
Initialise a new instance.
:param device: the device that this command acts upon.
:param logger: a logger for this command to log with.
"""
self._device = device
super().__init__(logger=logger)
[docs]
def do(
self: SKAController.IsCapabilityAchievableCommand,
*args: Any,
**kwargs: Any,
) -> bool:
"""
Stateless hook for device IsCapabilityAchievable() command.
:param args: positional arguments to the command. There is a single
positional argument: an array consisting of pairs of
* [nrInstances]: DevLong. Number of instances of the capability.
* [Capability types]: DevString. Type of capability.
:param kwargs: keyword arguments to the command. This command does
not take any, so this should be empty.
:return: Whether the capability is achievable
"""
argin = cast(tuple[list[int], list[str]], args[0])
return self._device.execute_IsCapabilityAchievable(argin)
[docs]
def execute_IsCapabilityAchievable(
self: SKAController[ComponentManagerT], argin: tuple[list[int], list[str]]
) -> bool:
"""
Check if provided capabilities can be achieved by the resource(s).
To modify behaviour for this command, modify the do() method of
the command class.
: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
"""
handler = cast(
Callable[[tuple[list[int], list[str]]], bool],
self._get_overridden_command("IsCapabilityAchievable"),
)
if handler is not None:
return handler(argin)
return super().execute_IsCapabilityAchievable(argin)
# ----------
# Run server
# ----------
def main(*args: str, **kwargs: str) -> int:
"""
Entry point for module.
:param args: positional arguments
:param kwargs: named arguments
:return: exit code
"""
def _create_component_manager(
self: SKAController[ControllerComponentManager],
) -> ControllerComponentManager:
return ControllerComponentManager(self.logger)
SKAController.create_component_manager = _create_component_manager # type: ignore
return cast(int, SKAController.run_server(args=args or None, **kwargs))
if __name__ == "__main__":
main()