diff --git a/eive_tmtc/tmtc/acs/acs_ctrl.py b/eive_tmtc/tmtc/acs/acs_ctrl.py index 2b6fec2..3206056 100644 --- a/eive_tmtc/tmtc/acs/acs_ctrl.py +++ b/eive_tmtc/tmtc/acs/acs_ctrl.py @@ -333,10 +333,8 @@ def pack_acs_ctrl_command(p: ServiceProviderParams): # noqa C901 break else: print("The line does not have the required length of 69 characters") - tle = line1.encode()+line2.encode() - q.add_pus_tc( - create_action_cmd(ACS_CONTROLLER, ActionId.UPDATE_TLE,tle) - ) + tle = line1.encode() + line2.encode() + q.add_pus_tc(create_action_cmd(ACS_CONTROLLER, ActionId.UPDATE_TLE, tle)) elif op_code in OpCodes.SET_PARAMETER_SCALAR: q.add_log_cmd(f"{Info.SET_PARAMETER_SCALAR}") set_acs_ctrl_param_scalar(q) @@ -1059,7 +1057,9 @@ def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes): ) ] current_idx += inc_len_vec - source = struct.unpack(fmt_source, hk_data[current_idx: current_idx + inc_len_source])[0] + source = struct.unpack( + fmt_source, hk_data[current_idx : current_idx + inc_len_source] + )[0] current_idx += inc_len_source if GPS_COURCE_DICT.get(source) is not None: pw.dlog(f"GPS Source: {GPS_COURCE_DICT[source]}") diff --git a/eive_tmtc/tmtc/acs/gps.py b/eive_tmtc/tmtc/acs/gps.py index 9de4692..41a8a7f 100644 --- a/eive_tmtc/tmtc/acs/gps.py +++ b/eive_tmtc/tmtc/acs/gps.py @@ -19,9 +19,11 @@ from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter _LOGGER = logging.getLogger(__name__) + class GpsInfo: MAX_SATELLITES = 30 + class OpCode: OFF = "off" ON = "on" @@ -70,7 +72,9 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper): ) -def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str): +def pack_gps_command( # noqa: C901 + object_id: bytes, q: DefaultPusQueueHelper, op_code: str +): # noqa: C901 if op_code in OpCode.RESET_GNSS: # TODO: This needs to be re-implemented _LOGGER.warning("Reset pin handling needs to be re-implemented") @@ -80,34 +84,52 @@ def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str): raise ValueError("invalid interval") q.add_log_cmd(f"GPS: {Info.ENABLE_CORE_HK}") cmds = create_enable_periodic_hk_command_with_interval( - diag=False, sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK), interval_seconds=interval + diag=False, + sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK), + interval_seconds=interval, ) for cmd in cmds: q.add_pus_tc(cmd) if op_code in OpCode.DISABLE_CORE_HK: q.add_log_cmd(f"gps: {Info.DISABLE_CORE_HK}") - q.add_pus_tc(create_disable_periodic_hk_command(diag=False, sid=make_sid(object_id=object_id, - set_id=SetId.CORE_HK))) + q.add_pus_tc( + create_disable_periodic_hk_command( + diag=False, sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK) + ) + ) if op_code in OpCode.REQ_CORE_HK: q.add_log_cmd(f"GPS: {Info.REQ_CORE_HK}") - q.add_pus_tc(create_request_one_hk_command(sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK))) + q.add_pus_tc( + create_request_one_hk_command( + sid=make_sid(object_id=object_id, set_id=SetId.CORE_HK) + ) + ) if op_code in OpCode.ENABLE_SKYVIEW_HK: interval = float(input("Please specify interval in floating point seconds: ")) if interval <= 0: raise ValueError("invalid interval") q.add_log_cmd(f"GPS: {Info.ENABLE_SKYVIEW_HK}") cmds = create_enable_periodic_hk_command_with_interval( - diag=False, sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK), interval_seconds=interval + diag=False, + sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK), + interval_seconds=interval, ) for cmd in cmds: q.add_pus_tc(cmd) if op_code in OpCode.DISABLE_SKYVIEW_HK: q.add_log_cmd(f"gps: {Info.DISABLE_SKYVIEW_HK}") - q.add_pus_tc(create_disable_periodic_hk_command(diag=False, sid=make_sid(object_id=object_id, - set_id=SetId.SKYVIEW_HK))) + q.add_pus_tc( + create_disable_periodic_hk_command( + diag=False, sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK) + ) + ) if op_code in OpCode.REQ_SKYVIEW_HK: q.add_log_cmd(f"GPS: {Info.REQ_SKYVIEW_HK}") - q.add_pus_tc(create_request_one_hk_command(sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK))) + q.add_pus_tc( + create_request_one_hk_command( + sid=make_sid(object_id=object_id, set_id=SetId.SKYVIEW_HK) + ) + ) if op_code in OpCode.ON: q.add_log_cmd(f"GPS: {Info.ON}") q.add_pus_tc(create_mode_command(object_id, Mode.ON, 0)) @@ -131,7 +153,7 @@ def handle_gps_data( def handle_core_data(pw: PrintWrapper, hk_data: bytes): - if len(hk_data) < 4*8+4+2+8: + if len(hk_data) < 4 * 8 + 4 + 2 + 8: pw.dlog( f"GPS Core dataset with size {len(hk_data)} does not have expected size" f" of {4*8+4+2+8} bytes" @@ -177,7 +199,7 @@ def handle_core_data(pw: PrintWrapper, hk_data: bytes): def handle_skyview_data(pw: PrintWrapper, hk_data: bytes): - data_length = 8+GpsInfo.MAX_SATELLITES*(8+3*2+1) + data_length = 8 + GpsInfo.MAX_SATELLITES * (8 + 3 * 2 + 1) if len(hk_data) < data_length: pw.dlog( f"GPS Skyview dataset with size {len(hk_data)} does not have expected size" @@ -193,24 +215,46 @@ def handle_skyview_data(pw: PrintWrapper, hk_data: bytes): inc_len_int16 = struct.calcsize(fmt_str_int16) inc_len_double = struct.calcsize(fmt_str_double) inc_len_uint8 = struct.calcsize(fmt_str_uint8) - unix = struct.unpack(fmt_str_unix, hk_data[current_idx: current_idx + inc_len_unix]) + unix = struct.unpack( + fmt_str_unix, hk_data[current_idx : current_idx + inc_len_unix] + ) current_idx += inc_len_unix - prn_id = struct.unpack(fmt_str_int16, hk_data[current_idx: current_idx + inc_len_int16]) + prn_id = struct.unpack( + fmt_str_int16, hk_data[current_idx : current_idx + inc_len_int16] + ) current_idx += inc_len_int16 - azimuth = struct.unpack(fmt_str_int16, hk_data[current_idx: current_idx + inc_len_int16]) + azimuth = struct.unpack( + fmt_str_int16, hk_data[current_idx : current_idx + inc_len_int16] + ) current_idx += inc_len_int16 - elevation = struct.unpack(fmt_str_int16, hk_data[current_idx: current_idx + inc_len_int16]) + elevation = struct.unpack( + fmt_str_int16, hk_data[current_idx : current_idx + inc_len_int16] + ) current_idx += inc_len_int16 - signal_to_noise = struct.unpack(fmt_str_double, hk_data[current_idx: current_idx + inc_len_double]) + signal_to_noise = struct.unpack( + fmt_str_double, hk_data[current_idx : current_idx + inc_len_double] + ) current_idx += inc_len_double - used = struct.unpack(fmt_str_uint8, hk_data[current_idx: current_idx + inc_len_uint8]) + used = struct.unpack( + fmt_str_uint8, hk_data[current_idx : current_idx + inc_len_uint8] + ) current_idx += inc_len_uint8 pw.dlog(f"Skyview Time: {unix} unix-sec") - pw.dlog("{:<8} {:<8} {:<8} {:<8} {:<8}".format('PRN_ID', 'AZ [°]', 'EL [°]', 'S2N [dBW]', 'USED')) + pw.dlog( + "{:<8} {:<8} {:<8} {:<8} {:<8}".format( + "PRN_ID", "AZ [°]", "EL [°]", "S2N [dBW]", "USED" + ) + ) for idx in range(GpsInfo.MAX_SATELLITES): - pw.dlog("{:<8} {:<8} {:<8} {:<8} {:<8}".format(prn_id[idx], azimuth[idx], elevation[idx], signal_to_noise[idx], - used[idx])) + pw.dlog( + "{:<8} {:<8} {:<8} {:<8} {:<8}".format( + prn_id[idx], + azimuth[idx], + elevation[idx], + signal_to_noise[idx], + used[idx], + ) + ) FsfwTmTcPrinter.get_validity_buffer( validity_buffer=hk_data[current_idx:], num_vars=6 ) - diff --git a/eive_tmtc/tmtc/power/plpcdu.py b/eive_tmtc/tmtc/power/plpcdu.py index 745d7b4..cfe9dac 100644 --- a/eive_tmtc/tmtc/power/plpcdu.py +++ b/eive_tmtc/tmtc/power/plpcdu.py @@ -85,6 +85,39 @@ class NormalSubmodesMask(enum.IntEnum): HPA_ON = 5 +class SubmodeForNormalMode(enum.IntEnum): + SSR_ON = 1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON + DRO_ON = 1 << NormalSubmodesMask.DRO_ON | ( + 1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON + ) + X8_ON = ( + 1 << NormalSubmodesMask.DRO_ON + | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) + | (1 << NormalSubmodesMask.X8_ON) + ) + TX_ON = ( + 1 << NormalSubmodesMask.DRO_ON + | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) + | (1 << NormalSubmodesMask.X8_ON) + | (1 << NormalSubmodesMask.TX_ON) + ) + MPA_ON = ( + 1 << NormalSubmodesMask.DRO_ON + | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) + | (1 << NormalSubmodesMask.X8_ON) + | (1 << NormalSubmodesMask.TX_ON) + | (1 << NormalSubmodesMask.MPA_ON) + ) + HPA_ON = ( + 1 << NormalSubmodesMask.DRO_ON + | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) + | (1 << NormalSubmodesMask.X8_ON) + | (1 << NormalSubmodesMask.TX_ON) + | (1 << NormalSubmodesMask.MPA_ON) + | (1 << NormalSubmodesMask.HPA_ON) + ) + + class ParamIds(enum.IntEnum): NEG_V_LOWER_BOUND = 0 NEG_V_UPPER_BOUND = 1 @@ -383,41 +416,17 @@ def request_wait_time() -> Optional[float]: def submode_mask_to_submode(on_tgt: NormalSubmodesMask) -> int: if on_tgt == NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON: - return 1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON + return SubmodeForNormalMode.SSR_ON if on_tgt == NormalSubmodesMask.DRO_ON: - return 1 << NormalSubmodesMask.DRO_ON | ( - 1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON - ) + return SubmodeForNormalMode.DRO_ON if on_tgt == NormalSubmodesMask.X8_ON: - return ( - 1 << NormalSubmodesMask.DRO_ON - | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) - | (1 << NormalSubmodesMask.X8_ON) - ) + return SubmodeForNormalMode.X8_ON if on_tgt == NormalSubmodesMask.TX_ON: - return ( - 1 << NormalSubmodesMask.DRO_ON - | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) - | (1 << NormalSubmodesMask.X8_ON) - | (1 << NormalSubmodesMask.TX_ON) - ) + return SubmodeForNormalMode.TX_ON if on_tgt == NormalSubmodesMask.MPA_ON: - return ( - 1 << NormalSubmodesMask.DRO_ON - | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) - | (1 << NormalSubmodesMask.X8_ON) - | (1 << NormalSubmodesMask.TX_ON) - | (1 << NormalSubmodesMask.MPA_ON) - ) + return SubmodeForNormalMode.MPA_ON if on_tgt == NormalSubmodesMask.HPA_ON: - return ( - 1 << NormalSubmodesMask.DRO_ON - | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) - | (1 << NormalSubmodesMask.X8_ON) - | (1 << NormalSubmodesMask.TX_ON) - | (1 << NormalSubmodesMask.MPA_ON) - | (1 << NormalSubmodesMask.HPA_ON) - ) + return SubmodeForNormalMode.HPA_ON def pack_wait_time_cmd(q: DefaultPusQueueHelper, param_id: int, print_str: str):