let's try a DB
All checks were successful
EIVE/-/pipeline/head This commit looks good

This commit is contained in:
Robin Müller 2023-11-13 08:50:51 +01:00
parent 99c6c8bbd0
commit 3fe51b08a6
Signed by: muellerr
GPG Key ID: A649FB78196E3849
5 changed files with 92 additions and 37 deletions

View File

@ -0,0 +1 @@
from .definitions import * # noqa

View File

@ -11,6 +11,7 @@ from spacepackets.ccsds import PacketId
from spacepackets.util import UnsignedByteField
DB_PATH = "tm.db"
PUS_APID = 0x65
CFDP_APID = 0x66
PUS_PACKET_ID = PacketId(PacketType.TM, True, PUS_APID)

View File

@ -1,7 +1,8 @@
"""HK Handling for EIVE OBSW"""
import dataclasses
import logging
from typing import List
import sqlite3
from typing import List, Optional, cast
from eive_tmtc.tmtc.acs.acs_ctrl import handle_acs_ctrl_hk_data
from eive_tmtc.tmtc.internal_err_reporter import handle_ier_hk_data
@ -18,7 +19,6 @@ from eive_tmtc.tmtc.tcs import handle_thermal_controller_hk_data
from eive_tmtc.tmtc.tcs.tmp1075 import handle_tmp_1075_hk_data
from spacepackets.ecss import PusTelemetry
from tmtccmd.tm.pus_3_fsfw_hk import (
Service3Base,
HkContentType,
Service3FsfwTm,
)
@ -60,9 +60,10 @@ def handle_hk_packet(
printer: FsfwTmTcPrinter,
hk_filter: HkFilter,
hk_level: int,
db: Optional[sqlite3.Connection],
):
tm_packet = Service3FsfwTm.unpack(raw_telemetry=raw_tm, custom_hk_handling=False)
named_obj_id = obj_id_dict.get(tm_packet.object_id.as_bytes)
named_obj_id = cast(ObjectIdU32, obj_id_dict.get(tm_packet.object_id.as_bytes))
if named_obj_id is None:
named_obj_id = tm_packet.object_id
if tm_packet.subservice == 25 or tm_packet.subservice == 26:
@ -75,6 +76,7 @@ def handle_hk_packet(
hk_packet=tm_packet,
tm=tm_packet.pus_tm,
hk_data=hk_data,
db=db,
)
return
try:
@ -89,6 +91,7 @@ def handle_hk_packet(
hk_packet=tm_packet,
tm=tm_packet.pus_tm,
hk_data=hk_data,
db=db,
)
except ValueError as e:
_LOGGER.exception(
@ -101,12 +104,14 @@ def handle_hk_packet(
def handle_regular_hk_print( # noqa C901: Complexity okay here
printer: FsfwTmTcPrinter,
object_id: ObjectIdU32,
hk_packet: Service3Base,
hk_packet: Service3FsfwTm,
tm: PusTelemetry,
hk_data: bytes,
db: Optional[sqlite3.Connection],
):
objb = object_id.as_bytes
set_id = hk_packet.set_id
assert tm.time_provider is not None
packet_dt = tm.time_provider.as_date_time()
pw = PrintWrapper(printer.file_logger)
"""This function is called when a Service 3 Housekeeping packet is received."""
@ -127,9 +132,23 @@ def handle_regular_hk_print( # noqa C901: Complexity okay here
elif objb == obj_ids.CORE_CONTROLLER_ID:
return handle_core_hk_data(pw=pw, hk_data=hk_data, set_id=set_id)
elif objb == obj_ids.PDU_1_HANDLER_ID:
return handle_pdu_data(pw=pw, pdu_idx=1, set_id=set_id, hk_data=hk_data)
return handle_pdu_data(
hk_packet=hk_packet,
con=db,
pw=pw,
pdu_idx=1,
set_id=set_id,
hk_data=hk_data,
)
elif objb == obj_ids.PDU_2_HANDLER_ID:
return handle_pdu_data(pw=pw, pdu_idx=2, set_id=set_id, hk_data=hk_data)
return handle_pdu_data(
hk_packet=hk_packet,
con=db,
pw=pw,
pdu_idx=2,
set_id=set_id,
hk_data=hk_data,
)
elif objb == obj_ids.PLOC_MPSOC_ID:
return handle_ploc_mpsoc_hk_data(pw=pw, hk_data=hk_data, set_id=set_id)
elif objb == obj_ids.ACU_HANDLER_ID:

View File

@ -1,5 +1,6 @@
import struct
from typing import List, Tuple
import sqlite3
from typing import List, Optional, Tuple
from eive_tmtc.tmtc.power.acu import acu_config_table_handler
from eive_tmtc.tmtc.power.common_power import (
@ -8,6 +9,7 @@ from eive_tmtc.tmtc.power.common_power import (
OBC_ENDIANNESS,
)
from eive_tmtc.tmtc.power.power import PcduSetIds
from tmtccmd.tm.pus_3_fsfw_hk import Service3FsfwTm
from tmtccmd.util import ObjectIdBase
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
from eive_tmtc.pus_tm.defs import PrintWrapper
@ -146,7 +148,14 @@ class DevicesInfoParser:
return "Unknown Type"
def handle_pdu_data(pw: PrintWrapper, pdu_idx: int, set_id: int, hk_data: bytes):
def handle_pdu_data(
hk_packet: Service3FsfwTm,
con: Optional[sqlite3.Connection],
pw: PrintWrapper,
pdu_idx: int,
set_id: int,
hk_data: bytes,
):
current_idx = 0
priv_idx = pdu_idx - 1
if set_id == SetId.AUX or set_id == SetId.AUX:
@ -219,6 +228,28 @@ def handle_pdu_data(pw: PrintWrapper, pdu_idx: int, set_id: int, hk_data: bytes)
f"Boot Count {boot_count} | Battery Mode {batt_mode} | "
f"Temperature {temperature} | VCC {vcc} | VBAT {vbat}"
)
if con is not None:
cursor = con.cursor()
if pdu_idx == 1:
tbl_name = "Pdu1"
channel_list = PDU1_CHANNELS_NAMES
else:
tbl_name = "Pdu2"
channel_list = PDU2_CHANNELS_NAMES
for idx, name in enumerate(channel_list):
cursor.execute(
f"CREATE TABLE {tbl_name}{name} IF NOT EXISTS "
f"(GenerationTime, OutEnable, Voltage, Current)"
)
value_list = [
hk_packet.pus_tm.time_provider.as_datetime(), # type: ignore
output_enb_list[idx],
voltage_list[idx],
current_list[idx],
]
cursor.execute(
f"INSERT INTO {tbl_name}{name} VALUES(?, ?, ?, ?)", value_list
)
pw.dlog(info)

View File

@ -1,60 +1,62 @@
#!/usr/bin/env python3
import sqlite3
import logging
import sys
import time
from importlib.metadata import version
from pathlib import Path
from typing import Tuple
import tmtccmd
from eive_tmtc.cfdp.fault_handler import EiveCfdpFaultHandler
from eive_tmtc.cfdp.tm import CfdpInCcsdsWrapper
from eive_tmtc.cfdp.user import EiveCfdpUser
from spacepackets.ecss import PusVerificator
from spacepackets.version import get_version as get_sp_version
from spacepackets.cfdp import (
ChecksumType,
TransmissionMode,
)
from eive_tmtc.pus_tc.tc_handler import TcHandler
from eive_tmtc.pus_tm.hk_handler import HkFilter
from tmtccmd.logging import add_colorlog_console_logger
from spacepackets.ecss import PusVerificator
from spacepackets.version import get_version as get_sp_version
from tmtccmd import BackendBase
from tmtccmd.cfdp.handler import CfdpInCcsdsHandler
from tmtccmd.cfdp.mib import (
LocalEntityCfg,
IndicationCfg,
LocalEntityCfg,
RemoteEntityCfg,
)
from tmtccmd import BackendBase
from tmtccmd.util import FileSeqCountProvider, PusFileSeqCountProvider
from tmtccmd.config import (
SetupWrapper,
default_json_path,
params_to_procedure_conversion,
)
from tmtccmd.config.args import (
PreArgsParsingWrapper,
ProcedureParamsWrapper,
SetupParams,
)
from tmtccmd.core import BackendRequest
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.logging import add_colorlog_console_logger
from tmtccmd.logging.pus import (
RawTmtcTimedLogWrapper,
RegularTmtcLogWrapper,
TimedLogWhen,
)
from tmtccmd.pus import VerificationWrapper
from tmtccmd.tm import SpecificApidHandlerBase, GenericApidHandlerBase, CcsdsTmHandler
from tmtccmd.core import BackendRequest
from tmtccmd.config import (
default_json_path,
SetupWrapper,
params_to_procedure_conversion,
)
from tmtccmd.config.args import (
SetupParams,
PreArgsParsingWrapper,
ProcedureParamsWrapper,
)
from tmtccmd.tm import CcsdsTmHandler, GenericApidHandlerBase, SpecificApidHandlerBase
from tmtccmd.util import FileSeqCountProvider, PusFileSeqCountProvider
from eive_tmtc import APP_LOGGER
from importlib.metadata import version
from eive_tmtc.cfdp.fault_handler import EiveCfdpFaultHandler
from eive_tmtc.cfdp.tm import CfdpInCcsdsWrapper
from eive_tmtc.cfdp.user import EiveCfdpUser
from eive_tmtc.config.definitions import (
PUS_APID,
CFDP_APID,
CFDP_LOCAL_ENTITY_ID,
CFDP_REMOTE_ENTITY_ID,
PUS_APID,
DB_PATH,
)
from eive_tmtc.config.hook import EiveHookObject
from eive_tmtc.pus_tc.tc_handler import TcHandler
from eive_tmtc.pus_tm.hk_handler import HkFilter
from eive_tmtc.pus_tm.pus_demux import pus_factory_hook
_LOGGER = APP_LOGGER
@ -80,6 +82,7 @@ class PusHandler(SpecificApidHandlerBase):
self.hk_level = hk_level
self.these_objs_hk_only = []
self.hk_filter = HkFilter(object_ids=self.these_objs_hk_only, set_ids=[])
self.tm_db = sqlite3.connect(DB_PATH)
def handle_tm(self, packet: bytes, _user_args: any):
# with open("tc.bin", "wb") as of:
@ -106,7 +109,7 @@ class CustomCcsdsTmHandler(CcsdsTmHandler):
_LOGGER.debug(f"Received packet {packet.hex(sep=',')} with APID {apid}")
def setup_params() -> (SetupWrapper, int):
def setup_params() -> Tuple[SetupWrapper, int]:
hook_obj = EiveHookObject(default_json_path())
params = SetupParams()
parser_wrapper = PreArgsParsingWrapper()
@ -190,7 +193,7 @@ def setup_tmtc_handlers(
raw_logger: RawTmtcTimedLogWrapper,
gui: bool,
hk_level: int,
) -> (CcsdsTmHandler, TcHandler):
) -> Tuple[CcsdsTmHandler, TcHandler]:
cfdp_in_ccsds_wrapper = setup_cfdp_handler()
verification_wrapper = VerificationWrapper(
verificator, _LOGGER, printer.file_logger