358 lines
12 KiB
Python
358 lines
12 KiB
Python
#! /usr/bin/python3.8
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
@file mod_exporter.py
|
|
@brief Mission Information Base Exporter for the SOURCE project by KSat.
|
|
@details
|
|
Parses OBSW which is based on FSFW developed by the Institute of Space Systems (IRS) Stuttgart.
|
|
Python 3.8 required
|
|
This exporter generates the MIB from the SOURCE On-Board Software directly
|
|
by using file parser implementations
|
|
This exporter has the following capabilities :
|
|
1. Export MIB tables CSV files
|
|
2. Export MIB tables into a SQL database
|
|
|
|
This exporter currently has parser for following data:
|
|
1. Objects
|
|
2. Returnvalues
|
|
3. Packet content (Telemetry/Telecommands)
|
|
4. Events
|
|
5. Subservices
|
|
6. Device Commands
|
|
7. Global datapool
|
|
|
|
@developers
|
|
Basic Instructions to implement new parserbase:
|
|
This parser uses a generic parser class. A specific parser implementation
|
|
can be built by implementing the generic parser class.
|
|
The parser generally takes a list with all files to parse and a dictionary
|
|
with the structure of the MiB table.
|
|
This website can be used to experiment with regular expressions: https://regex101.com/
|
|
|
|
TODO:
|
|
1. Maybe make this file object oriented too.
|
|
"""
|
|
import os
|
|
import pprint
|
|
|
|
from utility.mib_csv_writer import CsvWriter
|
|
from utility.mib_printer import Printer, PrettyPrinter
|
|
from utility.mib_sql_writer import SqlWriter
|
|
from utility import mib_globals as g
|
|
from parserbase.mib_file_list_parser import FileListParser
|
|
from packetcontent.packet_content_parser import (
|
|
PacketContentParser,
|
|
PACKET_CONTENT_DEFINITION_DESTINATION,
|
|
PACKET_CONTENT_CSV_NAME,
|
|
PACKET_CONTENT_HEADER_COLUMN,
|
|
SQL_CREATE_PACKET_DATA_CONTENT_CMD,
|
|
SQL_INSERT_PACKET_DATA_CMD,
|
|
SQL_DELETE_PACKET_DATA_CONTENT_CMD,
|
|
)
|
|
from subservice.subservice_parser import (
|
|
SubserviceParser,
|
|
SUBSERVICE_DEFINITION_DESTINATION,
|
|
SUBSERVICE_CSV_NAME,
|
|
SUBSERVICE_COLUMN_HEADER,
|
|
SQL_CREATE_SUBSVC_CMD,
|
|
SQL_DELETE_SUBSVC_CMD,
|
|
SQL_INSERT_INTO_SUBSVC_CMD,
|
|
)
|
|
from devicecommands.device_command_parser import (
|
|
DeviceHandlerInformationParser,
|
|
DeviceHandlerCommandParser,
|
|
DH_COMMAND_PACKET_DEFINITION_DESTINATION,
|
|
DH_DEFINITION_DESTINATION,
|
|
DH_COMMANDS_CSV_NAME,
|
|
DH_COMMAND_HEADER_COLUMNS,
|
|
SQL_CREATE_CMDTABLE_CMD,
|
|
SQL_INSERT_INTO_CMDTABLE_CMD,
|
|
SQL_DELETE_CMDTABLE_CMD,
|
|
)
|
|
from returnvalues.returnvalues_parser import (
|
|
InterfaceParser,
|
|
ReturnValueParser,
|
|
INTERFACE_DEFINITION_FILES,
|
|
RETURNVALUE_DESTINATIONS,
|
|
sql_retval_exporter,
|
|
CSV_RETVAL_FILENAME,
|
|
)
|
|
from objects.objects import (
|
|
ObjectDefinitionParser,
|
|
OBJECTS_DEFINITIONS,
|
|
export_object_file,
|
|
CSV_OBJECT_FILENAME,
|
|
sql_object_exporter,
|
|
)
|
|
|
|
DO_EXPORT_MIB = True
|
|
PRINT_TABLES_TO_CONSOLE = False
|
|
EXPORT_TO_CSV = True
|
|
EXPORT_TO_SQL = True
|
|
COPY_FILE = False
|
|
COPY_DESTINATION = "."
|
|
FILE_SEPARATOR = ";"
|
|
EXECUTE_SQL_COMMANDS = False
|
|
|
|
|
|
def main():
|
|
"""
|
|
Performs MIB generation.
|
|
"""
|
|
parse_mib()
|
|
|
|
|
|
def parse_mib():
|
|
"""
|
|
This is the core function. It builds parses all files,
|
|
builds all tables and returns them in a tuple.
|
|
The structure of respective tables is generated in a
|
|
separate functions and is easily modifiable:
|
|
:return:
|
|
"""
|
|
handle_subservices_generation()
|
|
print()
|
|
# handle_packet_content_generation()
|
|
# print()
|
|
# handle_device_handler_command_generation()
|
|
# print()
|
|
handle_returnvalue_generation()
|
|
print()
|
|
handle_objects_generation()
|
|
print()
|
|
handle_events_generation()
|
|
print()
|
|
|
|
|
|
def handle_subservices_generation():
|
|
print("MIB Exporter: Parsing subservices")
|
|
subservice_table = generate_subservice_table()
|
|
print("MIB Exporter: Found " + str(len(subservice_table)) + " subservice entries.")
|
|
if PRINT_TABLES_TO_CONSOLE:
|
|
print("MIB Exporter: Printing subservice table: ")
|
|
Printer.print_content(subservice_table)
|
|
if EXPORT_TO_CSV:
|
|
subservice_writer = CsvWriter(
|
|
SUBSERVICE_CSV_NAME, subservice_table, SUBSERVICE_COLUMN_HEADER
|
|
)
|
|
print("MIB Exporter: Exporting to file: " + SUBSERVICE_CSV_NAME)
|
|
subservice_writer.write_to_csv()
|
|
if EXPORT_TO_SQL:
|
|
print("MIB Exporter: Exporting subservices to SQL")
|
|
sql_writer = SqlWriter()
|
|
sql_writer.delete(SQL_DELETE_SUBSVC_CMD)
|
|
sql_writer.sql_writing_helper(
|
|
SQL_CREATE_SUBSVC_CMD, SQL_INSERT_INTO_SUBSVC_CMD, subservice_table
|
|
)
|
|
|
|
|
|
def generate_subservice_table():
|
|
"""Generate the subservice table."""
|
|
subservice_header_parser = FileListParser(
|
|
destination_corrected(SUBSERVICE_DEFINITION_DESTINATION)
|
|
)
|
|
subservice_header_list = subservice_header_parser.parse_header_files(
|
|
False, "MIB Exporter: Parsing subservice files: "
|
|
)
|
|
subservice_file_parser = SubserviceParser(subservice_header_list)
|
|
subservice_table = subservice_file_parser.parse_files()
|
|
return subservice_table
|
|
|
|
|
|
def handle_packet_content_generation():
|
|
print("MIB Exporter: Parsing packing content")
|
|
packet_content_table = generate_packet_content_table()
|
|
print(
|
|
"MIB Exporter: Found "
|
|
+ str(len(packet_content_table))
|
|
+ " packet content entries."
|
|
)
|
|
if PRINT_TABLES_TO_CONSOLE:
|
|
print("MIB Exporter: Print packet content table: ")
|
|
Printer.print_content(packet_content_table)
|
|
if EXPORT_TO_CSV:
|
|
packet_content_writer = CsvWriter(
|
|
PACKET_CONTENT_CSV_NAME, packet_content_table, PACKET_CONTENT_HEADER_COLUMN
|
|
)
|
|
print("MIB Exporter: Exporting to file " + PACKET_CONTENT_CSV_NAME)
|
|
packet_content_writer.write_to_csv()
|
|
if EXPORT_TO_SQL:
|
|
print("MIB Exporter: Exporting packet content to SQL")
|
|
sql_writer = SqlWriter()
|
|
|
|
sql_writer.sql_writing_helper(
|
|
SQL_CREATE_PACKET_DATA_CONTENT_CMD,
|
|
SQL_INSERT_PACKET_DATA_CMD,
|
|
packet_content_table,
|
|
SQL_DELETE_PACKET_DATA_CONTENT_CMD,
|
|
)
|
|
|
|
|
|
def generate_packet_content_table():
|
|
"""Generate packet content table"""
|
|
packet_data_header_parser = FileListParser(
|
|
destination_corrected(PACKET_CONTENT_DEFINITION_DESTINATION)
|
|
)
|
|
packet_data_header_list = packet_data_header_parser.parse_header_files(
|
|
False, "MIB Exporter: Parsing packet data files: "
|
|
)
|
|
packet_content_file_parser = PacketContentParser(packet_data_header_list)
|
|
packet_content_table = packet_content_file_parser.parse_files()
|
|
return packet_content_table
|
|
|
|
|
|
def handle_device_handler_command_generation():
|
|
print("MIB Exporter: Parsing device handler commands.")
|
|
dh_command_table = generate_device_command_table()
|
|
print(
|
|
"MIB Exporter: Found "
|
|
+ str(len(dh_command_table))
|
|
+ " device handler command entries"
|
|
)
|
|
if PRINT_TABLES_TO_CONSOLE:
|
|
print("MIB Exporter: Printing device handler command table: ")
|
|
Printer.print_content(dh_command_table)
|
|
if EXPORT_TO_CSV:
|
|
device_command_writer = CsvWriter(
|
|
DH_COMMANDS_CSV_NAME, dh_command_table, DH_COMMAND_HEADER_COLUMNS
|
|
)
|
|
print(
|
|
"MIB Exporter: Exporting device handler commands to " + DH_COMMANDS_CSV_NAME
|
|
)
|
|
device_command_writer.write_to_csv()
|
|
if EXPORT_TO_SQL:
|
|
print("MIB Exporter: Exporting device handler commands to SQL")
|
|
sql_writer = SqlWriter()
|
|
sql_writer.sql_writing_helper(
|
|
SQL_CREATE_CMDTABLE_CMD,
|
|
SQL_INSERT_INTO_CMDTABLE_CMD,
|
|
dh_command_table,
|
|
SQL_DELETE_CMDTABLE_CMD,
|
|
)
|
|
|
|
|
|
def generate_device_command_table(print_info_table: bool = False):
|
|
"""Generate device command table"""
|
|
info_header_file_parser = FileListParser(
|
|
destination_corrected(DH_DEFINITION_DESTINATION)
|
|
)
|
|
info_header_file_list = info_header_file_parser.parse_header_files(
|
|
False, "MIB Exporter: Parsing device handler informations: "
|
|
)
|
|
dh_information_parser = DeviceHandlerInformationParser(info_header_file_list)
|
|
dh_information_table = dh_information_parser.parse_files()
|
|
print(
|
|
"MIB Exporter: Found "
|
|
+ str(len(dh_information_table))
|
|
+ " device handler information entries."
|
|
)
|
|
if print_info_table:
|
|
Printer.print_content(
|
|
dh_information_table,
|
|
"MIB Exporter: Priting device handler command information table: ",
|
|
)
|
|
|
|
header_file_parser = FileListParser(
|
|
destination_corrected(DH_COMMAND_PACKET_DEFINITION_DESTINATION)
|
|
)
|
|
header_file_list = header_file_parser.parse_header_files(
|
|
False, "MIB Exporter: Parsing device handler command files: "
|
|
)
|
|
packet_subservice_parser = DeviceHandlerCommandParser(
|
|
header_file_list, dh_information_table
|
|
)
|
|
dh_command_table = packet_subservice_parser.parse_files()
|
|
return dh_command_table
|
|
|
|
|
|
def handle_returnvalue_generation():
|
|
print("MIB Exporter: Parsing returnvalues")
|
|
returnvalue_table = generate_returnvalue_table()
|
|
print("MIB Exporter: Found " + str(len(returnvalue_table)) + " returnvalues.")
|
|
if PRINT_TABLES_TO_CONSOLE:
|
|
print("MIB Exporter: Printing returnvalue table: ")
|
|
Printer.print_content(returnvalue_table)
|
|
if EXPORT_TO_CSV:
|
|
print("MIB Exporter: Exporting returnvalues to " + CSV_RETVAL_FILENAME)
|
|
ReturnValueParser.export_to_file(CSV_RETVAL_FILENAME, returnvalue_table)
|
|
if EXPORT_TO_SQL:
|
|
print("MIB Exporter: Export returnvalues to SQL: ")
|
|
sql_retval_exporter(returnvalue_table)
|
|
|
|
|
|
def generate_returnvalue_table():
|
|
interface_parser = InterfaceParser(
|
|
destination_corrected(INTERFACE_DEFINITION_FILES), False
|
|
)
|
|
interfaces = interface_parser.parse_files()
|
|
print("MIB Exporter: Found interfaces : " + str(len(interfaces)))
|
|
header_parser = FileListParser(destination_corrected(RETURNVALUE_DESTINATIONS))
|
|
header_list = header_parser.parse_header_files(
|
|
True, "MIB Exporter: Parsing header file list: "
|
|
)
|
|
returnvalue_parser = ReturnValueParser(interfaces, header_list, False)
|
|
returnvalue_table = returnvalue_parser.parse_files(False)
|
|
if PRINT_TABLES_TO_CONSOLE:
|
|
Printer.print_content(returnvalue_table, "Returnvalue Table: ")
|
|
return returnvalue_table
|
|
|
|
|
|
def handle_objects_generation():
|
|
print("MIB Exporter: Parsing Objects")
|
|
object_parser = ObjectDefinitionParser(destination_corrected(OBJECTS_DEFINITIONS))
|
|
object_table = object_parser.parse_files()
|
|
object_list_sorted = sorted(object_table.items())
|
|
print("MIB Exporter: Found " + str(len(object_table)) + " entries")
|
|
if EXPORT_TO_CSV:
|
|
print("MIB Exporter: Exporting to file: " + CSV_OBJECT_FILENAME)
|
|
export_object_file(CSV_OBJECT_FILENAME, object_list_sorted)
|
|
if EXPORT_TO_SQL:
|
|
print("MIB Exporter: Exporting objects into SQL table")
|
|
sql_object_exporter(object_list_sorted)
|
|
|
|
|
|
def handle_events_generation():
|
|
pass
|
|
|
|
|
|
def destination_corrected(destination_string):
|
|
"""
|
|
If headers are parsed here instead of the respective subdirectories,
|
|
the destination files are located in a different relative destination
|
|
"""
|
|
if isinstance(destination_string, list):
|
|
destination_list = []
|
|
for destination in destination_string:
|
|
destination_list.append(destination[3:])
|
|
return destination_list
|
|
|
|
return destination_string[3:]
|
|
|
|
|
|
def handle_external_file_running():
|
|
"""
|
|
Generates the MIB parser from external files
|
|
TODO: Make this stuff OOP too. Retvals and objects were already refactored
|
|
"""
|
|
os.chdir("events")
|
|
os.system("python event_parser.py")
|
|
os.chdir("..")
|
|
print_string = "Exported to file: MIB_Events.csv\r\n"
|
|
return print_string
|
|
|
|
|
|
def update_globals():
|
|
"""Updates the global variables"""
|
|
g.PP = pprint.PrettyPrinter(indent=0, width=250)
|
|
g.doExportMIB = DO_EXPORT_MIB
|
|
g.executeSQLcommands = False
|
|
g.printToConsole = PRINT_TABLES_TO_CONSOLE
|
|
g.exportToCSV = EXPORT_TO_CSV
|
|
g.copyFile = COPY_FILE
|
|
g.copyDestination = COPY_DESTINATION
|
|
g.fileSeparator = FILE_SEPARATOR
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|