ble-dongle: Add general purpose command

backend/command_execution.py

@@ -214,6 +214,8 @@ def receive(
             responses_received.append(commands.HeartbeatResponse(payload))
         elif command_id == CommandId.command_led_response:
             responses_received.append(commands.LEDResponse(payload))
+        elif command_id == CommandId.command_gp_response:
+            responses_received.append(commands.GPResponse(payload))
         else:
             raise RuntimeError
 

backend/commands.py

@@ -22,11 +22,15 @@ class CommandId(Enum):
     command_led_request = 0x4
     command_led_response = 0x5
 
+    command_gp_request = 0x6
+    command_gp_response = 0x7
+
 
 def get_command_id_from_name(name: str) -> CommandId:
     return {
         "log": CommandId.command_log,
         "led": CommandId.command_led_request,
+        "gp": CommandId.command_gp_request,
     }[name]
 
 
@@ -37,6 +41,7 @@ def get_request_class(
         CommandId.command_log: LogCommand,
         CommandId.command_heartbeat_request: HeartbeatRequest,
         CommandId.command_led_request: LEDRequest,
+        CommandId.command_gp_request: GPRequest,
     }[command_id]
 
 
@@ -46,6 +51,7 @@ def get_response_class(
     return {
         CommandId.command_heartbeat_response: HeartbeatResponse,
         CommandId.command_led_response: LEDResponse,
+        CommandId.command_gp_response: GPResponse,
     }[command_id]
 
 
@@ -275,3 +281,48 @@ class LEDRequest(Request):
             payload=payload,
             serial=serial,
         )
+
+
+class GPResponse(Response):
+    was_successful = True
+
+    def unpack_payload(
+        self,
+        data: bytes,
+    ):
+        self.was_successful = bool(data[0])
+        if self.was_successful:
+            self._logger.debug("GP command was successful")
+        else:
+            self._logger.debug("GP command was not successful")
+
+
+class GPRequest(Request):
+    def __init__(
+        self,
+        command_id: Union[int, str],
+    ) -> None:
+        super().__init__()
+        self.command_id = int(command_id)
+
+    @property
+    def identifier(self) -> CommandId:
+        return CommandId.command_gp_request
+
+    @property
+    def timeout(self) -> float:
+        return 0.1
+
+    def process_response(self, response: Response):
+        if not isinstance(response, GPResponse):
+            raise TypeError(f"{response} is not a {GPResponse}")
+
+    def execute(self, serial: Serial):
+        payload = pack(
+            ">B",
+            self.command_id,
+        )
+        self.send_command(
+            payload=payload,
+            serial=serial,
+        )

nucleo-wb55-dongle-ble/app/Command.cpp

@@ -13,6 +13,7 @@
 #include "commands.hpp"
 #include "HeartbeatCommand.hpp"
 #include "LedCommand.hpp"
+#include "GPCommand.hpp"
 
 std::queue<Command*> command_queue;
 
@@ -44,6 +45,9 @@ void handle_received_command(uint8_t command_id, uint8_t * payload_ptr, uint16_t
   case COMMAND_LED_REQUEST:
     push_command(new LedRequest(payload_ptr, size));
     break;
+  case COMMAND_GP_REQUEST:
+    push_command(new GPRequest(payload_ptr, size));
+    break;
 
   default:
     break;

nucleo-wb55-dongle-ble/app/Command.hpp

@@ -20,6 +20,8 @@ typedef enum : uint8_t {
     COMMAND_HEARTBEAT_RESPONSE = 0x3,
     COMMAND_LED_REQUEST = 0x4,
     COMMAND_LED_RESPONSE = 0x5,
+    COMMAND_GP_REQUEST = 0x6,
+    COMMAND_GP_RESPONSE = 0x7,
 } CommandId;
 
 

nucleo-wb55-dongle-ble/app/GPCommand.cpp

@@ -0,0 +1,47 @@
+/*
+ * GPCommand.cpp
+ *
+ *  Created on: Jul 18, 2021
+ *      Author: Andreas Berthoud
+ */
+
+#include "GPCommand.hpp"
+#include "app_ble.h"
+
+GPResponse::GPResponse(uint16_t response_identifier, bool was_successful) 
+  : Response(response_identifier) {
+  (this->payload_ptr + this->get_payload_size())[0] = (uint8_t)was_successful;
+  this->add_to_payload_size(1);
+}
+
+GPRequest::GPRequest(uint8_t * payload_ptr, uint16_t size) : Request(payload_ptr, size) {
+  uint16_t expected_length = this->buffer_offset + 1;
+  if (expected_length != size) {
+    this->has_error = true;
+    return;
+  }
+
+  uint8_t * data_ptr = payload_ptr + this->buffer_offset;
+  this->command_id = data_ptr[0];
+}
+
+GPResponse * GPRequest::execute_request(uint16_t response_identifier) {
+  bool was_successful = true;
+
+  if (this->has_error) {
+    was_successful = false;
+  } else {
+    switch (this->command_id)
+    {
+    case 1:
+      APP_BLE_Key_Button1_Action();
+      break;
+    
+    default:
+      break;
+    }
+  }
+
+  return new GPResponse(response_identifier, was_successful);
+}
+

nucleo-wb55-dongle-ble/app/GPCommand.hpp

@@ -0,0 +1,47 @@
+/*
+ * GPCommand.hpp
+ *
+ *  Created on: Jul 18, 2021
+ *      Author: Andreas Berthoud
+ */
+
+#ifndef GPCOMMAND_HPP_
+#define GPCOMMAND_HPP_
+
+#include "Request.hpp"
+#include "Response.hpp"
+
+class GPResponse : public Response {
+public:
+  GPResponse(uint16_t response_identifier, bool was_successful);
+
+  CommandId get_command_id() override { return COMMAND_GP_RESPONSE; }
+
+private:
+  bool was_successful = false;
+};
+
+class GPRequest : public Request {
+public:
+  /**
+   * @brief Construct a new general purpose request object
+   * 
+   * @param payload_ptr 
+   *    HEADER | command_id |
+   *           | 1 byte     |
+   * 
+   * @param size 
+   */
+  GPRequest(uint8_t * payload_ptr, uint16_t size);
+
+
+  GPResponse * execute_request(uint16_t response_identifier);
+
+  CommandId get_command_id() override { return COMMAND_GP_REQUEST; }
+
+private:
+  bool has_error = false;
+  int command_id = 0;
+};
+
+#endif /* GPCOMMAND_HPP_ */