eive/eive-tmtc
eive
/
eive-tmtc
13
0
Fork 0
Code Issues Pull Requests 1 Projects Releases 97 Wiki Activity

finished p60 dock HK handling

This commit is contained in:
Robin Müller 2022-04-05 19:27:55 +02:00
parent 2c58a77338
commit a038e4c175
No known key found for this signature in database
GPG Key ID: 71B58F8A3CDFA9AC
4 changed files with 154 additions and 189 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
pus_tm/event_handler.py
View File

@ -5,6 +5,7 @@ from config.object_ids import get_object_ids
from tmtccmd.tm import Service5Tm
from tmtccmd.logging import get_console_logger
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.fsfw import parse_fsfw_events_csv, EventDictT, EventInfo
@ -24,7 +25,13 @@ def get_event_dict() -> EventDictT:
return __EVENT_DICT
def handle_event_packet(file_logger: logging.Logger, tm: Service5Tm) -> str:
def handle_event_packet(
raw_tm: bytes,
printer: FsfwTmTcPrinter,
file_logger: logging.Logger
) -> str:
tm = Service5Tm.unpack(raw_telemetry=raw_tm)
printer.handle_long_tm_print(packet_if=tm, info_if=tm)
additional_event_info = ""
event_dict = get_event_dict()
info = event_dict.get(tm.event_id)

23
pus_tm/factory_hook.py
View File

@ -43,36 +43,35 @@ def pus_factory_hook(raw_tm_packet: bytes):
return
service_type = raw_tm_packet[7]
subservice_type = raw_tm_packet[8]
tm_packet = None
file_logger = FSFW_PRINTER.file_logger
obj_id_dict = get_object_ids()
try:
if service_type == 1:
handle_service_1_packet(printer=FSFW_PRINTER, raw_tm=raw_tm_packet)
elif service_type == 3:
tm_packet = Service3FsfwTm.unpack(
raw_telemetry=raw_tm_packet, custom_hk_handling=False
)
named_obj_id = obj_id_dict.get(tm_packet.object_id.as_bytes)
if named_obj_id is None:
named_obj_id = tm_packet.object_id
handle_hk_packet(
printer=FSFW_PRINTER,
object_id=named_obj_id,
hk_packet=tm_packet,
packet_if=tm_packet
raw_tm=raw_tm_packet,
obj_id_dict=obj_id_dict
)
elif service_type == 5:
tm_packet = Service5Tm.unpack(raw_telemetry=raw_tm_packet)
handle_event_packet(file_logger=file_logger, tm=tm_packet)
handle_event_packet(
raw_tm=raw_tm_packet,
printer=FSFW_PRINTER,
file_logger=file_logger
)
elif service_type == 8:
tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm_packet)
FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
elif service_type == 17:
tm_packet = Service17TMExtended.unpack(raw_telemetry=raw_tm_packet)
FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
elif service_type == 20:
tm_packet = Service20FsfwTm.unpack(raw_telemetry=raw_tm_packet)
FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
elif service_type == 200:
tm_packet = Service200FsfwTm.unpack(raw_telemetry=raw_tm_packet)
FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
else:
LOGGER.info(
f"The service {service_type} is not implemented in Telemetry Factory"

309
pus_tm/hk_handling.py
View File

@ -5,13 +5,12 @@ import datetime
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.config.definitions import HkReplyUnpacked
from tmtccmd.tm.base import PusTmInterface
from tmtccmd.tm.service_3_fsfw_housekeeping import Service3Base, HkContentType
from tmtccmd.tm.service_3_fsfw_housekeeping import Service3Base, HkContentType, Service3FsfwTm
from tmtccmd.logging import get_console_logger
from pus_tc.devs.bpx_batt import BpxSetIds
from pus_tc.devs.syrlinks_hk_handler import SetIds
from pus_tc.devs.imtq import ImtqSetIds
from tmtccmd.pus.obj_id import ObjectId
from tmtccmd.pus.obj_id import ObjectId, ObjectIdDictT
from config.object_ids import (
SYRLINKS_HANDLER_ID,
IMTQ_HANDLER_ID,
@ -26,31 +25,39 @@ LOGGER = get_console_logger()
def handle_hk_packet(
printer: FsfwTmTcPrinter,
object_id: ObjectId,
hk_packet: Service3Base,
packet_if: PusTmInterface
raw_tm: bytes,
obj_id_dict: ObjectIdDictT,
printer: FsfwTmTcPrinter,
):
if packet_if.subservice == 25 or packet_if.subservice == 26:
hk_data = packet_if.tm_data[8:]
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)
if named_obj_id is None:
named_obj_id = tm_packet.object_id
if tm_packet.subservice == 25 or tm_packet.subservice == 26:
hk_data = tm_packet.tm_data[8:]
printer.generic_hk_print(
content_type=HkContentType.HK,
object_id=object_id,
set_id=hk_packet.set_id,
hk_data=hk_data
object_id=named_obj_id,
set_id=tm_packet.set_id,
hk_data=hk_data,
)
handle_regular_hk_print(
printer=printer,
object_id=object_id,
hk_packet=hk_packet,
object_id=named_obj_id,
hk_packet=tm_packet,
hk_data=hk_data,
)
if packet_if.subservice == 10 or packet_if.subservice == 12:
if tm_packet.subservice == 10 or tm_packet.subservice == 12:
LOGGER.warning("HK definitions printout not implemented yet")
def handle_regular_hk_print(
printer: FsfwTmTcPrinter, object_id: ObjectId, hk_packet: Service3Base, hk_data: bytes
printer: FsfwTmTcPrinter,
object_id: ObjectId,
hk_packet: Service3Base,
hk_data: bytes,
):
object_id = object_id.as_bytes
set_id = hk_packet.set_id
@ -74,7 +81,7 @@ def handle_regular_hk_print(
elif object_id == BPX_HANDLER_ID:
return handle_bpx_hk_data(hk_data=hk_data, set_id=set_id)
elif object_id == CORE_CONTROLLER_ID:
return handle_core_hk_data(hk_data=hk_data, set_id=set_id)
return handle_core_hk_data(printer=printer, hk_data=hk_data)
elif object_id == P60_DOCK_HANDLER:
return handle_p60_hk_data(printer=printer, hk_data=hk_data)
else:
@ -376,23 +383,26 @@ def handle_bpx_hk_data(hk_data: bytes, set_id: int) -> HkReplyUnpacked:
return reply
def handle_core_hk_data(hk_data: bytes, set_id: int) -> HkReplyUnpacked:
reply = HkReplyUnpacked()
reply.header_list = ["Chip Temperature [°C]", "PS Voltage [mV]", "PL Voltage [mV]"]
temperature = struct.unpack("!f", hk_data[0:4])
ps_voltage = struct.unpack("!f", hk_data[4:8])
pl_voltage = struct.unpack("!f", hk_data[8:12])
tx_agc_value = struct.unpack("!H", hk_data[2:4])
reply.content_list = [temperature, ps_voltage, pl_voltage]
reply.validity_buffer = hk_data[12:]
reply.num_of_vars = 3
return reply
def handle_core_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
fmt_str = "!fffH"
inc_len = struct.calcsize(fmt_str)
(
temperature,
ps_voltage,
pl_voltage,
tx_agc_value
) = struct.unpack(fmt_str, hk_data[0:0 + inc_len])
printout = f"Chip Temperature [°C] {temperature} | PS Voltage [mV] {ps_voltage} | " \
f"PL Voltage [mV] {pl_voltage} | TX AGC {tx_agc_value}"
log_to_both(printer, printout)
printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=4)
P60_INDEX_LIST = [
"ACU VCC",
"PDU1 VCC",
"X3 IDLE",
"X3 IDLE VCC",
"PDU2 VCC",
"ACU VBAT",
"PDU1 VBAT",
@ -405,14 +415,15 @@ P60_INDEX_LIST = [
"GS5V",
]
WDT_LIST = ["GND", "I2C", "CAN", "CSP0", "CSP1"]
def handle_p60_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes) -> HkReplyUnpacked:
reply = HkReplyUnpacked()
def handle_p60_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
current_idx = 0
current_list = []
for idx in range(0, 13):
for idx in range(13):
current_list.append(
struct.unpack('!h', hk_data[current_idx : current_idx + 2])[0]
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
voltage_list = []
@ -430,160 +441,108 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes) -> HkReplyUnpac
printer.file_logger.info(header_str)
for idx in range(13):
out_enb = f"{out_enb_list[idx]}".ljust(6)
content_line = f"{P60_INDEX_LIST[idx].ljust(24)} | {out_enb} | " \
f"{voltage_list[idx]:05} | {current_list[idx]:04}"
content_line = (
f"{P60_INDEX_LIST[idx].ljust(24)} | {out_enb} | "
f"{voltage_list[idx]:05} | {current_list[idx]:04}"
)
print(content_line)
printer.file_logger.info(content_line)
reply.header_list.append("Temp 0 [C]")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
fmt_str = "!hhIIIhBBB"
inc_len = struct.calcsize(fmt_str)
(
temp0,
temp1,
boot_cause,
boot_count,
uptime,
reset_cause,
batt_mode,
heater_on,
conv_5v_on,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len
util_info = (
f"Batt Mode {batt_mode} | Boot Count {boot_count} | Heater On {heater_on}"
)
current_idx += 2
reply.header_list.append("Temp 1 [C]")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
util_info2 = (
f"Reset Cause {reset_cause} | Boot Cause {boot_cause} | Uptime {uptime} | "
f"Conv 5V on {conv_5v_on}"
)
current_idx += 2
reply.header_list.append("Boot Cause")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("Boot Count")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("Uptime")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("Reset Cause")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Battery Mode")
reply.content_list.append(hk_data[current_idx])
current_idx += 1
reply.header_list.append("Heater On")
reply.content_list.append(hk_data[current_idx])
current_idx += 1
reply.header_list.append("Conv 5V Enable Status")
reply.content_list.append(hk_data[current_idx])
current_idx += 1
print(util_info)
print(util_info2)
printer.file_logger.info(util_info)
printer.file_logger.info(util_info2)
latchup_list = []
for idx in range(0, 13):
reply.header_list.append(f"Latchup {P60_INDEX_LIST[idx]}")
reply.content_list.append(
latchup_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Dock VBAT [mV]")
reply.content_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Dock VCC Current [mA]")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Batt Charge [mA]")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Batt Voltage [mV]")
reply.content_list.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Batt Temp 0 [C]")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
reply.header_list.append("Batt Temp 1 [C]")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
for idx in range(0, 7):
reply.header_list.append(f"Device {idx} type")
reply.content_list.append(hk_data[current_idx])
fmt_str = "!HhhHhh"
inc_len = struct.calcsize(fmt_str)
(
dock_vbat,
dock_vcc_current,
batt_current,
batt_voltage,
batt_temp_0,
batt_temp_1
) = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
current_idx += inc_len
device_types = []
device_statuses = []
for idx in range(8):
device_types.append(hk_data[current_idx])
current_idx += 1
for idx in range(0, 7):
reply.header_list.append(f"Device {idx} status")
reply.content_list.append(hk_data[current_idx])
for idx in range(8):
device_statuses.append(hk_data[current_idx])
current_idx += 1
reply.header_list.append("De-Arm Status")
reply.content_list.append(hk_data[current_idx])
dearm_status = hk_data[current_idx]
current_idx += 1
reply.header_list.append("GND WDT Reboots")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("I2C WDT Reboots")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("CAN WDT Reboots")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("CSP0 WDT Reboots")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("CSP1 WDT Reboots")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("GND WDT time left [s]")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("I2C WDT time left [s]")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("CAN WDT time left [s]")
reply.content_list.append(
struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
reply.header_list.append("CSP0 WDT Pings Left")
reply.content_list.append(hk_data[current_idx])
current_idx += 1
reply.header_list.append("CSP1 WDT Pings left")
reply.content_list.append(hk_data[current_idx])
current_idx += 1
reply.header_list.append("Batt Charge Current")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
wdt_reboots_list = []
for idx in range(5):
wdt_reboots_list.append(struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0])
current_idx += 4
time_pings_left_list = []
for idx in range(3):
time_pings_left_list.append(struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0])
current_idx += 4
for idx in range(2):
time_pings_left_list.append(hk_data[current_idx])
current_idx += 1
batt_charge_current = struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
current_idx += 2
reply.header_list.append("Batt Discharge Current")
reply.content_list.append(
struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
)
batt_discharge_current = struct.unpack(
"!h", hk_data[current_idx : current_idx + 2]
)[0]
current_idx += 2
reply.header_list.append("ANT6 Depl Status")
reply.content_list.append(
struct.unpack("!b", hk_data[current_idx : current_idx + 1])[0]
)
ant6_depl_status = struct.unpack("!b", hk_data[current_idx : current_idx + 1])[0]
current_idx += 1
reply.header_list.append("AR6 Depl Status")
reply.content_list.append(
struct.unpack("!b", hk_data[current_idx : current_idx + 1])[0]
)
ar6_depl_status = struct.unpack("!b", hk_data[current_idx : current_idx + 1])[0]
current_idx += 1
return reply
wdt_info = "WDT Type | Reboots | Time or Pings left (CSP only)"
log_to_both(printer, wdt_info)
for idx in range(len(wdt_reboots_list)):
log_to_both(
printer,
f"{WDT_LIST[idx].ljust(5)} | "
f"{wdt_reboots_list[idx]:010} | {time_pings_left_list[idx]:010}",
)
temps = (
f"In C: Temp 0 {temp0 / 10.0} | Temp 1 {temp1 / 10.0} | "
f"Batt Temp 0 {batt_temp_0 / 10.0} | Batt Temp 1 {batt_temp_1 / 10.0}"
)
batt_info = (
f"Batt: Current {batt_current} | Volt {batt_voltage} | "
f"Charge Current {batt_charge_current} | Discharge Current {batt_discharge_current}"
)
log_to_both(printer, temps)
log_to_both(printer, batt_info)
misc_info = f"Dearm {dearm_status} | ANT6 Depl {ant6_depl_status} | AR6 Deply {ar6_depl_status}"
log_to_both(printer, misc_info)
printer.print_validity_buffer(validity_buffer=hk_data[current_idx:], num_vars=36)
def log_to_both(printer: FsfwTmTcPrinter, string: str):
print(string)
printer.file_logger.info(string)

2
tmtccmd

@ -1 +1 @@
Subproject commit dfeaec0f452c32f2e3fa7eadaf91e3297fb8124e
Subproject commit e0f1f8c77e2a182181689743dbee065f0b894a26