"""HK Handling for EIVE OBSW""" import logging # from pus_tm.tcp_server_objects import TCP_SEVER_SENSOR_TEMPERATURES from eive_tmtc.tmtc.acs.acs_ctrl import handle_raw_mgm_data, handle_acs_ctrl_hk_data from eive_tmtc.tmtc.power.plpcdu import handle_plpcdu_hk from eive_tmtc.tmtc.payload.rad_sensor import handle_rad_sensor_data from eive_tmtc.tmtc.acs.sus import handle_sus_hk from eive_tmtc.tmtc.payload.ploc_supervisor import handle_supv_hk_data from eive_tmtc.tmtc.acs.reaction_wheels import handle_rw_hk_data from eive_tmtc.tmtc.com.syrlinks_handler import handle_syrlinks_hk_data from eive_tmtc.tmtc.tcs import handle_thermal_controller_hk_data from tmtccmd.tm.pus_3_fsfw_hk import ( Service3Base, HkContentType, Service3FsfwTm, ) from tmtccmd.util.obj_id import ObjectIdU32, ObjectIdDictT from eive_tmtc.tmtc.power.bpx_batt import handle_bpx_hk_data from eive_tmtc.tmtc.acs.gps import handle_gps_data from eive_tmtc.tmtc.acs.gyros import handle_gyros_hk_data from eive_tmtc.tmtc.power.tm import ( handle_pdu_data, handle_p60_hk_data, handle_acu_hk_data, handle_pcdu_hk, ) from eive_tmtc.tmtc.acs.imtq import ( ImtqSetId, handle_self_test_data, handle_eng_set, handle_calibrated_mtm_measurement, handle_raw_mtm_measurement, ) from eive_tmtc.pus_tm.defs import FsfwTmTcPrinter from eive_tmtc.tmtc.core import handle_core_hk_data from eive_tmtc.tmtc.acs.mgms import handle_mgm_hk_data import eive_tmtc.config.object_ids as obj_ids _LOGGER = logging.getLogger(__name__) FORWARD_SENSOR_TEMPS = False def handle_hk_packet( raw_tm: bytes, obj_id_dict: ObjectIdDictT, printer: FsfwTmTcPrinter, ): 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:] if FORWARD_SENSOR_TEMPS: # TODO: Maybe use singleton? # TCP_SEVER_SENSOR_TEMPERATURES.report_raw_hk_data( # object_id=named_obj_id, set_id=tm_packet.set_id, hk_data=hk_data # ) pass printer.generic_hk_tm_print( content_type=HkContentType.HK, object_id=named_obj_id, set_id=tm_packet.set_id, hk_data=hk_data, ) try: handle_regular_hk_print( printer=printer, object_id=named_obj_id, hk_packet=tm_packet, hk_data=hk_data, ) except ValueError as e: _LOGGER.exception( f"{e} error when parsing HK data coming from {named_obj_id}" ) 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: ObjectIdU32, hk_packet: Service3Base, hk_data: bytes, ): objb = object_id.as_bytes set_id = hk_packet.set_id """This function is called when a Service 3 Housekeeping packet is received.""" if objb in [obj_ids.RW1_ID, obj_ids.RW2_ID, obj_ids.RW3_ID, obj_ids.RW4_ID]: return handle_rw_hk_data(printer, object_id, set_id, hk_data) elif objb == obj_ids.SYRLINKS_HANDLER_ID: return handle_syrlinks_hk_data(printer=printer, hk_data=hk_data, set_id=set_id) elif objb == obj_ids.IMTQ_HANDLER_ID: if (set_id >= ImtqSetId.POSITIVE_X_TEST) and ( set_id <= ImtqSetId.NEGATIVE_Z_TEST ): return handle_self_test_data(printer, hk_data) elif set_id == ImtqSetId.ENG_HK_SET: return handle_eng_set(printer, hk_data) elif set_id == ImtqSetId.CAL_MTM_SET: return handle_calibrated_mtm_measurement(printer, hk_data) elif set_id == ImtqSetId.RAW_MTM_SET: return handle_raw_mtm_measurement(printer, hk_data) else: _LOGGER.info("Service 3 TM: IMTQ handler reply with unknown set id") elif objb == obj_ids.GPS_CONTROLLER: return handle_gps_data(printer=printer, hk_data=hk_data) elif objb == obj_ids.PCDU_HANDLER_ID: return handle_pcdu_hk(printer=printer, set_id=set_id, hk_data=hk_data) elif objb == obj_ids.BPX_HANDLER_ID: return handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, printer=printer) elif objb == obj_ids.CORE_CONTROLLER_ID: return handle_core_hk_data(printer=printer, hk_data=hk_data, set_id=set_id) elif objb == obj_ids.PDU_1_HANDLER_ID: return handle_pdu_data( printer=printer, pdu_idx=1, set_id=set_id, hk_data=hk_data ) elif objb == obj_ids.PDU_2_HANDLER_ID: return handle_pdu_data( printer=printer, pdu_idx=2, set_id=set_id, hk_data=hk_data ) elif objb == obj_ids.ACU_HANDLER_ID: return handle_acu_hk_data(printer=printer, hk_data=hk_data, set_id=set_id) elif objb == obj_ids.RAD_SENSOR_ID: return handle_rad_sensor_data(printer=printer, hk_data=hk_data, set_id=set_id) elif objb in [obj_ids.RW1_ID, obj_ids.RW2_ID, obj_ids.RW3_ID, obj_ids.RW4_ID]: return handle_rw_hk_data( printer=printer, object_id=object_id, set_id=set_id, hk_data=hk_data ) if objb in [ obj_ids.SUS_0_N_LOC_XFYFZM_PT_XF, obj_ids.SUS_1_N_LOC_XBYFZM_PT_XB, obj_ids.SUS_2_N_LOC_XFYBZB_PT_YB, obj_ids.SUS_3_N_LOC_XFYBZF_PT_YF, obj_ids.SUS_4_N_LOC_XMYFZF_PT_ZF, obj_ids.SUS_5_N_LOC_XFYMZB_PT_ZB, obj_ids.SUS_6_R_LOC_XFYBZM_PT_XF, obj_ids.SUS_7_R_LOC_XBYBZM_PT_XB, obj_ids.SUS_8_R_LOC_XBYBZB_PT_YB, obj_ids.SUS_9_R_LOC_XBYBZB_PT_YF, obj_ids.SUS_10_R_LOC_XMYBZF_PT_ZF, obj_ids.SUS_11_R_LOC_XBYMZB_PT_ZB, ]: return handle_sus_hk( object_id=object_id, hk_data=hk_data, printer=printer, set_id=set_id ) elif objb == obj_ids.P60_DOCK_HANDLER: return handle_p60_hk_data(printer=printer, set_id=set_id, hk_data=hk_data) elif objb in [ obj_ids.GYRO_0_ADIS_HANDLER_ID, obj_ids.GYRO_1_L3G_HANDLER_ID, obj_ids.GYRO_2_ADIS_HANDLER_ID, obj_ids.GYRO_3_L3G_HANDLER_ID, ]: return handle_gyros_hk_data( object_id=object_id, hk_data=hk_data, printer=printer, set_id=set_id ) elif objb in [ obj_ids.MGM_0_LIS3_HANDLER_ID, obj_ids.MGM_1_RM3100_HANDLER_ID, obj_ids.MGM_2_LIS3_HANDLER_ID, obj_ids.MGM_3_RM3100_HANDLER_ID, ]: return handle_mgm_hk_data( object_id=object_id, hk_data=hk_data, printer=printer, set_id=set_id ) elif objb == obj_ids.PL_PCDU_ID: return handle_plpcdu_hk(set_id=set_id, hk_data=hk_data, printer=printer) elif objb == obj_ids.THERMAL_CONTROLLER_ID: return handle_thermal_controller_hk_data( object_id=object_id, printer=printer, set_id=set_id, hk_data=hk_data ) elif objb == obj_ids.PLOC_SUPV_ID: return handle_supv_hk_data(set_id=set_id, hk_data=hk_data, printer=printer) elif objb == obj_ids.ACS_CONTROLLER: return handle_acs_ctrl_hk_data(printer, set_id, hk_data) else: _LOGGER.info( f"Service 3 TM: Parsing for object {object_id} and set ID {set_id} " f"has not been implemented." )