# -*- coding: utf-8 -*-
#
# This file is part of the SKA PST project.
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE for more info.
"""This module provides an implementation of the DSP.FT PST component manager."""
from __future__ import annotations
import logging
from typing import Any
from overrides import override
from ska_pst.lmc.component import PstProcessApiSubcomponentManager
from ska_pst.lmc.dsp.dsp_ft_process_api import (
PstDspFlowThroughProcessApi,
PstDspFlowThroughProcessApiGrpc,
PstDspFlowThroughProcessApiSimulator,
)
from ska_pst.lmc.dsp.dsp_pipeline_model import (
DspPipelineMonitorData,
DspPipelineMonitorDataStore,
)
from ska_pst.lmc.dsp.dsp_util import calculate_dsp_subband_resources
__all__ = ["PstDspFlowThroughComponentManager"]
[docs]class PstDspFlowThroughComponentManager(
PstProcessApiSubcomponentManager[
DspPipelineMonitorData,
DspPipelineMonitorData,
PstDspFlowThroughProcessApi,
DspPipelineMonitorDataStore,
]
):
"""Component manager for the DSP.FT component for the PST.LMC subsystem."""
_api: PstDspFlowThroughProcessApi
def __init__(
self: PstDspFlowThroughComponentManager,
*,
device_name: str,
process_api_endpoint: str,
api: PstDspFlowThroughProcessApi | None = None,
logger: logging.Logger | None = None,
**kwargs: Any,
):
"""
Initialise instance of the component manager.
:param device_interface: an abstract interface of the TANGO device.
:type device_interface: PstApiDeviceInterface[DspFlowThroughMonitorData]
:param api: an API object used to delegate functionality to.
:type api: `PstProcessApi`
:param logger: a logger for this object to use
:type logger: `logging.Logger`
"""
logger = logger or logging.getLogger(__name__)
logger.debug(
f"Setting up DSP.FT component manager with device_name='{device_name}'"
+ f" and api_endpoint='{process_api_endpoint}'"
)
api = api or PstDspFlowThroughProcessApiSimulator(
logger=logger,
)
super().__init__(
device_name=device_name,
subcomponent_name="dsp_flow_through",
process_api_endpoint=process_api_endpoint,
api=api,
logger=logger,
data_store=DspPipelineMonitorDataStore(),
monitor_data_prefix="dsp",
**kwargs,
)
@override
def _simulator_api(self: PstDspFlowThroughComponentManager) -> PstDspFlowThroughProcessApi:
"""Get instance of the simulator API."""
self.logger.debug("DSP.FT component manager setting up simulated API")
return PstDspFlowThroughProcessApiSimulator(
logger=self.logger,
)
@override
def _grpc_api(self: PstDspFlowThroughComponentManager) -> PstDspFlowThroughProcessApi:
"""Get instance of a gRPC API."""
self.logger.debug("DSP.FT component manager setting up gRPC API")
return PstDspFlowThroughProcessApiGrpc(
client_id=self.subcomponent_id,
grpc_endpoint=self.process_api_endpoint,
logger=self.logger,
)
@override
def validate_configure_scan(self: PstDspFlowThroughComponentManager, configuration: dict) -> None:
"""
Validate a ConfigureScan request sent from CSP.LMC to the DSP.FT sub-component.
This asserts the request can be converted to DSP.FT resources and then calls the process API to
perform the validation.
:param configuration: configuration that would be used when the configure_beam and configure_scan
methods are called.
:type configuration: dict
"""
dsp_resources = calculate_dsp_subband_resources(beam_id=self.beam_id, **configuration)
self._api.validate_configure_beam(configuration=dsp_resources[1])
self._api.validate_configure_scan(configuration=configuration)
@override
def _configure_beam(self: PstDspFlowThroughComponentManager, configuration: dict) -> None:
"""
Configure the beam of the the component with the resources.
:param configuration: configuration for beam
:type configuration: dict
"""
dsp_resources = calculate_dsp_subband_resources(beam_id=self.beam_id, **configuration)
self._api.configure_beam(configuration=dsp_resources[1])
# delegate getting properties from
def __getattr__(self: PstDspFlowThroughComponentManager, name: str) -> Any:
"""
Get attribute of component manager.
This is a Python dunder method that is used to get attributes/properties
that have not been found by the ``__getattribute__`` method.
This allows delegating to properties of the ``monitor_data`` property
:param name: the name of the attribute
:type name: str
:return: the attribute value
:rtype: Any
:raises: AttributeError if attribute cannot be found on monitoring data
"""
return getattr(self.monitor_data, name)