#
# 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.
"""
SKAObsDevice.
A generic base device for Observations for SKA. It inherits
SKABaseDevice class. Any device implementing an obsMode will inherit
from SKAObsDevice instead of just SKABaseDevice.
"""
from __future__ import annotations
from typing import Any, Callable, TypeVar, cast
from ska_control_model import ObsState, ResultCode
from ..base import BaseComponentManager, BaseInterface, SKABaseDevice
from ..commands import DeviceInitCommand
from ..type_hints import DevVarLongStringArrayType
from .obs_interface import ObsInterface
__all__ = ["ObsDeviceComponentManager", "SKAObsDevice", "main"]
[docs]
class ObsDeviceComponentManager(BaseComponentManager):
"""A stub for an observing device component manager."""
# TODO
ComponentManagerT = TypeVar("ComponentManagerT", bound=ObsDeviceComponentManager)
[docs]
class SKAObsDevice(SKABaseDevice[ComponentManagerT], ObsInterface):
"""
A generic base device for Observations for SKA.
:class:`~ska_tango_base.obs.obs_device.SKAObsDevice` inherits from
the :class:`~ska_tango_base.base.base_device.SKABaseDevice` and
:class:`~ska_tango_base.obs.obs_interface.ObsInterface`, 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]
class InitCommand(DeviceInitCommand):
"""A class for the SKAObsDevice's init_device() "command"."""
[docs]
def do(
self: SKAObsDevice.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 = "SKAObsDevice Init command completed OK"
self.logger.info(message)
self._completed()
return (ResultCode.OK, message)
[docs]
def create_component_manager(
self: SKAObsDevice[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 'SKAObsDevice' is an abstract base class."
)
# -----------------
# Device Properties
# -----------------
# ---------------
# General methods
# ---------------
def _update_commanded_obs_state(
self: SKAObsDevice[ComponentManagerT], commanded_obs_state: ObsState
) -> None:
self._commanded_obs_state = commanded_obs_state
# ----------
# Attributes
# ----------
# --------
# Commands
# --------
# ----------
# Run server
# ----------
[docs]
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: SKAObsDevice[ObsDeviceComponentManager],
) -> ObsDeviceComponentManager:
return ObsDeviceComponentManager(self.logger)
SKAObsDevice.create_component_manager = _create_component_manager # type: ignore
return cast(int, SKAObsDevice.run_server(args=args or None, **kwargs))
if __name__ == "__main__":
main()