# -*- coding: utf-8 -*-
"""some utils to get information """
import logging
from typing import List, Union
from casacore import tables
from realtime.receive.core import Antenna, msutils
from realtime.receive.core.baselines import Baselines
from realtime.receive.modules.tm.base_tm import TelescopeManager
logger = logging.getLogger(__name__)
def _load_antennas_from_table(input_table_name: str) -> List[Antenna]:
"""
If the input table name exists - set the input table load the values into a dictionary
:param str input_table_name: The actual ANTENNA table in a measurement set
"""
table = tables.table(input_table_name, ack=False)
name = table.col("NAME")
pos = table.col("POSITION")
diam = table.col("DISH_DIAMETER")
antennas = []
for row in range(table.nrows()):
location = {
"geocentric": {
"x": pos[row][0],
"y": pos[row][1],
"z": pos[row][2],
}
}
fixed_delays = {}
niao = 0.0
antenna = Antenna(
interface="test",
station_label=name[row],
station_id=row,
diameter=diam[row],
location=location,
fixed_delays=fixed_delays,
niao=niao,
)
antennas.append(antenna)
return antennas
[docs]
class MeasurementSetTM(TelescopeManager):
"""
Telescope Manager that reads its model information from a Measurement Set.
"""
def __init__(self, ms: Union[msutils.MeasurementSet, str]):
if isinstance(ms, str):
_ms = msutils.MeasurementSet.open(ms)
else:
_ms = ms
logger.info("Attempting to build model from %s", _ms.name)
antennas = _load_antennas_from_table(f"{_ms.name}/ANTENNA")
num_baselines = msutils.calc_baselines(_ms)
antenna1 = _ms.read_column("ANTENNA1", 0, num_baselines)
antenna2 = _ms.read_column("ANTENNA2", 0, num_baselines)
baselines = Baselines(antenna1, antenna2)
super().__init__(antennas, baselines)
if isinstance(ms, str):
_ms.close()