"""
@file subservice_parser.py
@brief Parses the Subservice definitions for the Mission Information Base.
@details Used by the MIB Exporter, inherits generic File Parser
@author R. Mueller
@date 14.11.2019
Example Stringset to scan for:
enum Subservice: uint8_t {
//!< [EXPORT] : [COMMAND] Perform connection test
CONNECTION_TEST = 1,
//!< [EXPORT] : [REPLY] Connection test reply
CONNECTION_TEST_REPORT = 2,
EVENT_TRIGGER_TEST = 128, //!< [EXPORT] : [COMMAND] Trigger test reply and test event
MULTIPLE_EVENT_TRIGGER_TEST = 129, //!< [EXPORT] : [COMMAND] Trigger multiple events (5)
MULTIPLE_CONNECTION_TEST = 130 //!< [EXPORT] : [COMMAND] Trigger multiple connection tests
};
"""
import re
from enum import Enum
from fsfwgen.parserbase.file_list_parser import FileListParser
from fsfwgen.parserbase.parser import FileParser
from fsfwgen.utility.csv_writer import CsvWriter
from fsfwgen.utility.printer import Printer
SUBSERVICE_DEFINITION_DESTINATION = ["../../mission/", "../../fsfw/pus/"]
SUBSERVICE_CSV_NAME = "mib_subservices.csv"
SUBSERVICE_COLUMN_HEADER = [
"Service",
"Subservice Name",
"Subservice Number",
"Type",
"Comment",
]
SQL_DELETE_SUBSVC_CMD = """
DROP TABLE IF EXISTS Subservice;
"""
SQL_CREATE_SUBSVC_CMD = """
CREATE TABLE IF NOT EXISTS Subservice(
id INTEGER PRIMARY KEY,
service INTEGER,
subsvcName TEXT,
subsvcNumber INTEGER,
type TEXT CHECK( type IN ('TC','TM')),
comment TEXT
)
"""
SQL_INSERT_INTO_SUBSVC_CMD = """
INSERT INTO Subservice(service,subsvcName,subsvcNumber,type,comment)
VALUES(?,?,?,?,?)
"""
class SubserviceColumns(Enum):
"""
Specifies order of MIB columns
"""
SERVICE = 0
NAME = 1
NUMBER = 2
TYPE = 3
COMMENT = 4
Clmns = SubserviceColumns
def main():
"""
If this file is run separately, this main will be run.
:return:
"""
header_parser = FileListParser(SUBSERVICE_DEFINITION_DESTINATION)
header_file_list = header_parser.parse_header_files(
False, "Parsing subservice header files: "
)
packet_subservice_parser = SubserviceParser(header_file_list)
subservice_table = packet_subservice_parser.parse_files()
Printer.print_content(subservice_table, "Printing subservice table:")
print("Found " + str(len(subservice_table)) + " subservice entries.")
subservice_writer = CsvWriter(
SUBSERVICE_CSV_NAME, subservice_table, SUBSERVICE_COLUMN_HEADER
)
subservice_writer.write_to_csv()
subservice_writer.move_csv("..")
# TODO: Not really happy with the multi-line implementation, but this is not trivial..
# Right not, we are not using the last lines stored, we just store the string
# of the last line (if its only a comment). It propably would be better to always
# scan 3 or 4 lines at once. However, this is not easy too..
# pylint: disable=too-few-public-methods
class SubserviceParser(FileParser):
"""
This parser class can parse the subservice definitions.
"""
def __init__(self, file_list: list):
super().__init__(file_list)
# Column System allows reshuffling of table columns in constructor
self.clmns_len = SubserviceColumns.__len__()
# this table includes the current new table entry,
# which will be updated for target parameter
self.dict_entry_list = list(range(self.clmns_len))
self.dict_entry_list[Clmns.COMMENT.value] = ""
self.subservice_enum_found = False
# This list will store the last three lines for longer comments.
self.last_line_list = ["", "", ""]
# If an export command was found, cache the possibility of a match.
self.possible_match_on_next_lines = False
# This is called for every file
def _handle_file_parsing(self, file_name: str, *args: any, **kwargs):
self_print_parsing_info = False
if len(args) == 1 and isinstance(args[0], bool):
self_print_parsing_info = args[0]
# Read service from file name
service_match = re.search("Service[^0-9]*([0-9]{1,3})", file_name)
if service_match:
self.dict_entry_list[Clmns.SERVICE.value] = service_match.group(1)
self.dict_entry_list[Clmns.NAME.value] = " "
file = open(file_name, "r")
if self_print_parsing_info:
print("Parsing " + file_name + " ...")
# Scans each line for possible variables
for line in file.readlines():
self.__handle_line_reading(line)
def __handle_line_reading(self, line):
"""
Handles the reading of single lines.
:param line:
:return:
"""
# Case insensitive matching
enum_match = re.search(r"[\s]*enum[\s]*Subservice([^\n]*)", line, re.IGNORECASE)
if enum_match:
self.subservice_enum_found = True
if self.subservice_enum_found:
self.__handle_enum_scanning(line)
self.last_line_list[2] = self.last_line_list[1]
self.last_line_list[1] = self.last_line_list[0]
self.last_line_list[0] = line
def __handle_enum_scanning(self, line: str):
"""
Two-line reading. First check last line. For export command.
"""
self.__scan_for_export_command(self.last_line_list[0])
subservice_match = self.__scan_subservices(line)
if subservice_match:
self.index = self.index + 1
dict_entry_tuple = tuple(self.dict_entry_list[: self.clmns_len])
self.mib_table.update({self.index: dict_entry_tuple})
self.__clear_tuple()
def __clear_tuple(self):
self.dict_entry_list[Clmns.NAME.value] = ""
self.dict_entry_list[Clmns.TYPE.value] = ""
self.dict_entry_list[Clmns.NUMBER.value] = ""
self.dict_entry_list[Clmns.COMMENT.value] = ""
self.possible_match_on_next_lines = False
def __scan_for_export_command(self, line: str) -> bool:
command_string = re.search(
r"([^\[]*)\[export\][: ]*\[([\w]*)\][\s]*([^\n]*)", line, re.IGNORECASE
)
if command_string:
# Check whether there is a separated export command
# (export command is not on same line as subservice definition)
# ugly solution but has worked so far.
string = command_string.group(1).lstrip()
if len(string) <= 8:
self.possible_match_on_next_lines = True
if self.__scan_for_type(line):
self.__scan_for_comment(line)
return True
self.__add_possible_comment_string(line)
return False
def __add_possible_comment_string(self, line):
"""
If no command was found, the line might be a continuation of a comment.
Strip whitespaces and comment symbols and add to comment buffer.
"""
possible_multiline_comment = line.lstrip()
possible_multiline_comment = possible_multiline_comment.lstrip("/")
possible_multiline_comment = possible_multiline_comment.lstrip("<")
possible_multiline_comment = possible_multiline_comment.lstrip("!")
possible_multiline_comment = possible_multiline_comment.rstrip()
if len(possible_multiline_comment) > 0:
self.dict_entry_list[Clmns.COMMENT.value] += possible_multiline_comment
def __scan_subservices(self, line):
"""
Scan for subservice match.
:param line:
:return:
"""
subservice_match = re.search(
r"[\s]*([\w]*)[\s]*=[\s]*([0-9]{1,3})(?:,)?(?:[ /!<>]*([^\n]*))?", line
)
if subservice_match:
self.dict_entry_list[Clmns.NAME.value] = subservice_match.group(1)
self.dict_entry_list[Clmns.NUMBER.value] = subservice_match.group(2)
# I am assuming that an export string is longer than 7 chars.
if len(subservice_match.group(3)) > 7:
# Export command on same line overrides old commands. Read for comment.
if self.__process_comment_string(subservice_match.group(3)):
return True
# Check whether exporting was commanded on last lines
return bool(self.possible_match_on_next_lines)
if re.search(r"}[\s]*;", line):
self.subservice_enum_found = False
return subservice_match
def __process_comment_string(self, comment_string) -> bool:
# look for packet type specifier
export_command_found = self.__scan_for_type(comment_string)
# Look for everything after [EXPORT] : [TYPESPECIFIER] as comment
if export_command_found:
self.__scan_for_comment(comment_string)
return export_command_found
def __scan_for_type(self, string) -> bool:
type_match = re.search(r"\[reply\]|\[tm\]", string, re.IGNORECASE)
if type_match:
self.dict_entry_list[Clmns.TYPE.value] = "TM"
return True
type_match = re.search(r"\[command\]|\[tc\]", string, re.IGNORECASE)
if type_match:
self.dict_entry_list[Clmns.TYPE.value] = "TC"
return True
self.dict_entry_list[Clmns.TYPE.value] = "Unspecified"
return False
def __scan_for_comment(self, comment_string):
comment_match = re.search(r":[\s]*\[[\w]*\][\s]*([^\n]*)", comment_string)
if comment_match:
self.dict_entry_list[Clmns.COMMENT.value] = comment_match.group(1)
def _post_parsing_operation(self):
pass
if __name__ == "__main__":
main()