Merge pull request 'PL PCDU enum' (#229) from new-pl-pcdu-enum into main

Reviewed-on: #229
2023-08-15 13:07:40 +02:00 · 2023-08-15 13:07:40 +02:00 · 23f21b40eb
commit 23f21b40eb
parent d6f6aff139 a6b3ccb4cc
3 changed files with 109 additions and 56 deletions
--- a/eive_tmtc/tmtc/acs/acs_ctrl.py
+++ b/eive_tmtc/tmtc/acs/acs_ctrl.py
@ -334,9 +334,7 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):  # noqa C901
            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)
-        )
+        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]}")
--- 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))
@ -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
    )
-
--- 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):