eive-tmtc/eive_tmtc/pus_tm/event_handler.py

import logging
import datetime
import sys
import base64

from eive_tmtc.config.definitions import PRINT_RAW_EVENTS_B64_STR
from eive_tmtc.config.events import get_event_dict
from eive_tmtc.config.object_ids import get_object_ids
from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.pus_tm.verification_handler import generic_retval_printout
from eive_tmtc.tmtc.acs.subsystem import AcsMode
from eive_tmtc.tmtc.core import SdState, SdCardSelect
from tmtccmd.pus.s200_fsfw_mode import Mode
from tmtccmd.pus.s201_fsfw_health import FsfwHealth

from tmtccmd.pus.s5_fsfw_event import Service5Tm
from tmtccmd.fsfw import EventInfo
from spacepackets.ccsds.time import CdsShortTimestamp

_LOGGER = logging.getLogger(__name__)


def handle_event_packet(  # noqa C901: Complexity okay here
    raw_tm: bytes, pw: PrintWrapper
):  # noqa C901: Complexity okay here
    if PRINT_RAW_EVENTS_B64_STR:
        print(f"PUS Event TM Base64: {base64.b64encode(raw_tm)}")
    tm = Service5Tm.unpack(data=raw_tm, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE)
    event_dict = get_event_dict()
    event_def = tm.event_definition
    info = event_dict.get(event_def.event_id)
    if info is None:
        _LOGGER.warning(f"Event ID {event_def.event_id} has no information")
        info = EventInfo()
        info.name = "Unknown event"
    obj_ids = get_object_ids()
    obj_id_obj = obj_ids.get(event_def.reporter_id)
    if obj_id_obj is None:
        _LOGGER.warning(f"Object ID 0x{event_def.reporter_id.hex(sep=',')} has no name")
        obj_name = event_def.reporter_id.hex(sep=",")
    else:
        obj_name = obj_id_obj.name
    timestamp = CdsShortTimestamp.unpack(tm.timestamp)
    generic_event_string = (
        f"Object {obj_name} generated Event {info.name} (ID: {event_def.event_id:#04x})"
        f" at {timestamp.as_date_time()}"
    )
    _LOGGER.info(generic_event_string)
    pw.file_logger.info(
        f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}:"
        f" {generic_event_string}"
    )
    specific_handler = False
    if info.name == "MODE_TRANSITION_FAILED":
        reason = generic_retval_printout(event_def.param1)
        for string in reason:
            pw.dlog(f"Reason from event parameter 1: {string}")
            pw.dlog(f"Mode, sequence or table: {event_def.param2:#08x}")
        specific_handler = True
    if info.name == "SUPV_UPDATE_PROGRESS" or info.name == "WRITE_MEMORY_FAILED":
        additional_event_info = f"Additional info: {info.info}"
        context = (
            f"Progress Percent: {event_def.param1 >> 24 & 0xff} | Sequence Count:"
            f" {event_def.param1 & 0xffff} | Bytes Written: {event_def.param2}"
        )
        pw.dlog(additional_event_info)
        pw.dlog(context)
        specific_handler = True
    if info.name == "MODE_INFO":
        specific_handler = True
        mode_name = "Unknown"
        if obj_name == "ACS_SUBSYSTEM":
            acs_mode = AcsMode(event_def.param1)
            pw.dlog(f"ACS Mode: {acs_mode!r}")
        elif obj_name == "ACS_CONTROLLER_ID":
            mode_name = Mode(event_def.param1)
            submode = AcsMode(event_def.param2)
            pw.dlog(f"Mode: {mode_name!r}")
            pw.dlog(f"ACS Mode: {submode!r}")
        else:
            if event_def.param1 == Mode.OFF:
                mode_name = "Off"
            elif event_def.param1 == Mode.ON:
                mode_name = "On"
            elif event_def.param1 == Mode.NORMAL:
                mode_name = "Normal"
            elif event_def.param1 == Mode.RAW:
                mode_name = "Raw"
            pw.dlog(
                f"Mode Number {event_def.param1}, Mode Name {mode_name}, "
                f"Submode: {event_def.param2}"
            )
    if info.name == "INDIVIDUAL_BOOT_COUNTS":
        boot_count_00 = (event_def.param1 >> 16) & 0xFFFF
        boot_count_01 = event_def.param1 & 0xFFFF
        boot_count_10 = (event_def.param2 >> 16) & 0xFFFF
        boot_count_11 = event_def.param2 & 0xFFFF
        pw.dlog(f"Boot count 0 0: {boot_count_00}")
        pw.dlog(f"Boot count 0 1: {boot_count_01}")
        pw.dlog(f"Boot count 1 0: {boot_count_10}")
        pw.dlog(f"Boot count 1 1: {boot_count_11}")
    if info.name == "REBOOT_COUNTER":
        boot_count = (event_def.param1 << 32) | event_def.param2
        pw.dlog(f"Total boot count: {boot_count}")
    if info.name == "VERSION_INFO" or info.name == "FIRMWARE_INFO":
        specific_handler = True
        ver_major = (event_def.param1 >> 24) & 0xFF
        ver_minor = (event_def.param1 >> 16) & 0xFF
        ver_rev = (event_def.param1 >> 8) & 0xFF
        has_git_sha = bool(event_def.param1 & 0xFF)
        git_sha = ""
        if has_git_sha:
            p2_as_bytes = event_def.param2.to_bytes(4, sys.byteorder)
            git_sha = p2_as_bytes.decode("ascii")
        if info.name == "VERSION_INFO":
            name = "OBSW version: "
        else:
            name = "Firmware version: "
        pw.dlog(f"{name} v{ver_major}.{ver_minor}.{ver_rev}")
        if has_git_sha:
            pw.dlog(f"Git SHA first four letters: {git_sha}")
    if info.name == "CLOCK_SET":
        specific_handler = True
        old_time = event_def.param1
        new_time = event_def.param2
        old_time_dt = datetime.datetime.fromtimestamp(old_time, datetime.timezone.utc)
        new_time_dt = datetime.datetime.fromtimestamp(new_time, datetime.timezone.utc)
        pw.dlog(f"Old time (UTC): {old_time_dt}")
        pw.dlog(f"New time (UTC): {new_time_dt}")
    if info.name == "CLOCK_DUMP_LEGACY":
        specific_handler = True
        # param 1 is timeval seconds, param 2 is timeval subsecond milliseconds
        time = event_def.param1 + event_def.param2 / 1000.0
        time_dt = datetime.datetime.fromtimestamp(time, datetime.timezone.utc)
        pw.dlog(f"Current time: {time_dt}")
    if (
        info.name == "CLOCK_DUMP"
        or info.name == "CLOCK_DUMP_BEFORE_SETTING_TIME"
        or info.name == "CLOCK_DUMP_AFTER_SETTING_TIME"
    ):
        specific_handler = True
        # param 1 is timeval seconds, param 2 is timeval subsecond microseconds
        time = event_def.param1 + event_def.param2 / 1000000.0
        time_dt = datetime.datetime.fromtimestamp(time, datetime.timezone.utc)
        pw.dlog(f"Clock dump event {info.name}. Current time: {time_dt}")
    if info.name == "ACTIVE_SD_INFO":
        sd_0_state = (event_def.param2 >> 16) & 0xFFFF
        sd_1_state = event_def.param2 & 0xFFFF
        active_sd = event_def.param1
        try:
            active_sd = SdCardSelect(event_def.param1)
            sd_0_state = SdState((event_def.param2 >> 16) & 0xFFFF)
            sd_1_state = SdState(event_def.param2 & 0xFFFF)
        except IndexError:
            _LOGGER.error(f"Received invalid event fields for event {event_def}")
        finally:
            pw.dlog(
                f"Active SD card {active_sd!r} | SD 0 State {sd_0_state!r} | SD 1 "
                f"State {sd_1_state!r}"
            )
    if info.name == "HEALTH_INFO":
        specific_handler = True
        health = FsfwHealth(event_def.param1)
        pw.dlog(f"{obj_name}: {health!r}")
    if info.name == "CHANGING_MODE":
        mode = event_def.param1
        submode = event_def.param2
        pw.dlog(f"Mode Number {mode}, Submode: {submode}")
    if not specific_handler:
        additional_event_info = (
            f"Additional info: {info.info} | P1: {event_def.param1} | "
            f"P2: {event_def.param2}"
        )
        pw.dlog(additional_event_info)
    if not specific_handler:
        # printer.handle_long_tm_print(packet_if=tm.pus_tm, info_if=tm.pus_tm)
        pass