eive-tmtc/pus_tm/event_handler.py

import os.path

from config.object_ids import get_object_ids

from tmtccmd.logging import get_console_logger
from tmtccmd.fsfw import parse_fsfw_events_csv, EventDictT, EventInfo


LOGGER = get_console_logger()
DEFAULT_EVENTS_CSV_PATH = "config/events.csv"
__EVENT_DICT = None


def get_event_dict() -> EventDictT:
    global __EVENT_DICT
    if __EVENT_DICT is None:
        if os.path.exists(DEFAULT_EVENTS_CSV_PATH):
            __EVENT_DICT = parse_fsfw_events_csv(DEFAULT_EVENTS_CSV_PATH)
        else:
            LOGGER.warning(f"No Event CSV file found at {DEFAULT_EVENTS_CSV_PATH}")
            __EVENT_DICT = dict()
    return __EVENT_DICT


def handle_event_packet(
    object_id: bytes, event_id: int, param_1: int, param_2: int
) -> str:
    additional_event_info = ""
    event_dict = get_event_dict()
    info = event_dict.get(event_id)
    if info is None:
        LOGGER.warning(f"Event ID {event_id} has no information")
        info = EventInfo()
        info.name = "Unknown event"
    obj_ids = get_object_ids()
    obj_id_obj = obj_ids.get(bytes(object_id))
    if obj_id_obj is None:
        LOGGER.warning(f"Object ID 0x{object_id.hex()} has no name")
        obj_name = object_id.hex()
    else:
        obj_name = obj_id_obj.name
    generic_event_string = f"Object {obj_name} generated Event {event_id} | {info.name}"
    if info.info != "":
        additional_event_info = (
            f" | Additional info: {info.info} | P1: {param_1} | P2: {param_2}"
        )
    return generic_event_string + additional_event_info