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

Parse log file packets

parent 78128dbc
No related branches found
No related tags found
1 merge request!191Read log files generated by users
from pathlib import Path
from typing import List, Literal, Optional
from typing import List, Literal, Optional, Dict, Callable, Tuple
LogFileFormat = Literal['rt2ms', 'logpeek', 'sohstationviewer']
PACKETS = ['SH', 'SC', 'OM', 'DS', 'AD', 'CD', 'FD', 'EH', 'ET']
......@@ -47,16 +47,45 @@ def detect_log_file_packet_format(packet: List[str]) -> LogFileFormat:
return 'logpeek'
def read_soh_packet_base(packet: List[str]):
pass
def parse_soh_packet_no_type(packet: List[str]):
eh_et_lines = []
soh_lines = packet
return eh_et_lines, soh_lines
def read_soh_packet_rt2ms(packet: List[str]):
pass
def parse_soh_packet_logpeek(packet: List[str]):
eh_et_lines = []
soh_lines = []
for line in packet:
if line.startswith('DAS: '):
eh_et_lines.append(line)
else:
soh_lines.append(line)
return eh_et_lines, soh_lines
def parse_soh_packet_rt2ms(packet: List[str]):
eh_et_lines = []
soh_lines = []
if packet[0].startswith('EH') or packet[0].startswith('ET'):
# The event info is summarized in the last line of an event info packet
eh_et_lines = [packet[-1]]
else:
soh_lines = packet[1:]
return eh_et_lines, soh_lines
def parse_soh_packet_sohstationviewer(packet: List[str]):
eh_et_lines = []
soh_lines = []
if packet[0].startswith('Events:'):
eh_et_lines = packet[1:]
else:
soh_lines = packet
return eh_et_lines, soh_lines
def read_soh_packet_rt2ms(packet: List[str]):
pass
class LogFile:
......@@ -86,6 +115,14 @@ class LogFile:
self.file.close()
PACKET_PARSERS: Dict[Optional[LogFileFormat], Callable] = {
None: parse_soh_packet_no_type,
'sohstationviewer': parse_soh_packet_sohstationviewer,
'rt2ms': parse_soh_packet_rt2ms,
'logpeek': parse_soh_packet_logpeek
}
class LogFileReader:
"""
Class that reads a log file.
......@@ -94,16 +131,21 @@ class LogFileReader:
self.file_path = file_path
self.packet_reader = None
self.log_file_type: Optional[LogFileFormat] = None
self.eh_et_lines = []
self.soh_lines = []
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)
pass
parser = PACKET_PARSERS[self.log_file_type]
eh_et_lines, soh_lines = parser(packet)
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/9BB3.log'))
a = LogFileReader(Path('/Users/kle/PycharmProjects/sohstationviewer/RT130_9BB3_2016326_2016365.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