Implement safety-critical command handling

gs/backend/obc_utils/command_utils.py

@@ -16,12 +16,14 @@
     CmdCallbackId,
     CmdMsg,
     CmdResponseErrorCode,
+    create_cmd_arm,
     create_cmd_downlink_logs_next_pass,
     create_cmd_downlink_telem,
     create_cmd_download_data,
     create_cmd_end_of_frame,
     create_cmd_erase_app,
     create_cmd_exec_obc_reset,
+    create_cmd_execute,
     create_cmd_i2c_probe,
     create_cmd_mirco_sd_format,
     create_cmd_ping,
@@ -222,6 +224,56 @@ def parse_cmd_downlink_logs_next_pass() -> ArgumentParser:
     return parser
 
 
+def parse_cmd_arm() -> ArgumentParser:
+    """
+    A function to parse the arguments for the arm command
+    """
+    parent_parser = arg_parse()
+    parser = ArgumentParser(parents=[parent_parser], add_help=False, exit_on_error=False)
+    parser.add_argument(
+        "-ad",
+        "--arm_data",
+        required=True,
+        dest="arg1",
+        type=int,
+        help="The arm data for the arm command",
+    )
+    parser.add_argument(
+        "-aid",
+        "--arm_id_data",
+        required=True,
+        dest="arg2",
+        type=int,
+        help="The arm id data for the arm command",
+    )
+    return parser
+
+
+def parse_cmd_execute() -> ArgumentParser:
+    """
+    A function to parse the arguments for the execute command
+    """
+    parent_parser = arg_parse()
+    parser = ArgumentParser(parents=[parent_parser], add_help=False, exit_on_error=False)
+    parser.add_argument(
+        "-ed",
+        "--execute_data",
+        required=True,
+        dest="arg1",
+        type=int,
+        help="The execute data for the execute command",
+    )
+    parser.add_argument(
+        "-eid",
+        "--execute_id_data",
+        required=True,
+        dest="arg2",
+        type=int,
+        help="The execute id data for the execute command",
+    )
+    return parser
+
+
 # End of specific command parsers
 
 
@@ -257,6 +309,8 @@ def generate_command(args: str) -> tuple[CmdMsg | None, bool]:
         create_cmd_download_data,
         create_cmd_verify_crc,
         create_cmd_i2c_probe,
+        create_cmd_arm,
+        create_cmd_execute,
     ]
 
     # Loop through each of the specific parses and see if we get a valid parse on any of them

interfaces/obc_gs_interface/commands/__init__.py

@@ -54,6 +54,24 @@ class SetProgrammingSessionCmdData(Structure):
     _fields_ = [("programmingSession", c_uint)]
 
 
+class CmdArmCmdData(Structure):
+    """
+    The python equivalent class for the cmd_arm_cmd_data_t structure in the C implementation
+    """
+
+    _fields_ = [("cmdArm", c_uint)]
+    _fields_ = [("armId", c_uint)]
+
+
+class CmdExecuteCmdData(Structure):
+    """
+    The python equivalent class for the cmd_execute_cmd_data_t structure in the C implementation
+    """
+
+    _fields_ = [("cmdExecute", c_uint)]
+    _fields_ = [("execId", c_uint)]
+
+
 # NOTE: When adding commands only add their data to the following union type as shown with RtcSyncCmdData and
 # DownlinkLogsNextPassCmdData
 class _U(Union):
@@ -66,6 +84,8 @@ class _U(Union):
         ("downlinkLogsNextPass", DownlinkLogsNextPassCmdData),
         ("downloadData", DownloadDataCmdData),
         ("setProgrammingSession", SetProgrammingSessionCmdData),
+        ("cmdArm", CmdArmCmdData),
+        ("cmdExecute", CmdExecuteCmdData),
     ]
 
 
@@ -159,7 +179,9 @@ class CmdCallbackId(IntEnum):
     CMD_DOWNLOAD_DATA = 10
     CMD_VERIFY_CRC = 11
     CMD_I2C_PROBE = 12
-    NUM_CMD_CALLBACKS = 13
+    CMD_ARM = 13
+    CMD_EXECUTE = 14
+    NUM_CMD_CALLBACKS = 15
 
 
 # Path to File: interfaces/obc_gs_interface/commands/obc_gs_commands_response.h
@@ -393,6 +415,43 @@ def create_cmd_i2c_probe(unixtime_of_execution: int | None = None) -> CmdMsg:
     return cmd_msg
 
 
+def create_cmd_arm(cmd_arm_data: c_uint, cmd_arm_id_data: c_uint32, unixtime_of_execution: int | None = None) -> CmdMsg:
+    """
+    Function to create a CmdMsg structure for CMD_ARM
+
+    :param cmdArm: The ARM command callback struct ID
+    :param armId: The ID value associated to an arm command sent by an operator
+    :param unixtime_of_execution: A time of when to execute a certain event,
+                                  by default, it is set to None (i.e. a specific
+                                  time is not needed)
+    :return: CmdMsg structure for CMD_ARM
+    """
+    cmd_msg = CmdMsg(unixtime_of_execution)
+    cmd_msg.id = CmdCallbackId.CMD_ARM
+    cmd_msg.cmdArm.cmd_arm_data = c_uint(cmd_arm_data.value)
+    cmd_msg.cmdArm.cmd_arm_id_data = c_uint32(cmd_arm_id_data.value)
+    return cmd_msg
+
+
+def create_cmd_execute(
+    cmd_execute_data: c_uint, cmd_exec_id_data: c_uint32, unixtime_of_execution: int | None = None
+) -> CmdMsg:
+    """
+    Function to create a CmdMsg structure for CMD_EXECUTE
+    :param cmdExecute: The EXECUTE command callback struct ID
+    :param execId: The ID value associated to an execute command sent by an operator
+    :param unixtime_of_execution: A time of when to execute a certain event,
+                                  by default, it is set to None (i.e. a specific
+                                  time is not needed)
+    :return: CmdMsg structure for CMD_EXECUTE
+    """
+    cmd_msg = CmdMsg(unixtime_of_execution)
+    cmd_msg.id = CmdCallbackId.CMD_EXECUTE
+    cmd_msg.cmdExecute.cmd_execute_data = c_uint(cmd_execute_data.value)
+    cmd_msg.execId.cmd_exec_id_data = c_uint32(cmd_exec_id_data.value)
+    return cmd_msg
+
+
 # ######################################################################
 # ||                                                                  ||
 # ||             Command Pack and Unpack Implementations              ||

interfaces/obc_gs_interface/commands/command_response_callbacks.py

@@ -3,6 +3,8 @@
 
 from interfaces.obc_gs_interface.commands import CmdCallbackId, unpack_command_response
 from interfaces.obc_gs_interface.commands.command_response_classes import (
+    CmdArmRes,
+    CmdExecuteRes,
     CmdI2CProbeRes,
     CmdRes,
     CmdRtcSyncRes,
@@ -73,12 +75,58 @@ def parse_cmd_i2c_probe(cmd_response: CmdRes, data: bytes) -> CmdI2CProbeRes:
     return CmdI2CProbeRes(cmd_response.cmd_id, cmd_response.error_code, cmd_response.response_length, valid_addresses)
 
 
+def parse_cmd_arm(cmd_response: CmdRes, data: bytes) -> CmdArmRes:
+    """
+    A function to parse the raw data from the response of CMD_ARM
+    :param cmd_response: Basic command response
+    :param data: The raw bytes containing the data that needs to be parsed
+    :return: CmdArmRes (i.e. A command response with no data for CMD_ARM)
+    """
+    if cmd_response.cmd_id != CmdCallbackId.CMD_ARM:
+        raise ValueError("Wrong command id for parsing the name command")
+
+    cmd_arm_data = int.from_bytes(data[:4], "little")
+    cmd_arm_id_data = int.from_bytes(data[4:9], "little")
+
+    return CmdArmRes(
+        cmd_response.cmd_id,
+        cmd_response.error_code,
+        cmd_response.response_length,
+        cmd_arm_data,
+        cmd_arm_id_data,
+    )
+
+
+def parse_cmd_execute(cmd_response: CmdRes, data: bytes) -> CmdExecuteRes:
+    """
+    A function to parse the raw data from the response of CMD_EXECUTE
+    :param cmd_response: Basic command response
+    :param data: The raw bytes containing the data that needs to be parsed
+    :return: CmdExecuteRes (i.e. A command response with no data for CMD_EXECUTE)
+    """
+    if cmd_response.cmd_id != CmdCallbackId.CMD_EXECUTE:
+        raise ValueError("Wrong command id for parsing the name command")
+
+    cmd_execute_data = int.from_bytes(data[:4], "little")
+    cmd_exec_id_data = int.from_bytes(data[4:9], "little")
+
+    return CmdExecuteRes(
+        cmd_response.cmd_id,
+        cmd_response.error_code,
+        cmd_response.response_length,
+        cmd_execute_data,
+        cmd_exec_id_data,
+    )
+
+
 # Function array where each index corresponds to the command enum value + 1
 
 parse_func_dict: dict[CmdCallbackId, Callable[..., CmdRes]] = defaultdict(lambda: parse_cmd_with_no_data)
 parse_func_dict[CmdCallbackId.CMD_VERIFY_CRC] = parse_cmd_verify_crc
 parse_func_dict[CmdCallbackId.CMD_RTC_SYNC] = parse_cmd_rtc_sync
 parse_func_dict[CmdCallbackId.CMD_I2C_PROBE] = parse_cmd_i2c_probe
+parse_func_dict[CmdCallbackId.CMD_ARM] = parse_cmd_arm
+parse_func_dict[CmdCallbackId.CMD_EXECUTE] = parse_cmd_execute
 
 
 def parse_command_response(data: bytes) -> CmdRes:

interfaces/obc_gs_interface/commands/command_response_classes.py

@@ -100,6 +100,44 @@ def __str__(self) -> str:
         return formatted_string
 
 
+@dataclass
+class CmdArmRes(CmdRes):
+    """
+    Class for storing the response to CMD_ARM
+    """
+
+    cmd_arm: int
+    cmd_arm_id: int
+
+    def __str__(self) -> str:
+        """
+        Overriding the str method for a better representation of what's happening
+        """
+        formatted_string = super().__str__()
+        formatted_string += "Cmd Arm: " + str(self.cmd_arm) + "\n"
+        formatted_string += "Cmd Arm Id:" + str(self.cmd_arm_id) + "\n"
+        return formatted_string
+
+
+@dataclass
+class CmdExecuteRes(CmdRes):
+    """
+    Class for storing the response to CMD_EXECUTE
+    """
+
+    cmd_execute: int
+    cmd_exec_id: int
+
+    def __str__(self) -> str:
+        """
+        Overriding the str method for a better representation of what's happening
+        """
+        formatted_string = super().__str__()
+        formatted_string += "Cmd Execute: " + str(self.cmd_execute) + "\n"
+        formatted_string += "Cmd Execute Id:" + str(self.cmd_exec_id) + "\n"
+        return formatted_string
+
+
 if __name__ == "__main__":
     cmd = CmdVerifyCrcRes(CmdCallbackId.CMD_VERIFY_CRC, CmdResponseErrorCode.CMD_RESPONSE_ERROR, 4, 0x12345678)
     print(cmd)

interfaces/obc_gs_interface/commands/obc_gs_command_data.h

@@ -17,6 +17,18 @@ typedef struct {
   uint8_t logLevel;
 } downlink_logs_next_pass_cmd_data_t;
 
+// CMD_ARM
+typedef struct {
+  cmd_callback_id_t cmdArmData;
+  uint32_t armIdData;
+} cmd_arm_cmd_data_t;
+
+// CMD_EXECUTE
+typedef struct {
+  cmd_callback_id_t cmdExecuteData;
+  uint32_t execIdData;
+} cmd_execute_cmd_data_t;
+
 /* -------------------------- */
 /* BL Command Data Structures */
 /* -------------------------- */
@@ -47,6 +59,8 @@ typedef struct {
     downlink_logs_next_pass_cmd_data_t downlinkLogsNextPass;
     download_data_cmd_data_t downloadData;
     set_programming_session_cmd_data_t setProgrammingSession;
+    cmd_arm_cmd_data_t cmdArm;
+    cmd_execute_cmd_data_t cmdExecute;
   };
 
   uint32_t timestamp;  // Unix timestamp in seconds

interfaces/obc_gs_interface/commands/obc_gs_command_id.h

@@ -49,5 +49,7 @@ typedef enum {
   CMD_DOWNLOAD_DATA,
   CMD_VERIFY_CRC,
   CMD_I2C_PROBE,
+  CMD_ARM,
+  CMD_EXECUTE,
   NUM_CMD_CALLBACKS
 } cmd_callback_id_t;

interfaces/obc_gs_interface/commands/obc_gs_command_pack.c

@@ -46,6 +46,12 @@ static void packVerifyCrcCmdData(uint8_t* buffer, uint32_t* offset, const cmd_ms
 // CMD_I2C_PROBE
 static void packI2CProbeCmdData(uint8_t* buffer, uint32_t* offset, const cmd_msg_t* msg);
 
+// CMD_ARM
+static void packCmdArmCmdData(uint8_t* buffer, uint32_t* offset, const cmd_msg_t* msg);
+
+// CMD_EXECUTE
+static void packCmdExecuteCmdData(uint8_t* buffer, uint32_t* offset, const cmd_msg_t* msg);
+
 typedef void (*pack_func_t)(uint8_t*, uint32_t*, const cmd_msg_t*);
 
 static const pack_func_t packFns[] = {
@@ -61,6 +67,8 @@ static const pack_func_t packFns[] = {
     [CMD_ERASE_APP] = packEraseAppCmdData,
     [CMD_VERIFY_CRC] = packVerifyCrcCmdData,
     [CMD_I2C_PROBE] = packI2CProbeCmdData,
+    [CMD_ARM] = packCmdArmCmdData,
+    [CMD_EXECUTE] = packCmdExecuteCmdData,
     // Add more functions for other commands as needed
 };
 
@@ -153,3 +161,15 @@ static void packVerifyCrcCmdData(uint8_t* buffer, uint32_t* offset, const cmd_ms
 static void packI2CProbeCmdData(uint8_t* buffer, uint32_t* offset, const cmd_msg_t* msg) {
   // No data to pack
 }
+
+// CMD_ARM
+static void packCmdArmCmdData(uint8_t* buffer, uint32_t* offset, const cmd_msg_t* cmdMsg) {
+  packUint32(cmdMsg->cmdArm.cmdArmData, buffer, offset);
+  packUint32(cmdMsg->cmdArm.armIdData, buffer, offset);
+}
+
+// CMD_EXECUTE
+static void packCmdExecuteCmdData(uint8_t* buffer, uint32_t* offset, const cmd_msg_t* cmdMsg) {
+  packUint32(cmdMsg->cmdExecute.cmdExecuteData, buffer, offset);
+  packUint32(cmdMsg->cmdExecute.execIdData, buffer, offset);
+}

interfaces/obc_gs_interface/commands/obc_gs_command_unpack.c

@@ -49,6 +49,12 @@ static void unpackVerifyCrcCmdData(const uint8_t* buffer, uint32_t* offset, cmd_
 // CMD_I2C_PROBE
 static void unpackI2CProbeCmdData(const uint8_t* buffer, uint32_t* offset, cmd_msg_t* msg);
 
+// CMD_ARM
+static void unpackCmdArmCmdData(const uint8_t* buffer, uint32_t* offset, cmd_msg_t* msg);
+
+// CMD_EXECUTE
+static void unpackCmdExecuteCmdData(const uint8_t* buffer, uint32_t* offset, cmd_msg_t* msg);
+
 typedef void (*unpack_func_t)(const uint8_t*, uint32_t*, cmd_msg_t*);
 
 static const unpack_func_t unpackFns[] = {
@@ -64,6 +70,8 @@ static const unpack_func_t unpackFns[] = {
     [CMD_ERASE_APP] = unpackEraseAppCmdData,
     [CMD_VERIFY_CRC] = unpackVerifyCrcCmdData,
     [CMD_I2C_PROBE] = unpackI2CProbeCmdData,
+    [CMD_ARM] = unpackCmdArmCmdData,
+    [CMD_EXECUTE] = unpackCmdExecuteCmdData,
     // Add more functions for other commands as needed
 };
 
@@ -165,3 +173,15 @@ static void unpackVerifyCrcCmdData(const uint8_t* buffer, uint32_t* offset, cmd_
 static void unpackI2CProbeCmdData(const uint8_t* buffer, uint32_t* offset, cmd_msg_t* cmdMsg) {
   // No data to unpack
 }
+
+// CMD_ARM
+static void unpackCmdArmCmdData(const uint8_t* buffer, uint32_t* offset, cmd_msg_t* cmdMsg) {
+  cmdMsg->cmdArm.cmdArmData = unpackUint32(buffer, offset);
+  cmdMsg->cmdArm.armIdData = unpackUint32(buffer, offset);
+}
+
+// CMD_EXECUTE
+static void unpackCmdExecuteCmdData(const uint8_t* buffer, uint32_t* offset, cmd_msg_t* cmdMsg) {
+  cmdMsg->cmdExecute.cmdExecuteData = unpackUint32(buffer, offset);
+  cmdMsg->cmdExecute.execIdData = unpackUint32(buffer, offset);
+}