diff --git a/CAL/CalibrationCommon/lib/datacontainers/holography_specification.py b/CAL/CalibrationCommon/lib/datacontainers/holography_specification.py
index 544cf2201f7de88c539277b71ae6f86413fcd2a1..584fe3a5730ddd6d82ce66762d579206536fb86d 100644
--- a/CAL/CalibrationCommon/lib/datacontainers/holography_specification.py
+++ b/CAL/CalibrationCommon/lib/datacontainers/holography_specification.py
@@ -1,10 +1,14 @@
import datetime
-from collections import defaultdict
-import re
import os
+import re
+from collections import defaultdict
from glob import glob
+
class HolographyStationBeamSpecification(object):
+ """
+ Contains the data regarding the beam specification for a single station.
+ """
station_name = None
rcus_mode = None
sub_band_ids = ()
@@ -17,12 +21,24 @@ class HolographyStationBeamSpecification(object):
class HolographySpecification(object):
+ """
+ The HolographySpecification represents the set of input used to specify an holography
+ observation with the LOFAR telescope
+ """
hs_name_pattern = r'Holog-(?P<date>\d{8})-(?P<comment>.*)-(?P<id>\d{3}).txt'
def __init__(self, name, path):
+ """
+ Construct the holography specification object and extracts id date and comment from
+ the file name (see HolographySpecification.extract_id_date_comment_from_name)
+ :param name: file name of the holography specification
+ :type name: str
+ :param path: path of the holography specification file
+ :type path: str
+ """
self.path = os.path.join(path, name)
self.name = name
- self.id, self.date, self.comment = HolographySpecification.\
+ self.id, self.date, self.comment = HolographySpecification. \
extract_id_date_comment_from_name(name)
self.reference_station_names = None
self.target_station_names = None
@@ -34,12 +50,65 @@ class HolographySpecification(object):
def __repr__(self):
return 'HolographySpecification(%s, %s, %s, %s, %s)' % (
- self.id,
- self.date,
- self.comment,
- self.name,
- self.path,
- )
+ self.id,
+ self.date,
+ self.comment,
+ self.name,
+ self.path,
+ )
+
+ @staticmethod
+ def list_bsf_files_in_path(path):
+ """
+ List all the Holography beam specification files in the given path
+ :param path: path to the beam specification files
+ :type path: str
+ :return: the list of Holography Specifications
+ :rtype: list[HolographySpecification]
+ """
+ bsf_files_name_pattern = 'Holog-*.txt'
+ bsf_files_path_pattern = os.path.join(path, bsf_files_name_pattern)
+ matched_path_list = glob(bsf_files_path_pattern)
+ matched_file_path_list = list(
+ filter(lambda path_i: os.path.isfile(path_i), matched_path_list))
+ matched_file_name_list = list(map(lambda path_i: os.path.basename(path_i),
+ matched_file_path_list))
+ holography_beam_specification = HolographySpecification.create_hs_list_from_name_list_and_path(
+ matched_file_name_list, path)
+ return holography_beam_specification
+
+ def get_beam_specifications_per_station_name(self, station_name):
+ """
+ Returns a list of beam specifications for the given station name
+ :param station_name: name of the station
+ :type station_name: str
+ :return: a list of beam specifications
+ :rtype: list[HolographyStationBeamSpecification]
+ """
+ return self.station_specification_map[station_name]
+
+ def read_file(self):
+ lines = self._read_lines()
+ self._parse_header(lines[0])
+ self._parse_lines(lines[1:])
+ self._update_class_attributes()
+
+ @staticmethod
+ def create_hs_list_from_name_list_and_path(name_list, path):
+ return [HolographySpecification(name, path) for name in name_list]
+
+ @staticmethod
+ def is_holography_specification_file_name(name):
+ return re.match(HolographySpecification.hs_name_pattern, name) is not None
+
+ @staticmethod
+ def extract_id_date_comment_from_name(name):
+ match = re.match(HolographySpecification.hs_name_pattern, name)
+ date = match.group('date')
+ hs_id = int(match.group('id'))
+ comment = match.group('comment')
+ date = datetime.datetime.strptime(date, '%Y%m%d')
+ return hs_id, date, comment
def _read_lines(self):
with open(self.path, 'r') as fstream_in:
@@ -47,28 +116,46 @@ class HolographySpecification(object):
@staticmethod
def _split_header(line):
+ """
+ Split the header of holography specification file
+ :param line: line containing the header
+ :return: a dict containing the start_date,
+ the end_date, the rcu_mode, and the beam_switch_delay
+ :rtype: dict
+ """
date_regex = '\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}'
format = r'(?P<start_date>{date_regex})\s*' \
- '(?P<end_date>{date_regex})\s*' \
- '(?P<rcu_mode>\d*)\s*' \
- '(?P<beam_switch_delay>\d*.\d*)'.format(date_regex=date_regex)
+ '(?P<end_date>{date_regex})\s*' \
+ '(?P<rcu_mode>\d*)\s*' \
+ '(?P<beam_switch_delay>\d*.\d*)'.format(date_regex=date_regex)
match = re.match(format, line)
return match.groupdict()
def _parse_header(self, header):
+ """
+ Parse the header
+ :param header: string containing the header
+ """
date_time_format = '%Y-%m-%d %H:%M:%S'
- splitted_header = HolographySpecification._split_header(header)
+ split_header = HolographySpecification._split_header(header)
- self.start_datetime = datetime.datetime.strptime(splitted_header['start_date'],
- date_time_format)
- self.end_datetime = datetime.datetime.strptime(splitted_header['end_date'],
+ self.start_datetime = datetime.datetime.strptime(split_header['start_date'],
+ date_time_format)
+ self.end_datetime = datetime.datetime.strptime(split_header['end_date'],
date_time_format)
- self.rcu_mode = splitted_header['rcu_mode']
- self.beam_set_interval = splitted_header['beam_switch_delay']
+ self.rcu_mode = split_header['rcu_mode']
+ self.beam_set_interval = split_header['beam_switch_delay']
@staticmethod
def _split_line(line):
+ """
+ Split the header of holography specification file
+ :param line: line containing the header
+ :return: a dict containing the start_date,
+ the end_date, the rcu_mode, and the beam_switch_delay
+ :rtype: dict
+ """
range_regex = '(\d*\:\d*)|(\d*)'
ra_dec_regex = '\d*\.\d*,-?\d*\.\d*,\w*'
regex = r'^(?P<station_name>\w*)\s*' \
@@ -127,54 +214,38 @@ class HolographySpecification(object):
return return_value
@staticmethod
- def _parse_line(splitted_line):
-
-
+ def _parse_line(split_line):
+ """
+ Parses a line in the holography specification file
+ :param split_line: dict containing the row indexed by the column name
+ :type split_line: dict[str, str]
+ """
beam_specification = HolographyStationBeamSpecification()
- beam_specification.rcus_mode = int(splitted_line['rcus_mode'])
- beam_specification.sub_band_ids = [int(sub_band) for sub_band in splitted_line['sub_band'].split(',')]
- beam_specification.mode_description = splitted_line['mode_description']
+ beam_specification.rcus_mode = int(split_line['rcus_mode'])
+ beam_specification.sub_band_ids = [int(sub_band) for sub_band in
+ split_line['sub_band'].split(',')]
+ beam_specification.mode_description = split_line['mode_description']
beam_specification.rcus_involved = HolographySpecification._parse_integer_range_or_list(
- splitted_line['rcus'])
- beam_specification.beamlets = splitted_line['beamlets']
+ split_line['rcus'])
+ beam_specification.beamlets = split_line['beamlets']
beam_specification.station_pointing = HolographySpecification._parse_pointing(
- splitted_line['station_pointing'])
+ split_line['station_pointing'])
beam_specification.virtual_pointing = HolographySpecification._parse_pointing(
- splitted_line['virtual_pointing'])
+ split_line['virtual_pointing'])
if len(beam_specification.sub_band_ids) == 1:
beam_specification.station_type = 'target'
else:
beam_specification.station_type = 'reference'
- beam_specification.station_name = splitted_line['station_name']
+ beam_specification.station_name = split_line['station_name']
return beam_specification
- @staticmethod
- def list_bsf_files_in_path(path):
- """
- List all the Holography beam specification files in the given path
- :param path: path to the beam specification files
- :type path: str
- :return: the list of Holography Specifications
- :rtype: list[HolographySpecification]
- """
- bsf_files_name_pattern = 'Holog-*.txt'
- bsf_files_path_pattern = os.path.join(path, bsf_files_name_pattern)
- matched_path_list = glob(bsf_files_path_pattern)
- matched_file_path_list = list(
- filter(lambda path_i: os.path.isfile(path_i), matched_path_list))
- matched_file_name_list = list(map(lambda path_i: os.path.basename(path_i),
- matched_file_path_list))
- holography_beam_specification = HolographySpecification.create_hs_list_from_name_list_and_path(
- matched_file_name_list, path)
- return holography_beam_specification
-
def _parse_lines(self, lines):
- splitted_lines = HolographySpecification._split_lines(lines)
+ split_lines = HolographySpecification._split_lines(lines)
- for line in splitted_lines:
+ for line in split_lines:
parsed_line = HolographySpecification._parse_line(line)
self.station_specification_map[parsed_line.station_name] += [parsed_line]
@@ -182,42 +253,8 @@ class HolographySpecification(object):
def _update_class_attributes(self):
self.station_names = self.station_specification_map.keys()
self.reference_station_names = [station_name for station_name in
- self.station_specification_map
- if len(self.station_specification_map[station_name]) == 1]
- self.target_station_names = [station_name for station_name in
self.station_specification_map
- if len(self.station_specification_map[station_name]) > 1]
-
- def get_beam_specifications_per_station_name(self, station_name):
- """
- Returns a list of beam specifications for the given station name
- :param station_name: name of the station
- :type station_name: str
- :return: a list of beam specifications
- :rtype: list[HolographyStationBeamSpecification]
- """
- return self.station_specification_map[station_name]
-
- def read_file(self):
- lines = self._read_lines()
- self._parse_header(lines[0])
- self._parse_lines(lines[1:])
- self._update_class_attributes()
-
- @staticmethod
- def create_hs_list_from_name_list_and_path(name_list, path):
- return [HolographySpecification(name, path) for name in name_list]
-
- @staticmethod
- def is_holography_specification_file_name(name):
- return re.match(HolographySpecification.hs_name_pattern, name) is not None
-
- @staticmethod
- def extract_id_date_comment_from_name(name):
- match = re.match(HolographySpecification.hs_name_pattern, name)
- date = match.group('date')
- hs_id = int(match.group('id'))
- comment = match.group('comment')
- date = datetime.datetime.strptime(date, '%Y%m%d')
- return hs_id, date, comment
-
+ if len(self.station_specification_map[station_name]) == 1]
+ self.target_station_names = [station_name for station_name in
+ self.station_specification_map
+ if len(self.station_specification_map[station_name]) > 1]