Merge branch 'L2SS-2213-generate-caltables' into 'master'

Resolve L2SS-2213 "Generate caltables"

Closes L2SS-2213

See merge request !1091

README.md

@@ -150,6 +150,7 @@ Next change the version in the following places:
    through [https://git.astron.nl/lofar2.0/tango/-/tags](Deploy Tags)
 
 # Release Notes
+* 0.51.1 Generate caltables from LOFAR1 data or by generating dummies
 * 0.51.0 Add Calibration service to gRPC server
 * 0.50.1 Generate and combine CDB files for all stations in Gitlab CI/CD
 * 0.50.0 Add pqr to etrs rotation matrix to metadata

tangostationcontrol/VERSION

-0.50.1
+0.50.2

tangostationcontrol/toolkit/lofar1/calibration_table_LOFAR1.py

+from datetime import datetime
 import logging
 from glob import glob
 from os import path
@@ -62,6 +63,7 @@ class CalibrationTable:
     _HEADER_LINE_PATTERN = (
         r"(^[A-z]*\.[A-z]*\.[A-z]*\s=\s.*$)|(^[A-z]*\.[A-z]*\s=\s.*$)"
     )
+    _LINK_LINE_PATTERN = r"link (CalTable-[0-9]{3}-(LBA|HBA)\_(INNER|OUTER|SPARSE_(ODD|EVEN))-[0-9]{2,3}\_[0-9]{2,3}\.dat)$"
     _FREQUENCIES = 512
     _FLOATS_PER_FREQUENCY = 2
     _N_ANTENNAS_DUTCH = 96
@@ -82,10 +84,48 @@ class CalibrationTable:
     observation_band: str = ""
 
     @staticmethod
-    def load_from_file(file_path):
+    def generate_dummy(
+        observation_station: str, observation_antennaset: str, observation_band: str
+    ):
+        logger.warning(
+            "generating dummy calibration for %s %s_%s due to missing LOFAR1 files",
+            observation_station,
+            observation_antennaset,
+            observation_band,
+        )
+        header: dict = {
+            "calibration_name": "Dummy",
+            "calibration_date": datetime.now().strftime("%Y%m%d"),
+            "calibration_version": "0",
+            "calibration_ppsdelay": "[0]",
+            "observation_mode": "0",
+            "observation_source": "None",
+            "observation_date": datetime.now().strftime("%Y%m%d%H%M"),
+            "observation_station": observation_station,
+            "observation_antennaset": observation_antennaset,
+            "observation_band": observation_band,
+        }
+        if "CS" in observation_station or "RS" in observation_station:
+            data = CalibrationTable._generate_data(CalibrationTable._N_ANTENNAS_DUTCH)
+        else:
+            data = CalibrationTable._generate_data(
+                CalibrationTable._N_ANTENNAS_INTERNATIONAL
+            )
+        return CalibrationTable(**header, data=data)
+
+    @staticmethod
+    def load_from_file(directory: str, filename: str):
+        file_path = f"{directory}/{filename}"
         logger.info("loading file %s", file_path)
         with open(file_path, "rb") as file_stream:
             header = CalibrationTable._extract_header(file_stream)
+            if "link" in header:
+                logger.warning(
+                    "linked file, using file %s as replacement for %s",
+                    header["link"],
+                    file_path,
+                )
+                return CalibrationTable.load_from_file(directory, header["link"])
             data_raw = file_stream.read().rstrip(b"\n")
         try:
             data = CalibrationTable._parse_data(data_raw)
@@ -180,7 +220,7 @@ class CalibrationTable:
         f_stream.write(data_packed)
 
     @staticmethod
-    def _extract_header(fstream: BinaryIO):
+    def _extract_header(fstream: BinaryIO) -> dict:
         header = {}
         for _ in range(CalibrationTable._MAX_HEADER_LINES):
             line = fstream.readline().decode("utf8").rstrip("\n")
@@ -197,6 +237,10 @@ class CalibrationTable:
                 )
                 value = value.strip()
                 header[key] = value
+            elif fullmatch(CalibrationTable._LINK_LINE_PATTERN, line):
+                key, value = line.split(" ")
+                header[key] = value
+                break
             else:
                 logger.error('unrecognized line "%s"', line)
                 raise InvalidFileException('unrecognized line "%s"' % line)
@@ -267,6 +311,16 @@ class CalibrationTable:
 
         return complex_data
 
+    @staticmethod
+    def _generate_data(n_antennas: int):
+        complex_data = empty_ndarray(
+            [CalibrationTable._FREQUENCIES, n_antennas], dtype=complex
+        )
+        for elem in complex_data:
+            elem.real = 1.0
+            elem.imag = 0.0
+        return complex_data
+
     def __post_init__(self):
         self._parse_header()
 

tangostationcontrol/toolkit/lofar1/caltable.py

@@ -22,9 +22,9 @@ class LOFAR1CalTables(dict):
     }
     """
 
-    def __init__(self, station, tmpdir="."):
-        self.station = station
-        self.tmpdir = tmpdir
+    def __init__(self, station: str, tmpdir: str = "."):
+        self.station: str = station
+        self.tmpdir: str = tmpdir
 
         self.files = [
             f"CalTable-{self.station_nr:03}-HBA-110_190.dat",
@@ -34,25 +34,45 @@ class LOFAR1CalTables(dict):
             f"CalTable-{self.station_nr:03}-LBA_OUTER-10_90.dat",
         ]
 
-        self.download_all()
-        self.parse_all()
+        failed = self.download_all()
+        self.parse_files(list(filter(lambda v: v not in failed, self.files)))
+        self.generate_failed(failed)
+
+    def extract_file_info(self, file: str) -> tuple[str, str]:
+        filename, ext = file.rsplit(".", 1)
+        _caltable, _stationnr, antennaset, band = filename.split("-")
+        return (
+            antennaset,
+            band,
+        )
 
     @property
     def station_nr(self):
         return int(self.station[2:])
 
-    def download_all(self):
+    def download_all(self) -> list[str]:
+        failed: list[str] = []
         for filename in self.files:
             with open(f"{self.tmpdir}/{filename}", "wb") as f:
                 url = f"https://git.astron.nl/ro/lofar1-caltables/-/raw/main/{self.station.upper()}/{filename}"
+                try:
                     data = urllib.request.urlopen(url).read()
                     f.write(data)
+                except Exception:
+                    failed.append(filename)
+        return failed
 
-    def parse_all(self):
-        for f in self.files:
-            filename, ext = f.rsplit(".", 1)
-            _caltable, _stationnr, antennaset, band = filename.split("-")
+    def parse_files(self, files: list[str]):
+        for f in files:
+            antennaset, band = self.extract_file_info(f)
 
             self[f"{antennaset}-{band}"] = CalibrationTable.load_from_file(
-                f"{self.tmpdir}/{f}"
+                self.tmpdir, f
+            )
+
+    def generate_failed(self, files: list[str]):
+        for f in files:
+            antennaset, band = self.extract_file_info(f)
+            self[f"{antennaset}-{band}"] = CalibrationTable.generate_dummy(
+                self.station, antennaset, band
             )