eive-tmtc/eive_tmtc/pus_tm/pus_handler.py
Robin Mueller 2259d269dd
All checks were successful
EIVE/-/pipeline/pr-main This commit looks good
separate function for raw TM insertion
2023-11-13 13:19:42 +01:00

207 lines
8.1 KiB
Python

"""Core EIVE TM handler module
"""
import logging
import sqlite3
import uuid
from typing import Any
from spacepackets.ccsds.time import CdsShortTimestamp
from spacepackets.ecss import PusTelemetry
from spacepackets.ecss.pus_17_test import Service17Tm
from spacepackets.util import PrintFormats
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.logging.pus import RawTmtcTimedLogWrapper
from tmtccmd.pus import VerificationWrapper
from tmtccmd.pus.s20_fsfw_param import Service20FsfwTm, Service20ParamDumpWrapper
from tmtccmd.pus.s20_fsfw_param_defs import CustomSubservice as ParamSubservice
from tmtccmd.pus.s200_fsfw_mode import Service200FsfwTm
from tmtccmd.pus.s200_fsfw_mode import Subservice as ModeSubservice
from tmtccmd.tmtc import GenericApidHandlerBase, SpecificApidHandlerBase
from eive_tmtc.config.definitions import TM_DB_PATH, PUS_APID
from eive_tmtc.config.object_ids import get_object_ids
from .action_reply_handler import handle_action_reply
from .defs import PrintWrapper
from .event_handler import handle_event_packet
from .hk_handler import HkFilter, handle_hk_packet
from .verification_handler import generic_retval_printout, handle_service_1_fsfw_packet
_LOGGER = logging.getLogger(__name__)
class PusHandler(SpecificApidHandlerBase):
def __init__(
self,
wrapper: VerificationWrapper,
printer: FsfwTmTcPrinter,
raw_logger: RawTmtcTimedLogWrapper,
hk_level: int,
):
super().__init__(PUS_APID, None)
self.printer = printer
self.verif_wrapper = wrapper
self.raw_logger = raw_logger
self.hk_level = hk_level
self.these_objs_hk_only = []
self.hk_filter = HkFilter(object_ids=self.these_objs_hk_only, set_ids=[])
self.obj_id_dict = get_object_ids()
self.pw = PrintWrapper(printer.file_logger)
def handle_tm(self, packet: bytes, _user_args: Any):
self.pus_handler(
packet,
)
def pus_handler( # noqa C901 : Complexity okay here
self,
packet: bytes,
):
packet_uuid = uuid.uuid4()
if len(packet) < 8:
_LOGGER.warning("Detected packet shorter than 8 bytes!")
return
try:
tm_packet = PusTelemetry.unpack(packet, CdsShortTimestamp.empty())
# _LOGGER.info(f"Sequence count: {tm_packet.seq_count}")
except ValueError as value_error:
_LOGGER.warning(f"{value_error}")
_LOGGER.warning("Could not generate PUS TM object from raw data")
_LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}")
return
db_con = sqlite3.connect(TM_DB_PATH)
self._store_packet_in_db(db_con, packet, tm_packet, packet_uuid)
service = tm_packet.service
dedicated_handler = True
if service == 1:
handle_service_1_fsfw_packet(wrapper=self.verif_wrapper, raw_tm=packet)
elif service == 3:
handle_hk_packet(
db_con=db_con,
packet_uuid=packet_uuid,
printer=self.printer,
raw_tm=packet,
obj_id_dict=self.obj_id_dict,
hk_level=self.hk_level,
hk_filter=self.hk_filter,
)
elif service == 5:
handle_event_packet(raw_tm=packet, pw=self.pw)
elif service == 8:
handle_action_reply(
raw_tm=packet, printer=self.printer, obj_id_dict=self.obj_id_dict
)
elif service == 17:
pus17_tm = Service17Tm.unpack(
data=packet, time_reader=CdsShortTimestamp.empty()
)
if pus17_tm.subservice == 2:
self.verif_wrapper.dlog("Received Ping Reply TM[17,2]")
dedicated_handler = True
elif service == 20:
self._handle_param_packet(packet, tm_packet)
elif service == 200:
dedicated_handler = self._handle_mode_packet(packet, tm_packet)
else:
_LOGGER.info(
f"The service {service} is not implemented in Telemetry Factory"
)
tm_packet.print_source_data(PrintFormats.HEX)
dedicated_handler = True
if not dedicated_handler and tm_packet is not None:
self.printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
self.raw_logger.log_tm(tm_packet)
def _store_packet_in_db(
self,
db_con: sqlite3.Connection,
packet: bytes,
tm_packet: PusTelemetry,
packet_uuid: uuid.UUID,
):
cursor = db_con.cursor()
assert tm_packet.time_provider is not None
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS pus_tm(
packet_uuid TEXT PRIMARY KEY,
generation_time TEXT,
service NUM,
subservice NUM,
data_len NUM,
raw_data BLOB
)"""
)
cursor.execute(
"INSERT INTO pus_tm VALUES(?, ?, ?, ?, ?, ?)",
(
str(packet_uuid),
tm_packet.time_provider.as_datetime(),
tm_packet.service,
tm_packet.subservice,
len(packet),
packet,
),
)
db_con.commit()
def _handle_param_packet(self, raw_data: bytes, tm_packet: PusTelemetry):
param_packet = Service20FsfwTm.unpack(
raw_telemetry=raw_data, time_reader=CdsShortTimestamp.empty()
)
if tm_packet.subservice == ParamSubservice.TM_DUMP_REPLY:
param_wrapper = Service20ParamDumpWrapper(param_tm=param_packet)
try:
param = param_wrapper.get_param()
obj = self.obj_id_dict.get(param_wrapper.param_tm.object_id)
self.pw.dlog(f"Received parameter dump TM from {obj}")
self.pw.dlog(f"Parameter: {param}")
if param.rows == 1 and param.columns == 1:
try:
scalar_param = param.parse_scalar_param()
if isinstance(scalar_param, int):
self.pw.dlog(f"Scalar integer parameter: {scalar_param}")
elif isinstance(scalar_param, float):
self.pw.dlog(
f"Scalar floating point parameter: {scalar_param}"
)
except ValueError as e:
self.pw.dlog(f"received {e} trying to parse scalar parameter")
else:
# TODO: Could improve display further by actually displaying a matrix as a
# matrix using row and column information
self.pw.dlog(
"Received vector or matrix data:"
f" {param.param_raw.hex(sep=',')}"
)
except ValueError as e:
self.pw.dlog(f"received {e} when trying to parse parameters")
except NotImplementedError as e:
self.pw.dlog(f"received {e} when trying to parse parameters")
else:
self.pw.dlog(
f"unknown subservice {tm_packet.subservice} for parameter service"
)
def _handle_mode_packet(self, raw_data: bytes, _: PusTelemetry) -> bool:
tm_packet = Service200FsfwTm.unpack(
raw_telemetry=raw_data, time_reader=CdsShortTimestamp.empty()
)
if tm_packet.subservice == ModeSubservice.TM_CANT_REACH_MODE:
obj_id = tm_packet.object_id
obj_id_obj = self.obj_id_dict.get(obj_id)
retval = tm_packet.return_value
string_list = generic_retval_printout(retval)
self.pw.dlog(f"Received Mode Reply from {obj_id_obj}: Can't reach mode.")
for string in string_list:
self.pw.dlog(f"Reason: {string}")
return True
return False
class UnknownApidHandler(GenericApidHandlerBase):
def handle_tm(self, apid: int, packet: bytes, _user_args: Any):
_LOGGER.warning(
f"Packet with unknown APID {apid} detected: {packet.hex(sep=',')}"
)