# -*- 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 PST DSP component manager."""
from __future__ import annotations
import logging
from typing import Any
from overrides import override
from ska_pst.lmc.component import PstProcessApiSubcomponentManager
from .dsp_grpc_api_strategy import PstDspGrpcApiStrategy
from .dsp_model import DspMonitorData, DspMonitorDataStore
from .dsp_simulator import PstDspSimulator
from .dsp_util import calculate_dsp_subband_resources
__all__ = ["PstDspComponentManager"]
[docs]class PstDspComponentManager(
PstProcessApiSubcomponentManager[
DspMonitorData,
DspMonitorData,
DspMonitorDataStore,
]
):
"""Component manager for the DSP component for the PST.LMC subsystem."""
def __init__(
self: PstDspComponentManager,
*,
device_name: str,
process_api_endpoint: str,
logger: logging.Logger | None = None,
**kwargs: Any,
):
"""
Initialise instance of the component manager.
:param device_name: the name of the device that the component manager is for
:type device_name: str
:param process_api_endpoint: the URL for the gRPC process
:type process_api_endpoint: str
: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 component manager with device_name='{device_name}'"
+ f" and api_endpoint='{process_api_endpoint}'"
)
super().__init__(
device_name=device_name,
subcomponent_name="dsp",
process_api_endpoint=process_api_endpoint,
simulator=PstDspSimulator(),
grpc_strategy=PstDspGrpcApiStrategy(),
logger=logger,
data_store=DspMonitorDataStore(),
**kwargs,
)
@override
def validate_configure_scan(self: PstDspComponentManager, configuration: dict) -> None:
"""
Validate a ConfigureScan request sent from CSP.LMC to the DSP sub-component.
This asserts the request can be converted to DSP 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
"""
pst_processing_mode = configuration["pst_processing_mode"]
dsp_resources = calculate_dsp_subband_resources(beam_id=self.beam_id, **configuration)
self._api.validate_configure_beam(
configuration=dsp_resources[1], pst_processing_mode=pst_processing_mode
)
self._api.validate_configure_scan(configuration=configuration)
@override
def _configure_beam(self: PstDspComponentManager, configuration: dict) -> None:
"""
Configure the beam of the component with the resources.
:param configuration: configuration for beam
:type configuration: dict
"""
pst_processing_mode = configuration["pst_processing_mode"]
dsp_resources = calculate_dsp_subband_resources(beam_id=self.beam_id, **configuration)
self._api.configure_beam(configuration=dsp_resources[1], pst_processing_mode=pst_processing_mode)