import logging
from dataclasses import dataclass
from typing import Sequence
from realtime.receive.core import baseline_utils
logger = logging.getLogger(__name__)
[docs]
@dataclass
class Baselines:
"""
A sequence of antenna index combinations making up a complete
set of antenna baselines.
"""
antenna1: Sequence[int]
antenna2: Sequence[int]
def __init__(self, antenna1: Sequence[int], antenna2: Sequence[int]):
if len(antenna1) != len(antenna2):
raise ValueError(
"Mismatching sequence lengths of antennas provided"
f"{len(antenna1)} != {len(antenna2)}"
)
# Check we have a full set of baselines. To do this we generate the
# baselines with/without autocorrelations that would result from the
# input antennas, and verify that the input baselines are one of those
# (with antenna1/2 in the original order or swapped).
all_antennas = list(set(antenna1).union(antenna2))
autocorr_baselines = baseline_utils.generate_baselines(
all_antennas, lower_triangular=True, autocorr=True
)
noautocorr_baselines = baseline_utils.generate_baselines(
all_antennas, lower_triangular=True, autocorr=False
)
input_baselines_original = sorted(list(zip(antenna1, antenna2)))
input_baselines_swapped = sorted(list(zip(antenna2, antenna1)))
if sorted(autocorr_baselines) in (input_baselines_original, input_baselines_swapped):
logger.debug("Baselines with autocorrelations detected")
elif sorted(noautocorr_baselines) in (input_baselines_original, input_baselines_swapped):
logger.debug("Baselines without autocorrelations detected")
else:
raise ValueError(
f"Invalid baseline constituients provided ({antenna1=}, {antenna2=}), "
"they don't represent a full set of baselines, with or without auto-correlation "
f"for the number of antennas ({len(all_antennas)}) they contain."
)
self.antenna1 = antenna1
self.antenna2 = antenna2
[docs]
@staticmethod
def generate(num_stations: int, autocorr: bool, lower_triangular: bool = False):
"""
Generate a Baselines object for the given number of stations and taking
into account if autocorrelation is requested.
:param num_stations: The number of stations
:param autocorr: Whether autocorrelated baselines should be accounted for.
:param lower_triangular: Whether to generate baselines in
lower-triangular order
:see: :func:`realtime.receive.core.baseline_utils.baseline_indices`
"""
indices = baseline_utils.baseline_indices(num_stations, lower_triangular, autocorr)
return Baselines(
[i[0] for i in indices],
[i[1] for i in indices],
)
[docs]
def as_tuples(self):
"""
Returns the baselines as a list of tuples.
The dataclasses.astuples does not seem to work as I need to
interleave the antenna1 and antenna2 lists.
"""
list_of_tuples = []
for i in enumerate(self.antenna1):
new_tuple = (self.antenna1[i[0]], self.antenna2[i[0]])
list_of_tuples.append(new_tuple)
return list_of_tuples
def __len__(self):
return len(self.antenna1)