Skip to content
Snippets Groups Projects
Commit 488b7db0 authored by Kien Le's avatar Kien Le
Browse files

Get station code and network code from packet

parent a6baf296
No related branches found
No related tags found
1 merge request!191Read log files generated by users
......@@ -115,7 +115,8 @@ class LogFile:
self.file.close()
PACKET_PARSERS: Dict[Optional[LogFileFormat], Callable] = {
Parser = Callable[[List[str]], Tuple[List[str], List[str]]]
PACKET_PARSERS: Dict[Optional[LogFileFormat], Parser] = {
None: parse_soh_packet_no_type,
'sohstationviewer': parse_soh_packet_sohstationviewer,
'rt2ms': parse_soh_packet_rt2ms,
......@@ -123,29 +124,59 @@ PACKET_PARSERS: Dict[Optional[LogFileFormat], Callable] = {
}
def get_experiment_number(soh_lines: List[str]):
if not soh_lines[0].startswith('Station Channel Definition'):
return None
# The experiment number can be in either the first or second line after
# the header. These lines are indented, so we have to strip them of
# whitespace.
if soh_lines[1].strip().startswith('Experiment Number ='):
experiment_number_line = soh_lines[1].split()
elif soh_lines[2].strip().startswith('Experiment Number ='):
experiment_number_line = soh_lines[2].split()
else:
return None
# If the experiment number is not recorded, we know that it will be 0. In
# order to not have too many return statements, we add 0 to the experiment
# line instead of returning it immediately.
if len(experiment_number_line) < 4:
experiment_number_line.append('0')
return experiment_number_line[-1]
class LogFileReader:
"""
Class that reads a log file.
"""
def __init__(self, file_path: Path):
self.file_path = file_path
self.packet_reader = None
self.log_file_type: Optional[LogFileFormat] = None
self.eh_et_lines = []
self.soh_lines = []
self.station_code: Optional[str] = None
self.experiment_number: Optional[str] = None
def read(self):
log_file = LogFile(self.file_path)
for packet in log_file:
if self.log_file_type is None:
self.log_file_type = detect_log_file_packet_format(packet)
parser = PACKET_PARSERS[self.log_file_type]
eh_et_lines, soh_lines = parser(packet)
if self.station_code is None and soh_lines:
self.station_code = soh_lines[0].split(' ')[-1].strip()
if self.experiment_number is None and soh_lines:
found_experiment_number = get_experiment_number(soh_lines)
self.experiment_number = found_experiment_number
self.eh_et_lines.extend(eh_et_lines)
self.soh_lines.extend(soh_lines)
if __name__ == '__main__':
pass
a = LogFileReader(Path('/Users/kle/PycharmProjects/sohstationviewer/RT130_9BB3_2016326_2016365.log'))
a = LogFileReader(Path('/Users/kle/PycharmProjects/sohstationviewer/tests/test_data/potato.sdr/LOGS/RT130_92EB_2017150_2017150.log'))
a.read()
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