# -*- 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.
"""Module for providing the Simulated DSP Flow Through capability for PST."""
from __future__ import annotations
import logging
from random import randint
from overrides import override
from ska_pst.lmc.component import PstSimulator
from ska_pst.lmc.dsp.dsp_pipeline_model import DspPipelineMonitorData, DspPipelineMonitorDataStore
__all__ = ["PstDspFlowThroughSimulator"]
[docs]class PstDspFlowThroughSimulator(PstSimulator[DspPipelineMonitorData, DspPipelineMonitorDataStore]):
"""
Simulator for the DSP Flow Through process of the PST.LMC sub-system.
This is currently a stub to allow the DSP.FT functionality of the LMC
to work. When the monitoring attributes are known and how to simulate
them then this class will be update.
"""
def __init__(
self: PstDspFlowThroughSimulator,
num_subbands: int | None = None,
logger: logging.Logger | None = None,
) -> None:
"""
Initialise the DSP Flow Through simulator.
:param num_subbands: number of subbands, if None a random number is used.
:type num_subbands: int
:param logger: logger to use in the simulator, default is None.
:type logger: logging.Logger | None, optional
"""
super().__init__(data_store=DspPipelineMonitorDataStore(), logger=logger)
configuration: dict = {}
if num_subbands is not None:
configuration["num_subbands"] = num_subbands
self.configure_scan(configuration=configuration)
self._scan = False
@override
def configure_scan(self: PstDspFlowThroughSimulator, configuration: dict) -> None:
"""
Simulate configuring a scan.
Only the "num_subbands" parameter is used by this simulator.
:param configuration: the configuration to be configured
:type configuration: dict
"""
if "num_subbands" in configuration:
self.num_subbands = configuration["num_subbands"]
else:
self.num_subbands = randint(1, 4)
self._data_store.reset()
for idx in range(self.num_subbands):
self._data_store.update_subband(
subband_id=(idx + 1),
subband_data=DspPipelineMonitorData(),
)
@override
def _update(self: PstDspFlowThroughSimulator) -> None:
"""Simulate the update of DSP Flow Through data."""
for idx in range(self.num_subbands):
self._data_store.update_subband(
subband_id=(idx + 1),
subband_data=DspPipelineMonitorData(),
)
[docs] def get_data(self: PstDspFlowThroughSimulator) -> DspPipelineMonitorData:
"""
Get current DSP Flow Through data.
Updates the current simulated data and returns the latest data.
:returns: current simulated DSP Flow Through data.
:rtype: :py:class:`DspFlowThroughMonitorData`
"""
if self._scan:
self._update()
return self._data_store.monitor_data