Add Amateur Radio Repeat Code (#316)

Author: Mikefly123

circuitpython-workspaces/flight-software/src/pysquared/cdh.py

@@ -34,6 +34,8 @@ class CommandDataHandler:
     command_change_radio_modulation: str = "change_radio_modulation"
     command_send_joke: str = "send_joke"
 
+    oscar_password: str = "Hello World!"  # Default password for OSCAR commands
+
     def __init__(
         self,
         logger: Logger,
@@ -71,6 +73,29 @@ def listen_for_commands(self, timeout: int) -> None:
 
             msg: dict[str, str] = json.loads(json_str)
 
+            # Check for OSCAR password first
+            if msg.get("password") == self.oscar_password:
+                self._log.debug("OSCAR command received", msg=msg)
+                cmd = msg.get("command")
+                if cmd is None:
+                    self._log.warning("No OSCAR command found in message", msg=msg)
+                    self._packet_manager.send(
+                        f"No OSCAR command found in message: {msg}".encode("utf-8")
+                    )
+                    return
+
+                args: list[str] = []
+                raw_args = msg.get("args")
+                if isinstance(raw_args, list):
+                    args: list[str] = raw_args
+
+                # Delay to give the ground station time to switch to listening mode
+                time.sleep(self._send_delay)
+                self._packet_manager.send_acknowledgement()
+
+                self.oscar_command(cmd, args)
+                return
+
             # If message has password field, check it
             if msg.get("password") != self._config.super_secret_code:
                 self._log.debug(
@@ -170,3 +195,33 @@ def reset(self) -> None:
         self._packet_manager.send(data="Resetting satellite".encode("utf-8"))
         microcontroller.on_next_reset(microcontroller.RunMode.NORMAL)
         microcontroller.reset()
+
+    def oscar_command(self, command: str, args: list[str]) -> None:
+        """Handles OSCAR commands.
+
+        Args:
+            command: The OSCAR command to execute.
+            args: A list of arguments for the command.
+        """
+        if command == "ping":
+            self._log.info("OSCAR ping command received. Sending pong response.")
+            self._packet_manager.send(
+                f"Pong! {self._packet_manager.get_last_rssi()}".encode("utf-8")
+            )
+
+        elif command == "repeat":
+            if len(args) < 1:
+                self._log.warning("No message specified for repeat command")
+                self._packet_manager.send(
+                    "No message specified for repeat command.".encode("utf-8")
+                )
+                return
+            repeat_message = " ".join(args)
+            self._log.info("OSCAR repeat command received. Repeating message.")
+            self._packet_manager.send(repeat_message.encode("utf-8"))
+
+        else:
+            self._log.warning("Unknown OSCAR command received", command=command)
+            self._packet_manager.send(
+                f"Unknown OSCAR command received: {command}".encode("utf-8")
+            )

circuitpython-workspaces/flight-software/src/pysquared/hardware/radio/packetizer/packet_manager.py

@@ -201,6 +201,14 @@ def send_acknowledgement(self) -> None:
         self.send(b"ACK")
         self._logger.debug("Sent acknowledgment packet")
 
+    def get_last_rssi(self) -> int:
+        """Gets the RSSI of the last received packet.
+
+        Returns:
+            The RSSI of the last received packet.
+        """
+        return self._radio.get_rssi()
+
     def _unpack_data(self, packets: list[bytes]) -> bytes:
         """Unpacks a list of packets and reassembles the original data.
 

circuitpython-workspaces/ground-station/src/ground_station/ground_station.py

@@ -58,6 +58,7 @@ def send_receive(self):
             | 1: Reset                    |
             | 2: Change radio modulation  |
             | 3: Send joke                |
+            | 4: OSCAR commands           |
             ===============================
             """
             )
@@ -74,10 +75,15 @@ def handle_input(self, cmd_selection):
         Args:
             cmd_selection: The command selection input by the user.
         """
-        if cmd_selection not in ["1", "2", "3"]:
+        if cmd_selection not in ["1", "2", "3", "4"]:
             self._log.warning("Invalid command selection. Please try again.")
             return
 
+        # Handle OSCAR commands separately
+        if cmd_selection == "4":
+            self.handle_oscar_commands()
+            return
+
         message: dict[str, object] = {
             "name": self._config.cubesat_name,
             "password": self._config.super_secret_code,
@@ -128,6 +134,79 @@ def handle_input(self, cmd_selection):
             self._log.info("Received response", response=b.decode("utf-8"))
             break
 
+    def handle_oscar_commands(self):
+        """
+        Handle OSCAR command selection and sending.
+        """
+        try:
+            oscar_selection = input(
+                """
+            ===============================
+            | Select OSCAR command        |
+            | 1: Ping                     |
+            | 2: Repeat message           |
+            ===============================
+            """
+            )
+
+            if oscar_selection not in ["1", "2"]:
+                self._log.warning("Invalid OSCAR command selection. Please try again.")
+                return
+
+            message: dict[str, object] = {
+                "password": self._cdh.oscar_password,
+            }
+
+            if oscar_selection == "1":
+                message["command"] = "ping"
+                message["args"] = []
+            elif oscar_selection == "2":
+                repeat_message = input("Enter message to repeat: ")
+                if not repeat_message.strip():
+                    self._log.warning("Empty message provided. Please try again.")
+                    return
+                message["command"] = "repeat"
+                message["args"] = repeat_message.split()
+
+            while True:
+                # Turn on the radio so that it captures any received packets to buffer
+                self._packet_manager.listen(1)
+
+                # Send the OSCAR message
+                self._log.info(
+                    "Sending OSCAR command",
+                    cmd=message["command"],
+                    args=message.get("args", []),
+                )
+                self._packet_manager.send(json.dumps(message).encode("utf-8"))
+
+                # Listen for ACK response
+                b = self._packet_manager.listen(1)
+                if b is None:
+                    self._log.info("No response received, retrying...")
+                    continue
+
+                if b != b"ACK":
+                    self._log.info(
+                        "No ACK response received, retrying...",
+                        response=b.decode("utf-8"),
+                    )
+                    continue
+
+                self._log.info("Received ACK")
+
+                # Now listen for the actual response
+                b = self._packet_manager.listen(1)
+                if b is None:
+                    self._log.info("No response received, retrying...")
+                    continue
+
+                self._log.info("Received OSCAR response", response=b.decode("utf-8"))
+                break
+
+        except KeyboardInterrupt:
+            self._log.debug("Keyboard interrupt received, exiting OSCAR mode.")
+
     def run(self):
         """Run the ground station interface."""
         while True: