# -*- 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 with data classes representing the model of the scan configuration."""
from __future__ import annotations
import dataclasses
from typing import Any
from ska_pst.lmc.dsp.dsp_util import POLARISATION_STATE_NPOL_MAP
[docs]@dataclasses.dataclass
class FlowThroughConfig:
"""Data class representing the Flow Through configuration used in testing."""
rescale_timescale: float = 0.0
"""
The timescale needed to calculate rescale stats, in seconds.
This value is how long in time to sample data before calculating the rescale statistics.
If periodic_update is true, then this is also the period of how often the rescale stats are recalculated.
If this value is set to 0.0, then PST will use the smallest chunk of data available to it to perform the
statistics calculation.
Default: 0.0
"""
rescale_periodic_update: bool = False
"An indicator for whether to recalculate the rescale statistics periodically."
rescale_algorithm: str = "MedianMAD"
"""The algorithm used to determine the scales and offsets when rescaling complex voltage data."""
channels_out: tuple[int, int] | None = None
"""
The selected channel as an exclusive range.
To select all the channels use (-1, -1) which the scan config generator will use as a sentinel
value for all channels
"""
polarisations: str | None = None
"""
The polarisations to select in ChanPolSelect.
Default is we test a random selection of "A", "B" or "Both"
when schema version is v3.x but "X", "Y" or "Both" when
schema version is v4.0 or greater.
"""
num_bits_out: int | None = None
"""
The number of bits to use when performing digitisation.
"""
requantisation_scale: float = 1.0
"""
The scale to apply to values before performing digitisation.
"""
@property
def use_robust_statistics(self: FlowThroughConfig) -> bool:
"""
Get the use robust statistics property.
If ``algorithm`` property is ``MedianMAD`` then this will return ``True``.
:return: the use robust statistics property.
:rtype: bool
"""
return self.rescale_algorithm == "MedianMAD"
@use_robust_statistics.setter
def use_robust_statistics(self: FlowThroughConfig, use_robust: bool) -> None:
"""
Set the use robust statistics property.
This will set the ``algorithm`` property to ``MedianMAD`` if the parameter is
``True`` else it will set the ``algorithm`` property to ``MeanStdDev``.
:param use_robust: the use robust statistics property value.
:type use_robust: bool
"""
self.rescale_algorithm = "MedianMAD" if use_robust else "MeanStdDev"
[docs]@dataclasses.dataclass
class DetectedFilterbankConfig:
"""Data class representing the Detected Filterbank configuration used in testing."""
dispersion_measure: float = 1.0
"""The dispersion measure for coherent/incoherent de-dispersion."""
rotation_measure: float | None = None
"""The rotation measure for phase-coherent Faraday rotation correction."""
nchan_out: int | None = None
"""
The number of output channels.
If not set this will be determined by number of input channels and frequency_decimation_factor.
"""
polarisation_state: str = "Stokes"
"""
The output polarisation state.
For Detected filterbank the only valid values are 'Intensity' and 'Stokes'.
"""
num_bits_out: int | None = None
"""
The number of bits to use when performing digitisation.
"""
time_decimation_factor: int = 1
"""The number of input samples per output time sample."""
frequency_decimation_factor: int = 1
"""The number of input frequency channels incoherently added to each output frequency channel."""
requantisation_scale: float = 1.0
"""Scale factor to govern the dynamic range for fixed precision output during requantisation."""
rescale_timescale: float = 1.0
"""Length of data to be used when determining the scaling factors used during requantisation."""
rescale_periodic_update: bool = False
"An indicator for whether to recalculate the rescale statistics periodically."
rescale_algorithm: str = "MedianMAD"
"""The algorithm used to determine the scales and offsets when rescaling complex voltage data."""
def __post_init__(self: DetectedFilterbankConfig) -> None:
"""Post constructor to ensure ``polarisation_state`` is a valid value."""
assert self.polarisation_state in {"Intensity", "Stokes"}, (
"expected polarisation_state to be either 'Intensity' or 'Stokes' "
f"but was {self.polarisation_state}"
)
@property
def npol(self: DetectedFilterbankConfig) -> int:
"""
The number of polarisations based on the ``polarisation_state``.
If ``polarisation_state == 'Intensity'`` then this returns 1 else this
returns 4.
"""
return POLARISATION_STATE_NPOL_MAP[self.polarisation_state]
def __setattr__(self: DetectedFilterbankConfig, name: str, value: Any) -> None:
"""Override setattr to enforce validation of polarisation_state."""
if name == "polarisation_state":
assert value in {"Intensity", "Stokes"}, (
"expected polarisation_state to be either 'Intensity' or 'Stokes' " f"but was {value}"
)
super().__setattr__(name, value)