# -*- 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 STAT capability."""
from __future__ import annotations
from typing import Any, Tuple
import numpy as np
from overrides import override
from ska_pst.lmc.component import SUBBAND_1, PstSimulator
from ska_pst.lmc.stat.stat_model import StatMonitorData, StatMonitorDataStore
[docs]class PstStatSimulator(PstSimulator[StatMonitorData, StatMonitorDataStore]):
"""Class used for simulating STAT data."""
def __init__(
self: PstStatSimulator,
**kwargs: Any,
) -> None:
"""Initialise the STAT simulator."""
# STAT was only written to support 1 subband
kwargs["num_subbands"] = 1
super().__init__(data_store=StatMonitorDataStore(), **kwargs)
self.configure_scan(configuration={})
@override
def configure_scan(self: PstStatSimulator, 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.
"""
self._data_store.update_subband(
subband_id=SUBBAND_1,
subband_data=StatMonitorData(),
)
@override
def _update(self: PstStatSimulator) -> None:
"""Simulate the update of STAT data."""
def _gen_stats() -> Tuple[float, float, int]:
data = np.random.randn(1024).astype(dtype=np.float32)
clipped_samples = np.abs(data) >= 3.0
num_clipped_samples = np.count_nonzero(clipped_samples)
data[clipped_samples] = 3.0 * np.sign(data[clipped_samples])
mean = np.mean(data, dtype=float)
variance = np.var(data, ddof=1, dtype=float)
return (mean, variance, num_clipped_samples)
data: dict = {}
for dim in ["real", "imag"]:
for pol in ["pol_a", "pol_b"]:
for suffix in ["", "_rfi_excised"]:
mean_key = f"{dim}_{pol}_mean_freq_avg{suffix}"
variance_key = f"{dim}_{pol}_variance_freq_avg{suffix}"
num_clipped_samples_key = f"{dim}_{pol}_num_clipped_samples{suffix}"
mean, variance, num_clipped_samples = _gen_stats()
data[mean_key] = mean
data[variance_key] = variance
data[num_clipped_samples_key] = num_clipped_samples
self._data_store.update_subband(
subband_id=SUBBAND_1,
subband_data=StatMonitorData(**data),
)
[docs] def get_data(self: PstStatSimulator) -> StatMonitorData:
"""
Get current STAT data.
Updates the current simulated data and returns the latest data.
:returns: current simulated STAT data.
:rtype: :py:class:`StatMonitorData`
"""
if self._scan:
self._update()
return self._data_store.monitor_data