175 lines
7.2 KiB
Python
175 lines
7.2 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, RAW_TM_PATH
|
|
|
|
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()
|
|
db_con = sqlite3.connect(TM_DB_PATH)
|
|
raw_tm_db_con = sqlite3.connect(RAW_TM_PATH)
|
|
raw_tm_db_con.execute("INSERT ")
|
|
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
|
|
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 _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=',')}"
|
|
)
|