# -*- coding: utf-8 -*-
#
# This file is part of the SKA Low MCCS project
#
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""This module implements the MCCS station beam device."""
from __future__ import annotations # allow forward references in type hints
import functools
import importlib.resources
import json
import logging
import sys
from typing import Any, Callable, Final, Optional
from ska_control_model import (
CommunicationStatus,
HealthState,
ObsStateModel,
PowerState,
ResultCode,
TaskStatus,
)
from ska_tango_base.base import CommandTracker
from ska_tango_base.commands import JsonValidator, SlowCommand, SubmittedSlowCommand
from ska_tango_base.obs import SKAObsDevice
from tango.server import attribute, command, device_property
from ska_low_mccs.station_beam.station_beam_component_manager import (
StationBeamComponentManager,
)
from ska_low_mccs.station_beam.station_beam_health_model import StationBeamHealthModel
__all__ = ["MccsStationBeam", "main"]
DevVarLongStringArrayType = tuple[list[ResultCode], list[Optional[str]]]
# pylint: disable=too-many-public-methods
[docs]class MccsStationBeam(SKAObsDevice):
"""An implementation of a station beam Tango device for MCCS."""
# -----------------
# Device Properties
# -----------------
BeamId = device_property(dtype=int, default_value=0)
# ---------------
# Initialisation
# ---------------
[docs] def __init__(self, *args: Any, **kwargs: Any) -> None:
"""
Initialise this device object.
:param args: positional args to the init
:param kwargs: keyword args to the init
"""
# We aren't supposed to define initialisation methods for Tango
# devices; we are only supposed to define an `init_device` method. But
# we insist on doing so here, just so that we can define some
# attributes, thereby stopping the linters from complaining about
# "attribute-defined-outside-init" etc. We still need to make sure that
# `init_device` re-initialises any values defined in here.
super().__init__(*args, **kwargs)
self.component_manager: StationBeamComponentManager
self._health_state: HealthState = HealthState.OK
self._health_model: StationBeamHealthModel
self.obs_state_model: ObsStateModel
[docs] def init_device(self: MccsStationBeam) -> None:
"""Initialise the device."""
super().init_device()
self._build_state = sys.modules["ska_low_mccs"].__version_info__
self._version_id = sys.modules["ska_low_mccs"].__version__
device_name = f'{str(self.__class__).rsplit(".", maxsplit=1)[-1][0:-2]}'
version = f"{device_name} Software Version: {self._version_id}"
properties = (
f"Initialised {device_name} device with properties:\n"
f"\tBeamId: {self.BeamId}\n"
)
self.logger.info(
"\n%s\n%s\n%s", str(self.GetVersionInfo()), version, properties
)
def _init_state_model(self: MccsStationBeam) -> None:
super()._init_state_model()
self._health_state = (
HealthState.OK
) # Beam starts empty which is OK. InitCommand.do() does this too late.
self._health_model = StationBeamHealthModel(
self._health_changed,
ignore_power_state=True,
thresholds={
"antenna_degraded_threshold": StationBeamHealthModel.DEGRADED_CRITERIA,
"antenna_failed_threshold": StationBeamHealthModel.FAILED_CRITERIA,
},
)
self.set_change_event("healthState", True, False)
self.obs_state_model = ObsStateModel(
logger=self.logger, callback=self._update_obs_state
)
[docs] def create_component_manager(
self: MccsStationBeam,
) -> StationBeamComponentManager:
"""
Create and return a component manager for this device.
:return: a component manager for this device.
"""
return StationBeamComponentManager(
self.BeamId,
self.logger,
self._communication_state_changed,
self._component_state_callback,
)
_schema_assign_resources: Final = json.loads(
importlib.resources.read_text(
"ska_low_mccs.schemas.station_beam",
"MccsStationBeam_AssignResources_3_0.json",
)
)
_schema_configure: Final = json.loads(
importlib.resources.read_text(
"ska_low_mccs.schemas.station_beam",
"MccsStationBeam_Configure_3_0.json",
)
)
_schema_scan: Final = json.loads(
importlib.resources.read_text(
"ska_low_mccs.schemas.common",
"Mccs_Scan_3_0.json",
)
)
[docs] def init_command_objects(self: MccsStationBeam) -> None:
"""Initialise the command handlers for commands supported by this device."""
super().init_command_objects()
def _callback(hook: str, running: bool) -> None:
action = "invoked" if running else "completed"
full_action = f"{hook}_{action}"
self.logger.debug(full_action)
self.obs_state_model.perform_action(full_action)
# commands with parameters defined by a Json schema specify a validator
# commands which affect the obsState specify a callback on entry/exit
for command_name, method_name, schema, state_model_hook in [
(
"AssignResources",
"assign_resources",
self._schema_assign_resources,
"assign",
),
("ReleaseAllResources", "release_all_resources", None, "release"),
("Configure", "configure", self._schema_configure, "configure"),
("Scan", "scan", self._schema_scan, None),
("EndScan", "end_scan", None, None),
("End", "end", None, None),
# ("Abort", "abort", None, "abort"),
("ObsReset", "obsreset", None, "obsreset"),
("Restart", "restart", None, "restart"),
("ApplyPointing", "apply_pointing", None, None),
]:
callback = (
None
if state_model_hook is None
else functools.partial(_callback, state_model_hook)
)
validator = (
None
if schema is None
else JsonValidator(
command_name,
schema,
logger=self.logger,
)
)
self.register_command_object(
command_name,
SubmittedSlowCommand(
command_name,
self._command_tracker,
self.component_manager,
method_name,
callback=callback,
logger=self.logger,
validator=validator,
),
)
self.register_command_object(
"Abort",
self.AbortCommand(
self._command_tracker,
self.component_manager,
callback=functools.partial(_callback, "abort"),
logger=self.logger,
),
)
[docs] class InitCommand(SKAObsDevice.InitCommand):
"""
A class for :py:class:`~.MccsStationBeam`'s Init command.
The :py:meth:`~.MccsStationBeam.InitCommand.do` method below is
called upon :py:class:`~.MccsStationBeam`'s initialisation.
"""
# pylint: disable-next=arguments-differ
[docs] def do( # type: ignore[override]
self: MccsStationBeam.InitCommand,
) -> tuple[ResultCode, str]:
"""
Initialise the attributes and properties of the MccsStationBeam.
:return: A tuple containing a return code and a string
message indicating status. The message is for
information purpose only.
"""
self._device._build_state = sys.modules["ska_low_mccs"].__version_info__
self._device._version_id = sys.modules["ska_low_mccs"].__version__
super().do()
return (ResultCode.OK, "Initialisation complete")
# ----------
# Callbacks
# ----------
# pylint: disable=too-many-arguments
def _component_state_callback(
self: MccsStationBeam,
fault: Optional[bool] = None,
power: Optional[PowerState] = None,
health: Optional[HealthState] = None,
trl: Optional[str] = None,
resources_changed: Optional[bool] = None,
configured_changed: Optional[bool] = None,
scanning_changed: Optional[bool] = None,
beam_locked: Optional[bool] = None,
) -> None:
"""
Handle change in this device's state.
This is a callback hook, called whenever the state changes. It
is responsible for updating the tango side of things i.e. making
sure the attribute is up to date, and events are pushed.
:param fault: An optional flag if the device is entering or
exiting a fault state.
:param power: An optional parameter with the new power state of the device.
:param health: An optional parameter with the new health state of the device.
:param trl: The TRL of the calling device or `None` if this
device is the caller.
:param resources_changed: An optional flag to update the resourced state
:param configured_changed: An optional flag to update the configuration state
:param scanning_changed: An optional flag indicating that this beam
scanning state has changed.
:param beam_locked: An optional flag with an update of beam locked state.
"""
if health is not None:
if trl is None:
# Do regular health update. This device called the callback.
if self._health_state != health:
self._health_state = health
self.push_change_event("healthState", health)
else:
# Call station health changed.
self._health_model.station_health_changed(HealthState(health))
# Probably more to do here with fault.
if fault is not None:
self._health_model.station_fault_changed(fault)
self.obs_state_model.perform_action("component_obsfault")
if beam_locked is not None:
self._health_model.is_beam_locked_changed(beam_locked)
if resources_changed is not None:
self._resources_changed(resources_changed)
if configured_changed is not None:
self._configuration_changed(configured_changed)
if scanning_changed is not None:
self._scanning_changed(scanning_changed)
def _communication_state_changed(
self: MccsStationBeam,
communication_state: CommunicationStatus,
) -> None:
"""
Handle change in communications status between component manager and component.
This is a callback hook, called by the component manager when
the communications status changes. It is implemented here to
drive the op_state.
:param communication_state: the status of communications between
the component manager and its component.
"""
action_map = {
CommunicationStatus.DISABLED: "component_disconnected",
CommunicationStatus.NOT_ESTABLISHED: "component_unknown",
CommunicationStatus.ESTABLISHED: "component_on", # it's an always-on device
}
self.op_state_model.perform_action(action_map[communication_state])
self._health_model.update_state(
communicating=communication_state == CommunicationStatus.ESTABLISHED
)
def _configuration_changed(self: MccsStationBeam, is_configured: Any) -> None:
"""
Handle the configuration change.
:param is_configured: if the configuration has changed
"""
if is_configured:
self.obs_state_model.perform_action("component_configured")
else:
self.obs_state_model.perform_action("component_unconfigured")
def _resources_changed(self: MccsStationBeam, is_resourced: Any) -> None:
"""
Handle change in station beam resources.
This is a callback hook, called by the component manager when
the resources of the station beam changes.
:param is_resourced: if the component has been resourced
"""
if is_resourced:
self.logger.debug("component_resourced")
self.obs_state_model.perform_action("component_resourced")
else:
self.logger.debug("component_unresourced")
self.obs_state_model.perform_action("component_unresourced")
def _scanning_changed(self: MccsStationBeam, is_scanning: bool) -> None:
"""
Handle change in station beam scanning status.
This is a callback hook, called by the component manager when
the scanning status of the station beam changes.
:param is_scanning: if the component is performing a scan
"""
if is_scanning:
self.logger.debug("component_scanning")
self.obs_state_model.perform_action("component_scanning")
else:
self.logger.debug("component_not_scanning")
self.obs_state_model.perform_action("component_not_scanning")
def _health_changed(self: MccsStationBeam, health: HealthState) -> None:
"""
Handle change in this device's health state.
This is a callback hook, called whenever the HealthModel's
evaluated health state changes. It is responsible for updating
the tango side of things i.e. making sure the attribute is up to
date, and events are pushed.
:param health: the new health value
"""
if self._health_state != health:
self._health_state = health
self.push_change_event("healthState", health)
# ----------
# Attributes
# ----------
@attribute(
dtype="DevString",
format="%s",
)
def healthModelParams(self: MccsStationBeam) -> str:
"""
Get the health params from the health model.
:return: the health params
"""
return json.dumps(self._health_model.health_params)
[docs] @healthModelParams.write # type: ignore[no-redef]
def healthModelParams(self: MccsStationBeam, argin: str) -> None:
"""
Set the params for health transition rules.
:param argin: JSON-string of dictionary of health states
"""
self._health_model.health_params = json.loads(argin)
self._health_model.update_health()
[docs] @attribute(dtype="DevLong", format="%i")
def subarrayId(self: MccsStationBeam) -> int:
"""
Return the subarray id.
:return: the subarray id
"""
return self.component_manager.subarray_id
[docs] @attribute(dtype="DevLong", format="%i", max_value=47, min_value=0)
def beamId(self: MccsStationBeam) -> int:
"""
Return the station beam id.
:return: the station beam id
"""
return self.component_manager.beam_id
[docs] @attribute(dtype=str)
def stationTrl(self: MccsStationBeam) -> str:
"""
Return the station TRL.
:return: the station TRL
"""
return self.component_manager.station_trl
[docs] @attribute(dtype="DevLong")
def stationId(self: MccsStationBeam) -> int:
"""
Return the station id.
:return: the station id
"""
return self.component_manager.station_id
[docs] @attribute(dtype="DevLong", format="%i", max_value=7, min_value=0)
def logicalBeamId(self: MccsStationBeam) -> int:
"""
Return the logical beam id.
:todo: this documentation needs to differentiate logical beam id
from beam id
:return: the logical beam id
"""
return self.component_manager.logical_beam_id
[docs] @attribute(
dtype="DevDouble",
unit="s",
max_value=1e37,
min_value=0,
)
def updateRate(self: MccsStationBeam) -> float:
"""
Return the update rate (in seconds) for this station beam.
:return: the update rate for this station beam
"""
return self.component_manager.update_rate
[docs] @attribute(dtype="DevBoolean")
def isBeamLocked(self: MccsStationBeam) -> bool:
"""
Return a flag indicating whether the beam is locked or not.
:return: whether the beam is locked or not
"""
return self.component_manager.is_beam_locked
[docs] @attribute(dtype=(("DevLong",),), max_dim_y=48, max_dim_x=8)
def channels(self: MccsStationBeam) -> list[list[int]]:
"""
Return the ids of the channels configured for this beam.
:return: channel ids
"""
return self.component_manager.channels
[docs] @attribute(dtype=("DevFloat",), max_dim_x=384)
def antennaWeights(self: MccsStationBeam) -> list[float]:
"""
Return the antenna weights configured for this beam.
:return: antenna weightd
"""
return self.component_manager.antenna_weights
[docs] @attribute(dtype=("DevDouble",), max_dim_x=5)
def desiredPointing(self: MccsStationBeam) -> list[float]:
"""
Return the desired pointing of this beam.
:return: the desired point of this beam, conforming to the Sky
Coordinate Set definition
"""
return self.component_manager.desired_pointing
[docs] @attribute(dtype=("DevDouble",), max_dim_x=384)
def pointingDelay(self: MccsStationBeam) -> list[float]:
"""
Return the pointing delay of this beam.
:return: the pointing delay of this beam
"""
return self.component_manager.pointing_delay
[docs] @attribute(dtype=("DevDouble",), max_dim_x=384)
def pointingDelayRate(self: MccsStationBeam) -> list[float]:
"""
Return the pointing delay rate of this beam.
:return: the pointing delay rate of this beam
"""
return self.component_manager.pointing_delay_rate
[docs] @attribute(dtype=("DevDouble",), max_dim_x=5)
def phaseCentre(self: MccsStationBeam) -> list[float]:
"""
Return the phase centre.
:return: the phase centre
"""
return self.component_manager.phase_centre
[docs] @attribute(dtype="DevString")
def apertureId(self: MccsStationBeam) -> str:
"""
Return the Aperture Id.
:return: the station beam Aperture Id, in the form APx.y y=substation
"""
return self.component_manager.aperture_id
[docs] @attribute(dtype="DevString")
def pointingReferenceFrame(self: MccsStationBeam) -> str:
"""
Return the Reference Frame ID for the pointing coordinates.
:return: the Reference Frame ID for the pointing coordinates
"""
return self.component_manager._pointing_reference_frame
[docs] @attribute(dtype="DevString")
def pointingTimestamp(self: MccsStationBeam) -> str:
"""
Return the timestamp used for the pointing coordinates.
:return: the timestamp used for the pointing coordinates.
"""
return self.component_manager._pointing_timestamp
[docs] @attribute(dtype="DevLong")
def hardwareBeamId(self: MccsStationBeam) -> int:
"""
Return the timestamp used for the pointing coordinates.
:return: the timestamp used for the pointing coordinates.
"""
return self.component_manager._hardware_beam_id
[docs] @attribute(dtype="DevLong")
def scanId(self: MccsStationBeam) -> int:
"""
Return the scan ID for the current scan.
:return: the scan ID or 0 if not scanning
"""
return self.component_manager._scan_id
[docs] @attribute(dtype="DevString")
def healthReport(self: MccsStationBeam) -> str:
"""
Get the health report.
:return: the health report.
"""
return self._health_model.health_report
# --------
# Commands
# --------
[docs] @command(dtype_in="DevString", dtype_out="DevVarLongStringArray")
def AssignResources(self: MccsStationBeam, argin: str) -> DevVarLongStringArrayType:
"""
Assign resources to the station beam with all relevant parameters.
:param argin: Assign Resources parameters encoded in a json string
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("AssignResources")
(result_code, unique_id) = handler(argin)
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def ApplyPointing(self: MccsStationBeam) -> DevVarLongStringArrayType:
"""
Apply pointing delays to antennas associated with the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("ApplyPointing")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_in="DevString", dtype_out="DevVarLongStringArray")
def Scan(self: MccsStationBeam, argin: str) -> DevVarLongStringArrayType:
"""
Start the scan associated with the station_beam.
:param argin: Configuration parameters encoded in a json string
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("Scan")
(result_code, unique_id) = handler(argin)
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def EndScan(self: MccsStationBeam) -> DevVarLongStringArrayType:
"""
Stop the current scan associated with the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("EndScan")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] class AbortCommand(SlowCommand[tuple[ResultCode, str]]):
"""A class for MccsSubarrayBeam's Abort() command."""
[docs] def __init__(
self: MccsStationBeam.AbortCommand,
command_tracker: CommandTracker,
component_manager: StationBeamComponentManager,
callback: Callable[[bool], None],
logger: logging.Logger | None = None,
) -> None:
"""
Initialise a new AbortCommand instance.
:param command_tracker: the device's command tracker
:param component_manager: the device's component manager
:param callback: callback to be called when this command
states and finishes
:param logger: a logger for this command object to yuse
"""
self._command_tracker = command_tracker
self._component_manager = component_manager
super().__init__(callback=callback, logger=logger)
[docs] def do(
self: MccsStationBeam.AbortCommand,
*args: Any,
**kwargs: Any,
) -> tuple[ResultCode, str]:
"""
Stateless hook for Abort() command functionality.
: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.
"""
command_id = self._command_tracker.new_command(
"Abort", completed_callback=self._completed
)
status, _ = self._component_manager.abort(
functools.partial(self._command_tracker.update_command_info, command_id)
)
# Abort() doesn't get queued, we skip the queue and start immediately in a
# separate thread.
assert status == TaskStatus.IN_PROGRESS
return ResultCode.STARTED, command_id
[docs] @command(dtype_out="DevVarLongStringArray")
def Abort(self: MccsStationBeam) -> DevVarLongStringArrayType:
"""
Abort the current scan associated with the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("Abort")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def ObsReset(self: MccsStationBeam) -> DevVarLongStringArrayType:
"""
Reset to IDLE the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("ObsReset")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def Restart(self: MccsStationBeam) -> DevVarLongStringArrayType:
"""
Restart to EMPTY the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("Restart")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def End(self: MccsStationBeam) -> DevVarLongStringArrayType:
"""
Deconfigure the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("End")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def ReleaseAllResources(
self: MccsStationBeam,
) -> DevVarLongStringArrayType:
"""
Release all allocated resources from the station_beam.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
handler = self.get_command_object("ReleaseAllResources")
(result_code, unique_id) = handler()
return ([result_code], [unique_id])
[docs] @command(dtype_out="DevVarLongStringArray")
def ToFault(
self: MccsStationBeam,
) -> DevVarLongStringArrayType:
"""
Put this station beam in ObsState.FAULT.
This is for use in testing only.
:return: A tuple containing a return code and a string message
indicating status. The message is for information
purpose only.
"""
self.obs_state_model._straight_to_state("FAULT")
return ([ResultCode.OK], ["Device sent to FAULT state."])
# ----------
# Run server
# ----------
[docs]def main(*args: str, **kwargs: str) -> int: # pragma: no cover
"""
Entry point for module.
:param args: positional arguments
:param kwargs: named arguments
:return: exit code
"""
return MccsStationBeam.run_server(args=args or None, **kwargs)
if __name__ == "__main__":
main()