Skip to content
Snippets Groups Projects
Commit 585529c5 authored by Mattia Mancini's avatar Mattia Mancini
Browse files

SSB-42: adding comments and reformatting code.

parent d0c378aa
No related branches found
No related tags found
1 merge request!44Merge back holography to master
import datetime import datetime
from collections import defaultdict
import re
import os import os
import re
from collections import defaultdict
from glob import glob from glob import glob
class HolographyStationBeamSpecification(object): class HolographyStationBeamSpecification(object):
"""
Contains the data regarding the beam specification for a single station.
"""
station_name = None station_name = None
rcus_mode = None rcus_mode = None
sub_band_ids = () sub_band_ids = ()
...@@ -17,12 +21,24 @@ class HolographyStationBeamSpecification(object): ...@@ -17,12 +21,24 @@ class HolographyStationBeamSpecification(object):
class HolographySpecification(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' hs_name_pattern = r'Holog-(?P<date>\d{8})-(?P<comment>.*)-(?P<id>\d{3}).txt'
def __init__(self, name, path): 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.path = os.path.join(path, name)
self.name = 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) extract_id_date_comment_from_name(name)
self.reference_station_names = None self.reference_station_names = None
self.target_station_names = None self.target_station_names = None
...@@ -34,12 +50,65 @@ class HolographySpecification(object): ...@@ -34,12 +50,65 @@ class HolographySpecification(object):
def __repr__(self): def __repr__(self):
return 'HolographySpecification(%s, %s, %s, %s, %s)' % ( return 'HolographySpecification(%s, %s, %s, %s, %s)' % (
self.id, self.id,
self.date, self.date,
self.comment, self.comment,
self.name, self.name,
self.path, 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): def _read_lines(self):
with open(self.path, 'r') as fstream_in: with open(self.path, 'r') as fstream_in:
...@@ -47,28 +116,46 @@ class HolographySpecification(object): ...@@ -47,28 +116,46 @@ class HolographySpecification(object):
@staticmethod @staticmethod
def _split_header(line): 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}' date_regex = '\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}'
format = r'(?P<start_date>{date_regex})\s*' \ format = r'(?P<start_date>{date_regex})\s*' \
'(?P<end_date>{date_regex})\s*' \ '(?P<end_date>{date_regex})\s*' \
'(?P<rcu_mode>\d*)\s*' \ '(?P<rcu_mode>\d*)\s*' \
'(?P<beam_switch_delay>\d*.\d*)'.format(date_regex=date_regex) '(?P<beam_switch_delay>\d*.\d*)'.format(date_regex=date_regex)
match = re.match(format, line) match = re.match(format, line)
return match.groupdict() return match.groupdict()
def _parse_header(self, header): def _parse_header(self, header):
"""
Parse the header
:param header: string containing the header
"""
date_time_format = '%Y-%m-%d %H:%M:%S' 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'], self.start_datetime = datetime.datetime.strptime(split_header['start_date'],
date_time_format) date_time_format)
self.end_datetime = datetime.datetime.strptime(splitted_header['end_date'], self.end_datetime = datetime.datetime.strptime(split_header['end_date'],
date_time_format) date_time_format)
self.rcu_mode = splitted_header['rcu_mode'] self.rcu_mode = split_header['rcu_mode']
self.beam_set_interval = splitted_header['beam_switch_delay'] self.beam_set_interval = split_header['beam_switch_delay']
@staticmethod @staticmethod
def _split_line(line): 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*)' range_regex = '(\d*\:\d*)|(\d*)'
ra_dec_regex = '\d*\.\d*,-?\d*\.\d*,\w*' ra_dec_regex = '\d*\.\d*,-?\d*\.\d*,\w*'
regex = r'^(?P<station_name>\w*)\s*' \ regex = r'^(?P<station_name>\w*)\s*' \
...@@ -127,54 +214,38 @@ class HolographySpecification(object): ...@@ -127,54 +214,38 @@ class HolographySpecification(object):
return return_value return return_value
@staticmethod @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 = HolographyStationBeamSpecification()
beam_specification.rcus_mode = int(splitted_line['rcus_mode']) beam_specification.rcus_mode = int(split_line['rcus_mode'])
beam_specification.sub_band_ids = [int(sub_band) for sub_band in splitted_line['sub_band'].split(',')] beam_specification.sub_band_ids = [int(sub_band) for sub_band in
beam_specification.mode_description = splitted_line['mode_description'] split_line['sub_band'].split(',')]
beam_specification.mode_description = split_line['mode_description']
beam_specification.rcus_involved = HolographySpecification._parse_integer_range_or_list( beam_specification.rcus_involved = HolographySpecification._parse_integer_range_or_list(
splitted_line['rcus']) split_line['rcus'])
beam_specification.beamlets = splitted_line['beamlets'] beam_specification.beamlets = split_line['beamlets']
beam_specification.station_pointing = HolographySpecification._parse_pointing( beam_specification.station_pointing = HolographySpecification._parse_pointing(
splitted_line['station_pointing']) split_line['station_pointing'])
beam_specification.virtual_pointing = HolographySpecification._parse_pointing( beam_specification.virtual_pointing = HolographySpecification._parse_pointing(
splitted_line['virtual_pointing']) split_line['virtual_pointing'])
if len(beam_specification.sub_band_ids) == 1: if len(beam_specification.sub_band_ids) == 1:
beam_specification.station_type = 'target' beam_specification.station_type = 'target'
else: else:
beam_specification.station_type = 'reference' beam_specification.station_type = 'reference'
beam_specification.station_name = splitted_line['station_name'] beam_specification.station_name = split_line['station_name']
return beam_specification 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): 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) parsed_line = HolographySpecification._parse_line(line)
self.station_specification_map[parsed_line.station_name] += [parsed_line] self.station_specification_map[parsed_line.station_name] += [parsed_line]
...@@ -182,42 +253,8 @@ class HolographySpecification(object): ...@@ -182,42 +253,8 @@ class HolographySpecification(object):
def _update_class_attributes(self): def _update_class_attributes(self):
self.station_names = self.station_specification_map.keys() self.station_names = self.station_specification_map.keys()
self.reference_station_names = [station_name for station_name in 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 self.station_specification_map
if len(self.station_specification_map[station_name]) > 1] if len(self.station_specification_map[station_name]) == 1]
self.target_station_names = [station_name for station_name in
def get_beam_specifications_per_station_name(self, station_name): self.station_specification_map
""" if len(self.station_specification_map[station_name]) > 1]
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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment