# -*- 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 handling mapping of values between different data sources."""
from __future__ import annotations
__all__ = [
"ValueMapping",
"to_si",
"split_values",
"map_dada_value",
]
import dataclasses
from typing import Any, Callable, Dict, List, Type, TypeAlias, TypeVar
import astropy.units as u
T = TypeVar("T")
ValueSupplier: TypeAlias = Callable[[dict], Any]
def _dict_key_mapping(key: str, alternate_key: str | None = None) -> ValueSupplier:
def _map(config: dict) -> Any:
try:
value = config
for k in key.split("/"):
value = value[k]
except KeyError:
if alternate_key is not None:
value = config
for k in alternate_key.split("/"):
value = value[k]
else:
raise
return value
return _map
[docs]@dataclasses.dataclass(kw_only=True)
class ValueMapping:
"""A data class to define the mapping between different metadata sources."""
config_key: str | None = dataclasses.field(default=None)
"""
Key to use to get value from scan configuration.
A value of ``None`` means there is no mapping.
"""
alternate_config_key: str | None = dataclasses.field(default=None)
"""
Alternate config key to use to get from the scan configuration.
If the ``config_key`` is set but does not exist in the configuration
but this property is set, then this value will be used as the
configuration key.
This key allows for verifying keys that have been renamed in
version of the PST schema, such as ``itrf`` being renamed to ``delay_centre``.
"""
metadata_key: str | None = dataclasses.field(default=None)
"""
Key to use to get value from the metadata file.
A value of ``None`` means there is no mapping.
"""
file_key: str | None = dataclasses.field(default=None)
"""
Key to use to get value from a DADA file header.
A value of ``None`` means there is no mapping.
"""
def __post_init__(self: ValueMapping) -> None:
"""Perform post initialisation of object."""
if self.config_key is None:
self._config_value_supplier = None
else:
self._config_value_supplier = _dict_key_mapping(self.config_key, self.alternate_config_key)
if self.metadata_key is None:
self._metadata_value_supplier = None
else:
self._metadata_value_supplier = _dict_key_mapping(self.metadata_key)
if self.file_key is None:
self._file_value_supplier = None
else:
self._file_value_supplier = _dict_key_mapping(self.file_key)
[docs] def config_value(self: ValueMapping, config: dict) -> Any | None:
"""Get the value from the scan configuration.
:param config: a dictionary of the scan configuration.
:type config: dict
:return: the value in the dictionary or None if the value does not
exist.
:rtype: Any | None
"""
if self._config_value_supplier is None:
return None
try:
return self._config_value_supplier(config)
except KeyError:
return None
[docs] def file_value(self: ValueMapping, header: dict) -> Any | None:
"""Get the value from a DADA file header.
:param config: a dictionary of the DADA file header values.
:type config: dict
:return: the value in the dictionary or None if the value does not
exist.
:rtype: Any | None
"""
if self._file_value_supplier is None:
return None
assert self.file_key is not None
try:
value = self._file_value_supplier(header)
if value is not None:
value = map_dada_value(dada_key=self.file_key, dada_value=value)
return value
except KeyError:
return None
[docs]def to_si(unit: u.UnitBase) -> Callable[[str], float]:
"""
Convert value to SI unit based on input unit.
Example is that FREQ and BW in the DADA files are in MHz
but the SI is Hz.
:param unit: the unit the value is in (e.g. u.MHz)
:type unit: u.UnitBase
:return: a callable to will convert a string value into
an float in SI quantity value.
:rtype: Callable[[str], float]
"""
def _map(value: str) -> float:
return u.Quantity(value, unit=unit).si.value
return _map
[docs]def split_values(dtype: Type[T], delimiter: str = ",") -> Callable[[str], List[T]]:
"""
Split a delimited string into a list of values of type T.
The default delimiter is a comma but this could be overridden.
An example usage of this is the ANTENNAE (strings),
ANT_WEIGHTS (floats), OS_FACTOR (ints separated by "/")
:param dtype: the data type the individual types should be.
:type dtype: Type[T]
:param delimiter: the delimiter to split values.
:type delimiter: str
:return: a callable that will convert a string into a list
of values of `Type[T]`
:rtype: Callable[[str], List[T]]
"""
def _map(value: str) -> List[T]:
return [dtype(v.strip()) for v in value.split(delimiter)] # type: ignore
return _map
DADA_HEADER_CONVERTER_MAPPING: Dict[str, Callable[[str], Any]] = {
"SCAN_ID": int,
"UDP_NSAMP": int,
"WT_NSAMP": int,
"UDP_NCHAN": int,
"FD_HAND": int,
"FD_SANG": float,
"FA_REQ": float,
"NANT": int,
"ANTENNAE": split_values(dtype=str),
"ANT_WEIGHTS": split_values(dtype=float),
"NPOL": int,
"NBIT": int,
"OS_FACTOR": split_values(dtype=int, delimiter="/"),
"DELAY_CENTRE": split_values(dtype=float),
"BMAJ": float, # not used atm
"BMIN": float, # not used atm
"SCANLEN_MAX": int,
"BW": to_si(unit=u.MHz),
"NCHAN": int,
"END_CHANNEL": int,
"FREQ": to_si(unit=u.MHz),
"END_CHANNEL_OUT": int,
"NCHAN_OUT": int,
"BW_OUT": to_si(unit=u.MHz),
"FREQ_OUT": to_si(unit=u.MHz),
"NBIT_OUT": int,
"CHAN_FT": split_values(dtype=int),
"DIGITIZER_SCALE": float,
"RESCALE_TIMESCALE": float,
"RESCALE_PERIODIC_UPDATE": bool,
"TSAMP": float,
"BYTES_PER_SECOND": float,
}
def map_dada_value(dada_key: str, dada_value: str) -> Any:
"""Map a DADA header value to a value that can be compared with scan config.
If there is no mapping applied then the default is just to return the string value.
:param dada_key: the DADA header key
:type dada_key: str
:param dada_value: the str value from the DADA header
:type dada_value: str
:return: a converted value
"""
try:
return DADA_HEADER_CONVERTER_MAPPING[dada_key](dada_value)
except KeyError:
return dada_value