# -*- 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 SMRB PST component manager."""
from __future__ import annotations
import logging
from typing import Any, List
from overrides import override
from ska_pst.lmc.component import PstProcessApiSubcomponentManager
from ska_pst.lmc.smrb.smrb_model import SmrbMonitorData, SmrbMonitorDataStore, SmrbSubbandMonitorData
from ska_pst.lmc.smrb.smrb_process_api import (
PstSmrbProcessApi,
PstSmrbProcessApiGrpc,
PstSmrbProcessApiSimulator,
)
from ska_pst.lmc.smrb.smrb_util import calculate_smrb_subband_resources
__all__ = ["PstSmrbComponentManager"]
[docs]class PstSmrbComponentManager(
PstProcessApiSubcomponentManager[
SmrbSubbandMonitorData, SmrbMonitorData, PstSmrbProcessApi, SmrbMonitorDataStore
]
):
"""Component manager for the SMRB component for the PST.LMC subsystem."""
def __init__(
self: PstSmrbComponentManager,
*,
device_name: str,
process_api_endpoint: str,
api: PstSmrbProcessApi | 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[SmrbMonitorData]
: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 SMRB component manager with device_name='{device_name}'"
+ f"and api_endpoint='{process_api_endpoint}'"
)
api = api or PstSmrbProcessApiSimulator(
logger=logger,
)
super().__init__(
device_name=device_name,
subcomponent_name="smrb",
process_api_endpoint=process_api_endpoint,
api=api,
logger=logger,
data_store=SmrbMonitorDataStore(),
**kwargs,
)
@property
def ring_buffer_utilisation(self: PstSmrbComponentManager) -> float:
"""
Get the percentage of the ring buffer elements that are full of data.
:returns: the percentage of the ring buffer elements that are full of data.
:rtype: float
"""
return self.monitor_data.ring_buffer_utilisation
@property
def ring_buffer_size(self: PstSmrbComponentManager) -> int:
"""
Get the capacity of the ring buffer, in bytes.
:returns: the capacity of the ring buffer, in bytes.
:rtype: int
"""
return self.monitor_data.ring_buffer_size
@property
def number_subbands(self: PstSmrbComponentManager) -> int:
"""
Get the number of sub-bands.
:returns: the number of sub-bands.
:rtype: int
"""
return self.monitor_data.number_subbands
@property
def ring_buffer_read(self: PstSmrbComponentManager) -> int:
"""
Get the amount of data, in bytes, that has been read.
:returns: the amount of data that has been read.
:rtype: int
"""
return self.monitor_data.ring_buffer_read
@property
def ring_buffer_written(self: PstSmrbComponentManager) -> int:
"""
Get the amount of data, in bytes, that has been written.
:returns: the amount of data that has been written.
:rtype: int
"""
return self.monitor_data.ring_buffer_written
@property
def subband_ring_buffer_utilisations(self: PstSmrbComponentManager) -> List[float]:
"""
Get the percentage of full ring buffer elements for each sub-band.
:returns: the percentage of full ring buffer elements for each sub-band.
:rtype: List[float]
"""
return self.monitor_data.subband_ring_buffer_utilisations
@property
def subband_ring_buffer_sizes(self: PstSmrbComponentManager) -> List[int]:
"""
Get the capacity of ring buffers for each sub-band.
:returns: the capacity of ring buffers, in bytes, for each sub-band.
:rtype: List[int]
"""
return self.monitor_data.subband_ring_buffer_sizes
@property
def subband_ring_buffer_read(self: PstSmrbComponentManager) -> List[int]:
"""
Get the capacity of ring buffers for each sub-band.
:returns: the capacity of ring buffers, in bytes, for each sub-band.
:rtype: List[int]
"""
return self.monitor_data.subband_ring_buffer_read
@property
def subband_ring_buffer_written(self: PstSmrbComponentManager) -> List[int]:
"""
Get the capacity of ring buffers for each sub-band.
:returns: the capacity of ring buffers, in bytes, for each sub-band.
:rtype: List[int]
"""
return self.monitor_data.subband_ring_buffer_written
@override
def _simulator_api(self: PstSmrbComponentManager) -> PstSmrbProcessApi:
"""Get instance of the simulator API."""
self.logger.debug("SMRB component manager setting up simulated API")
return PstSmrbProcessApiSimulator(
logger=self.logger,
)
@override
def _grpc_api(self: PstSmrbComponentManager) -> PstSmrbProcessApi:
"""Get instance of a gRPC API."""
self.logger.debug("SMRB component manager setting up gRPC API")
return PstSmrbProcessApiGrpc(
client_id=self.subcomponent_id,
grpc_endpoint=self.process_api_endpoint,
logger=self.logger,
)
@override
def validate_configure_scan(self: PstSmrbComponentManager, configuration: dict) -> None:
"""
Validate a ConfigureScan request sent from CSP.LMC to the SMRB sub-component.
This asserts the request can be converted to SMRB 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
"""
smrb_resources = calculate_smrb_subband_resources(beam_id=self.beam_id, **configuration)
self._api.validate_configure_beam(configuration=smrb_resources[1])
self._api.validate_configure_scan(configuration=configuration)
@override
def _configure_beam(self: PstSmrbComponentManager, configuration: dict) -> None:
"""
Configure beam resources in the component.
:param configuration: configuration for beam
:type configuration: dict
"""
smrb_resources = calculate_smrb_subband_resources(beam_id=self.beam_id, **configuration)
self._api.configure_beam(configuration=smrb_resources[1])