# -*- 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 SMRB capability."""
from __future__ import annotations
import copy
from random import randint
from typing import Any, List, Optional
from overrides import override
from ska_pst.lmc.component import PstSimulator
from ska_pst.lmc.smrb.smrb_model import SmrbMonitorData, SmrbMonitorDataStore, SmrbSubbandMonitorData
[docs]class PstSmrbSimulator(PstSimulator[SmrbSubbandMonitorData, SmrbMonitorDataStore]):
"""Class used for simulating SMRB data."""
def __init__(
self: PstSmrbSimulator,
subband_num_of_buffers: Optional[List[int]] = None,
subband_ring_buffer_sizes: Optional[List[int]] = None,
**kwargs: Any,
) -> None:
"""
Initialise the SMRB simulator.
:param num_subbands: number of subbands, if None a random number is used.
:type num_subbands: int
:param subband_ring_buffer_sizes: list of sizes of subbands
:type subband_ring_buffer_sizes: list of ints
:raises: AssertionError if length of subband sizes not the same as num_subbands.
"""
super().__init__(data_store=SmrbMonitorDataStore(), **kwargs)
configuration: dict = {}
if subband_num_of_buffers is not None:
configuration["subband_num_of_buffers"] = subband_num_of_buffers
if subband_ring_buffer_sizes is not None:
configuration["subband_ring_buffer_sizes"] = subband_ring_buffer_sizes
self.configure_scan(configuration=configuration)
@override
def configure_scan(self: PstSmrbSimulator, configuration: dict) -> None:
"""
Simulate configuring a scan.
:param configuration: the configuration to be configured
:type configuration: dict
:raises: AssertionError if length of subband sizes not the same as num_subbands.
"""
if "num_subbands" in configuration:
self.num_subbands = configuration["num_subbands"]
# set up subband ring buffer sizes
subband_ring_buffer_sizes = configuration.get(
"subband_ring_buffer_sizes", [1048576 * randint(4, 64) for _ in range(self.num_subbands)]
)
assert (
len(subband_ring_buffer_sizes) == self.num_subbands
), f"Expected length of subband_ring_buffer_sizes to be {self.num_subbands}"
subband_num_of_buffers = configuration.get(
"subband_num_of_buffers", [randint(4, 8) for _ in range(self.num_subbands)]
)
assert (
len(subband_num_of_buffers) == self.num_subbands
), f"Expected length of subband_num_of_buffers to be {self.num_subbands}"
self._data_store.reset()
for idx in range(self.num_subbands):
num_of_buffers = subband_num_of_buffers[idx]
buffer_size = subband_ring_buffer_sizes[idx]
self._data_store.update_subband(
subband_id=(idx + 1),
subband_data=SmrbSubbandMonitorData(
buffer_size=buffer_size,
num_of_buffers=num_of_buffers,
),
)
@override
def _update(self: PstSmrbSimulator) -> None:
"""Simulate the update of SMRB data."""
for subband_id, subband_data in self._data_store._subband_data.items():
subband_data = copy.copy(subband_data)
written = randint(2**10, 2**16)
# between 0-70% - too high and it causes TANGO k8s-tests to fail
utilisation = randint(0, 700) / 10.0
full = int(subband_data.num_of_buffers * utilisation / 100.0)
subband_data.total_written += written
subband_data.total_read += written
subband_data.full = full
self._data_store.update_subband(subband_id=subband_id, subband_data=subband_data)
[docs] def get_data(self: PstSmrbSimulator) -> SmrbMonitorData:
"""
Get current SMRB data.
Updates the current simulated data and returns the latest data.
:returns: current simulated SMRB data.
:rtype: :py:class:`SmrbMonitorData`
"""
if self._scan:
self._update()
return self._data_store.monitor_data