From 58457b05a6e104edf34baa5bb6291753839887a3 Mon Sep 17 00:00:00 2001
From: mancini <mancini@astron.nl>
Date: Thu, 11 Jul 2019 17:53:12 +0200
Subject: [PATCH] SSB-47: parsing header values and implemented read function
to obtain a dict object with the station and mode as keys
---
.../lib/datacontainers/calibration_table.py | 47 +++++++++++++++----
.../test/t_calibration_table.py | 5 +-
2 files changed, 41 insertions(+), 11 deletions(-)
diff --git a/CAL/CalibrationCommon/lib/datacontainers/calibration_table.py b/CAL/CalibrationCommon/lib/datacontainers/calibration_table.py
index 65b597a648b..f55296d30da 100644
--- a/CAL/CalibrationCommon/lib/datacontainers/calibration_table.py
+++ b/CAL/CalibrationCommon/lib/datacontainers/calibration_table.py
@@ -1,14 +1,15 @@
import logging
-from datetime import datetime
+from glob import glob
+from os import path
from re import fullmatch
from struct import iter_unpack, pack
from typing import BinaryIO
+from typing import Dict, Tuple
from typing import List
-from h5py import File
-from os import path
+
from dataclasses import dataclass, asdict, field
+from h5py import File
from numpy import empty as empty_ndarray, ndarray, fromiter as array_from_iter, float64, array_equal
-from glob import glob
logger = logging.getLogger(__name__)
@@ -81,21 +82,34 @@ def parse_data(data_buffer):
return complex_data
-@dataclass(init=True, repr=True, frozen=True, eq=True)
+@dataclass(init=True, repr=True, frozen=False, eq=True)
class CalibrationTable:
observation_station: str
- observation_mode: str
+ observation_mode: int
observation_antennaset: str
observation_band: str
observation_source: str
- observation_date: datetime
+ observation_date: str
calibration_version: int
calibration_name: str
- calibration_date: datetime
+ calibration_date: str
calibration_ppsdelay: List[int]
data: ndarray = field(compare=False)
comment: str = ''
+ def __parse_attributes(self):
+ self.observation_mode = int(self.observation_mode)
+ self.calibration_version = int(self.calibration_version)
+ if isinstance(self.calibration_ppsdelay, str):
+ self.calibration_ppsdelay = list(map(int, self.calibration_ppsdelay.
+ lstrip('[').
+ rstrip(']').
+ strip().
+ split(' ')))
+
+ def __post_init__(self):
+ self.__parse_attributes()
+
@staticmethod
def load_from_file(file_path):
with open(file_path, 'rb') as file_stream:
@@ -115,8 +129,12 @@ class CalibrationTable:
if key is 'data':
# skipping field data
continue
+ if key is 'calibration_ppsdelay':
+ serialized_value = '[%s ]' % ' '.join(map(str, self.calibration_ppsdelay))
+ else:
+ serialized_value = str(value)
serialized_name = _ATTRIBUTE_NAME_TO_SERIALIZED_NAME[key]
- serialized_line = '{} = {}\n'.format(serialized_name, value).encode('utf8')
+ serialized_line = '{} = {}\n'.format(serialized_name, serialized_value).encode('utf8')
f_stream.write(serialized_line)
f_stream.write(b'HeaderStop\n')
@@ -143,7 +161,6 @@ class CalibrationTable:
def __eq__(self, other):
return super().__eq__(other) and array_equal(self.data, other.data)
-
def store_to_file(self, file_path):
with open(file_path, 'wb') as file_stream:
self.__serialize_header(file_stream)
@@ -157,3 +174,13 @@ def read_calibration_tables_in_directory(directory_path: str):
return [CalibrationTable.load_from_file(file_path)
for file_path in glob(files, recursive=False)]
+
+
+def read_calibration_tables_per_station_mode(directory_path: str) -> Dict[Tuple[str, int],
+ CalibrationTable]:
+ list_of_calibration_tables = read_calibration_tables_in_directory(directory_path)
+ result = dict()
+ for calibration_table in list_of_calibration_tables:
+ result[calibration_table.observation_station, calibration_table.observation_mode] = \
+ calibration_table
+ return result
diff --git a/CAL/CalibrationCommon/test/t_calibration_table.py b/CAL/CalibrationCommon/test/t_calibration_table.py
index ceef98089c8..67e42b0075e 100644
--- a/CAL/CalibrationCommon/test/t_calibration_table.py
+++ b/CAL/CalibrationCommon/test/t_calibration_table.py
@@ -1,6 +1,6 @@
import unittest
from lofar.calibration.common.datacontainers.calibration_table import CalibrationTable,\
- UnvalidFileException, read_calibration_tables_in_directory
+ UnvalidFileException, read_calibration_tables_in_directory, read_calibration_tables_per_station_mode
from h5py import File as H5File
from os import getcwd
import logging
@@ -56,6 +56,9 @@ class TestCalibrationTable(unittest.TestCase):
obtained_calibration_table = calibration_tables[0]
self.assertEqual(expected_calibration_table, obtained_calibration_table)
+ def test_read_calibration_tables_per_station_mode(self):
+ calibration_tables_dict = read_calibration_tables_per_station_mode(getcwd())
+ self.assertIn(('CS401', 5), calibration_tables_dict.keys())
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
--
GitLab