Merge branch 'kranz/master' of https://egit.irs.uni-stuttgart.de/eive/eive-tmtc into kranz/master

2022-05-23 18:33:41 +02:00
parent c9f5a5652e 45bee6429e
commit dd7a19d134
13 changed files with 1036 additions and 852 deletions
--- a/pus_tc/cmd_definitions.py
+++ b/pus_tc/cmd_definitions.py
@ -1,4 +1,5 @@
 from pus_tc.devs.gps import GpsOpCodes
+from pus_tc.devs.rad_sensor import add_rad_sens_cmds
 from tmtccmd.config import (
    add_op_code_entry,
    add_service_op_code_entry,
@ -8,6 +9,7 @@ from tmtccmd.config import (
 )
 from config.definitions import CustomServiceList
 from pus_tc.devs.heater import add_heater_cmds
+from pus_tc.devs.rtd import specify_rtd_cmds
 from pus_tc.devs.reaction_wheels import add_rw_cmds
 from pus_tc.devs.bpx_batt import BpxOpCodes

@ -20,6 +22,7 @@ def get_eive_service_op_code_dict() -> ServiceOpCodeDictT:
    add_core_controller_definitions(cmd_dict=service_op_code_dict)
    add_pl_pcdu_cmds(cmd_dict=service_op_code_dict)
    add_pcdu_cmds(cmd_dict=service_op_code_dict)
+    specify_rtd_cmds(cmd_dict=service_op_code_dict)
    add_imtq_cmds(cmd_dict=service_op_code_dict)
    add_rad_sens_cmds(cmd_dict=service_op_code_dict)
    add_rw_cmds(cmd_dict=service_op_code_dict)
@ -775,20 +778,6 @@ def add_imtq_cmds(cmd_dict: ServiceOpCodeDictT):
    cmd_dict[CustomServiceList.IMTQ.value] = service_imtq_tuple


-def add_rad_sens_cmds(cmd_dict: ServiceOpCodeDictT):
-    op_code_dict_srv_rad_sensor = {
-        "0": ("Radiation Sensor: Set mode on", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "1": ("Radiation Sensor: Set mode normal", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "2": ("Radiation Sensor: Set mode off", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "3": ("Radiation Sensor: Start conversions", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "4": ("Radiation Sensor: Read conversions", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "5": ("Radiation Sensor: Enable debug output", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "6": ("Radiation Sensor: Disable debug putput", {OpCodeDictKeys.TIMEOUT: 2.0}),
-    }
-    service_rad_sensor_tuple = ("Radiation Sensor", op_code_dict_srv_rad_sensor)
-    cmd_dict[CustomServiceList.RAD_SENSOR.value] = service_rad_sensor_tuple
-
-
 def add_ploc_mpsoc_cmds(cmd_dict: ServiceOpCodeDictT):
    op_code_dict_srv_ploc_mpsoc = {
        "0": ("Ploc MPSoC: Set mode off", {OpCodeDictKeys.TIMEOUT: 2.0}),
@ -904,43 +893,9 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT):
        "55": ("PLOC Supervisor: Request ADC Report", {OpCodeDictKeys.TIMEOUT: 2.0}),
        "56": ("PLOC Supervisor: Reset PL", {OpCodeDictKeys.TIMEOUT: 2.0}),
        "57": ("PLOC Supervisor: Enable NVMs", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "58": ("PLOC Supervisor: Continue update", {OpCodeDictKeys.TIMEOUT: 2.0}),
    }
    service_ploc_supv_tuple = ("PLOC Supervisor", op_code_dict_srv_ploc_supv)
-
-    op_code_dict_srv_ploc_updater = {
-        "0": (
-            "Ploc Updater: Update uboot on partition A",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "1": (
-            "Ploc Updater: Update bitstream on partition A",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "2": (
-            "Ploc Updater: Update linux on partition A",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "3": (
-            "Ploc Updater: Update application on partition A",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "4": (
-            "Ploc Updater: Update uboot on partition B",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "5": (
-            "Ploc Updater: Update bitstream on partition B",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "6": (
-            "Ploc Updater: Update linux on partition B",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-        "7": (
-            "Ploc Updater: Update application on partition B",
-            {OpCodeDictKeys.TIMEOUT: 2.0},
-        ),
-    }
    cmd_dict[CustomServiceList.PLOC_SUPV.value] = service_ploc_supv_tuple
    cmd_dict[
        CustomServiceList.PLOC_MEMORY_DUMPER.value
--- a/pus_tc/devs/ploc_supervisor.py
+++ b/pus_tc/devs/ploc_supervisor.py
@ -98,6 +98,7 @@ class SupvActionIds:
    REQUEST_ADC_REPORT = 57
    RESET_PL = 58
    ENABLE_NVMS = 59
+    CONTINUE_UPDATE = 60


 class SupvHkIds:
@ -397,6 +398,11 @@ def pack_ploc_supv_commands(
        )
        command = PusTelecommand(service=8, subservice=128, ssc=72, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "58":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Continue update"))
+        command = object_id + struct.pack("!I", SupvActionIds.CONTINUE_UPDATE)
+        command = PusTelecommand(service=8, subservice=128, ssc=73, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())

    return tc_queue

--- a/pus_tc/devs/rad_sensor.py
+++ b/pus_tc/devs/rad_sensor.py
@ -7,11 +7,37 @@
 """
 import struct

-from tmtccmd.config.definitions import QueueCommands
+from config.definitions import CustomServiceList
+from tmtccmd.config import add_op_code_entry, add_service_op_code_entry
+from tmtccmd.config.definitions import QueueCommands, ServiceOpCodeDictT, OpCodeDictKeys

 from tmtccmd.tc.packer import TcQueueT
 from spacepackets.ecss.tc import PusTelecommand
 from pus_tc.service_200_mode import pack_mode_data, Modes
+from tmtccmd.tc.pus_3_fsfw_hk import generate_one_hk_command, make_sid
+from tmtccmd.utility import ObjectId
+
+
+class SetIds:
+    HK = 3
+
+
+class OpCodes:
+    ON = ["0", "on"]
+    NORMAL = ["1", "normal"]
+    OFF = ["2", "off"]
+    REQ_HK_ONCE = ["3", "hk-os"]
+    DEBUG_ON = ["10", "dbg-on"]
+    DEBUG_OFF = ["11", "dbg-off"]
+
+
+class Info:
+    ON = "Switch Rad Sensor on"
+    NORMAL = "Switch Rad Sensor normal"
+    OFF = "Switch Rad sensor off"
+    REQ_OS_HK = "Request one-shot HK"
+    DEBUG_ON = "Switch debug output on"
+    DEBUG_OFF = "Switch debug output off"


 class CommandIds:
@ -21,52 +47,64 @@ class CommandIds:
    DISABLE_DEBUG_OUTPUT = 5


-def pack_rad_sensor_test_into(object_id: bytearray, tc_queue: TcQueueT, op_code: str):
-    tc_queue.appendleft(
-        (
-            QueueCommands.PRINT,
-            "Testing radiation sensor handler with object id: 0x" + object_id.hex(),
-        )
+def add_rad_sens_cmds(cmd_dict: ServiceOpCodeDictT):
+
+    op_code_dict = dict()
+    add_op_code_entry(op_code_dict=op_code_dict, info=Info.ON, keys=OpCodes.ON)
+    add_op_code_entry(op_code_dict=op_code_dict, info=Info.OFF, keys=OpCodes.OFF)
+    add_op_code_entry(op_code_dict=op_code_dict, info=Info.NORMAL, keys=OpCodes.NORMAL)
+    add_op_code_entry(
+        op_code_dict=op_code_dict, info=Info.REQ_OS_HK, keys=OpCodes.REQ_HK_ONCE
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict, info=Info.DEBUG_ON, keys=OpCodes.DEBUG_ON
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict, info=Info.DEBUG_OFF, keys=OpCodes.DEBUG_OFF
+    )
+    add_service_op_code_entry(
+        srv_op_code_dict=cmd_dict,
+        name=CustomServiceList.RAD_SENSOR.value,
+        info="Radiation Sensor",
+        op_code_entry=op_code_dict,
    )

-    if op_code == "0":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Switch to mode on"))
-        mode_data = pack_mode_data(object_id, Modes.ON, 0)
-        command = PusTelecommand(service=200, subservice=1, ssc=41, app_data=mode_data)
-        tc_queue.appendleft(command.pack_command_tuple())

-    if op_code == "1":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Switch to mode normal"))
-        mode_data = pack_mode_data(object_id, Modes.NORMAL, 0)
-        command = PusTelecommand(service=200, subservice=1, ssc=42, app_data=mode_data)
-        tc_queue.appendleft(command.pack_command_tuple())
+def pack_rad_sensor_test_into(object_id: ObjectId, tc_queue: TcQueueT, op_code: str):
+    tc_queue.appendleft(
+        (QueueCommands.PRINT, f"Commanding Radiation sensor handler {object_id}")
+    )

-    if op_code == "2":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Switch to mode off"))
-        mode_data = pack_mode_data(object_id, Modes.OFF, 0)
-        command = PusTelecommand(service=200, subservice=1, ssc=42, app_data=mode_data)
+    if op_code in OpCodes.ON:
+        rad_sensor_mode_cmd(object_id, Modes.ON, Info.ON, tc_queue)
+    if op_code in OpCodes.NORMAL:
+        rad_sensor_mode_cmd(object_id, Modes.NORMAL, Info.NORMAL, tc_queue)
+    if op_code in OpCodes.OFF:
+        rad_sensor_mode_cmd(object_id, Modes.OFF, Info.OFF, tc_queue)
+    if op_code in OpCodes.REQ_HK_ONCE:
+        tc_queue.appendleft((QueueCommands.PRINT, f"Rad sensor: {Info.REQ_OS_HK}"))
+        cmd = generate_one_hk_command(
+            sid=make_sid(object_id.as_bytes, set_id=SetIds.HK), ssc=0
+        )
+        tc_queue.appendleft(cmd.pack_command_tuple())
+    if op_code in OpCodes.DEBUG_ON:
+        tc_queue.appendleft((QueueCommands.PRINT, f"Rad sensor: {Info.DEBUG_ON}"))
+        command = object_id.as_bytes + struct.pack("!I", CommandIds.ENABLE_DEBUG_OUTPUT)
+        command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
-
-    if op_code == "3":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Start  conversions"))
-        command = object_id + struct.pack("!I", CommandIds.START_CONVERSIONS)
-        command = PusTelecommand(service=8, subservice=128, ssc=43, app_data=command)
-        tc_queue.appendleft(command.pack_command_tuple())
-
-    if op_code == "4":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Read  conversions"))
-        command = object_id + struct.pack("!I", CommandIds.READ_CONVERSIONS)
-        command = PusTelecommand(service=8, subservice=128, ssc=44, app_data=command)
-        tc_queue.appendleft(command.pack_command_tuple())
-
-    if op_code == "5":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Enable debug output"))
-        command = object_id + struct.pack("!I", CommandIds.ENABLE_DEBUG_OUTPUT)
+    if op_code in OpCodes.DEBUG_OFF:
+        tc_queue.appendleft((QueueCommands.PRINT, f"Rad sensor: {Info.DEBUG_OFF}"))
+        command = object_id.as_bytes + struct.pack(
+            "!I", CommandIds.DISABLE_DEBUG_OUTPUT
+        )
        command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())

-    if op_code == "6":
-        tc_queue.appendleft((QueueCommands.PRINT, "Rad sensor: Disable debug output"))
-        command = object_id + struct.pack("!I", CommandIds.DISABLE_DEBUG_OUTPUT)
-        command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command)
-        tc_queue.appendleft(command.pack_command_tuple())
+
+def rad_sensor_mode_cmd(
+    object_id: ObjectId, mode: Modes, info: str, tc_queue: TcQueueT
+):
+    tc_queue.appendleft((QueueCommands.PRINT, f"Rad sensor: {info}"))
+    mode_data = pack_mode_data(object_id.as_bytes, mode, 0)
+    command = PusTelecommand(service=200, subservice=1, ssc=41, app_data=mode_data)
+    tc_queue.appendleft(command.pack_command_tuple())
--- a/pus_tc/devs/rtd.py
+++ b/pus_tc/devs/rtd.py
@ -0,0 +1,113 @@
+from typing import Optional
+
+from config.definitions import CustomServiceList
+from spacepackets.ecss import PusTelecommand
+from tmtccmd.config import ServiceOpCodeDictT, add_op_code_entry, add_service_op_code_entry
+from tmtccmd.tc.definitions import TcQueueT
+from tmtccmd.utility import ObjectId
+from tmtccmd.tc.pus_200_fsfw_modes import Modes, pack_mode_data, Subservices
+import config.object_ids as oids
+from config.object_ids import get_object_ids
+
+RTD_IDS = [
+    oids.RTD_0_PLOC_HSPD,
+    oids.RTD_1_PLOC_MISSIONBRD,
+    oids.RTD_2_4K_CAM,
+    oids.RTD_3_DAC_HSPD,
+    oids.RTD_4_STR,
+    oids.RTD_5_RW1_MX_MY,
+    oids.RTD_6_DRO,
+    oids.RTD_7_SCEX,
+    oids.RTD_8_X8,
+    oids.RTD_9_HPA,
+    oids.RTD_10_PL_TX,
+    oids.RTD_11_MPA,
+    oids.RTD_12_ACU,
+    oids.RTD_13_PLPCDU_HSPD,
+    oids.RTD_14_TCS_BRD,
+    oids.RTD_15_IMTQ
+]
+
+
+class OpCodes:
+    ON = ["0", "on"]
+    OFF = ["1", "off"]
+    NORMAL = ["2", "normal"]
+
+
+class Info:
+    ON = "Switch handler on"
+    OFF = "Switch handler off"
+    NORMAL = "Switch handler normal"
+
+
+def specify_rtd_cmds(cmd_dict: ServiceOpCodeDictT):
+    op_code_dict = dict()
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.ON,
+        info=Info.ON
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL,
+        info=Info.NORMAL
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.OFF,
+        info=Info.OFF
+    )
+    add_service_op_code_entry(
+        srv_op_code_dict=cmd_dict,
+        op_code_entry=op_code_dict,
+        name=CustomServiceList.RTD.value,
+        info="RTD commands"
+    )
+
+
+def pack_rtd_commands(op_code: str, object_id: Optional[ObjectId], tc_queue: TcQueueT):
+    if object_id is not None and object_id not in RTD_IDS:
+        print("Specified object ID not a valid RTD ID")
+        object_id = None
+    if object_id is None:
+        tgt_rtd_idx = prompt_rtd_idx()
+        object_id_dict = get_object_ids()
+        object_id = object_id_dict.get(RTD_IDS[tgt_rtd_idx])
+    if op_code in OpCodes.ON:
+        app_data = pack_mode_data(object_id=object_id.as_bytes, mode=Modes.ON, submode=0)
+        cmd = PusTelecommand(
+            service=200,
+            subservice=Subservices.TC_MODE_COMMAND,
+            app_data=app_data
+        )
+        tc_queue.appendleft(cmd.pack_command_tuple())
+    if op_code in OpCodes.NORMAL:
+        app_data = pack_mode_data(object_id=object_id.as_bytes, mode=Modes.NORMAL, submode=0)
+        cmd = PusTelecommand(
+            service=200,
+            subservice=Subservices.TC_MODE_COMMAND,
+            app_data=app_data
+        )
+        tc_queue.appendleft(cmd.pack_command_tuple())
+    if op_code in OpCodes.OFF:
+        app_data = pack_mode_data(object_id=object_id.as_bytes, mode=Modes.OFF, submode=0)
+        cmd = PusTelecommand(
+            service=200,
+            subservice=Subservices.TC_MODE_COMMAND,
+            app_data=app_data
+        )
+        tc_queue.appendleft(cmd.pack_command_tuple())
+
+
+def prompt_rtd_idx():
+    while True:
+        rtd_idx = input("Please specify RTD index [0-15]: ")
+        if not rtd_idx.isdigit():
+            print("Invalid input")
+            continue
+        rtd_idx = int(rtd_idx)
+        if rtd_idx < 0 or rtd_idx > 15:
+            print("Invalid device index")
+            continue
+        return rtd_idx
--- a/pus_tc/tc_packer_hook.py
+++ b/pus_tc/tc_packer_hook.py
@ -5,6 +5,7 @@ import os
 from collections import deque
 from typing import Union

+from pus_tc.devs.rtd import pack_rtd_commands
 from spacepackets.ecss import PusTelecommand
 from tmtccmd.com_if.com_interface_base import CommunicationInterface
 from tmtccmd.config.definitions import CoreServiceList, QueueCommands
@ -69,6 +70,7 @@ from config.object_ids import (
    SYRLINKS_HANDLER_ID,
    SOLAR_ARRAY_DEPLOYMENT_ID,
    RW_ASSEMBLY,
+    get_object_ids,
 )


@ -98,6 +100,7 @@ def pre_tc_send_cb(
 def pack_service_queue_user(
    service: Union[str, int], op_code: str, service_queue: TcQueueT
 ):
+    obj_id_man = get_object_ids()
    if service == CoreServiceList.SERVICE_5.value:
        return pack_generic_service5_test_into(tc_queue=service_queue)
    if service == CoreServiceList.SERVICE_17.value:
@ -111,6 +114,8 @@ def pack_service_queue_user(
        return pack_p60dock_cmds(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
+    if service == CustomServiceList.RTD.value:
+        return pack_rtd_commands(object_id=None, tc_queue=service_queue, op_code=op_code)
    if service == CustomServiceList.PDU1.value:
        object_id = PDU_1_HANDLER_ID
        return pack_pdu1_commands(
@ -170,7 +175,7 @@ def pack_service_queue_user(
            object_id=RW4_ID, rw_idx=4, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.RAD_SENSOR.value:
-        object_id = RAD_SENSOR_ID
+        object_id = obj_id_man.get(RAD_SENSOR_ID)
        return pack_rad_sensor_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )