# -*- coding: utf-8 -*-
"""some utils to get information """
import logging
from typing import Sequence, Union
import numpy as np
import pandas
from realtime.receive.core import Antenna, Baselines, baseline_utils
from realtime.receive.core.antenna_utils import load_antennas
from ska_sdp_datamodels.configuration import Configuration
logger = logging.getLogger(__name__)
[docs]
class TelescopeManager:
"""
Telescope Manager containing an immutable telescope model consisting on
the antennas/stations composing a subarray.
"""
def __init__(
self,
antennas: Union[Sequence[Antenna], str],
baselines: Baselines,
):
if isinstance(antennas, str):
self._antennas = load_antennas(antennas)
else:
self._antennas = tuple(antennas)
self._baselines = baselines
baseline_utils.validate_antenna_and_baseline_counts(self.num_stations, self.num_baselines)
self._as_configuration = Configuration.constructor(
names=[a.label for a in self._antennas],
xyz=[(a.x, a.y, a.z) for a in self._antennas],
mount=np.repeat("FIXED", len(self._antennas)),
diameter=[a.dish_diameter for a in self._antennas],
)
self._baselines_as_visibility_indices = pandas.MultiIndex.from_arrays(
list(zip(*baselines.as_tuples())), names=["antenna1", "antenna2"]
)
@property
def num_stations(self) -> int:
"""The number of stations used by the current observation"""
return len(self._antennas)
@property
def num_baselines(self) -> int:
"""The number of baselines used by the current observation"""
return len(self._baselines)
[docs]
def get_antennas(self) -> Sequence[Antenna]:
"""Returns all antennas"""
return self._antennas
[docs]
def get_baselines(self) -> Baselines:
"""Returns all baselines"""
return self._baselines
@property
def as_configuration(self) -> Configuration:
"""self -> SDP Configuration"""
return self._as_configuration
@property
def baselines_as_visibility_indices(self) -> pandas.MultiIndex:
"""Baselines as a MultiIndex suitable for the Visibility class"""
return self._baselines_as_visibility_indices