#
# 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"]
# pylint: disable-next=abstract-method
[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."""
Reset: Callable[[BaseInterface], DevVarLongStringArrayType] | None # type: ignore[assignment]
Standby: Callable[[BaseInterface], DevVarLongStringArrayType] | None # type: ignore[assignment]
Off: Callable[[BaseInterface], DevVarLongStringArrayType] | None # type: ignore[assignment]
On: Callable[[BaseInterface], DevVarLongStringArrayType] | None # type: ignore[assignment]
[docs]
class InitCommand(DeviceInitCommand):
# pylint: disable=protected-access # command classes are friend classes
"""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_obs_state(
self: SKAObsDevice[ComponentManagerT], obs_state: ObsState
) -> None:
"""
Perform Tango operations in response to a change in obsState.
This helper method is passed to the observation state model as a
callback, so that the model can trigger actions in the Tango
device.
:param obs_state: the new obs_state value
"""
self._obs_state = obs_state
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
"""
return cast(int, SKAObsDevice.run_server(args=args or None, **kwargs))
if __name__ == "__main__":
main()