add input handling to listen function

Author: hrfarmer

proveskit_ground_station/proveskit_ground_station.py

@@ -1,4 +1,5 @@
 import json
+import supervisor
 import time
 
 from pysquared.cdh import CommandDataHandler
@@ -25,13 +26,20 @@ def __init__(
     def listen(self):
         try:
             while True:
+                if supervisor.runtime.serial_bytes_available:
+                    typed = input().strip()
+                    if typed:
+                        self.handle_input(typed)
+
                 b = self._packet_manager.listen(1)
                 if b is not None:
                     self._log.info(
                         message="Received response", response=b.decode("utf-8")
                     )
+
         except KeyboardInterrupt:
-            self._log.debug("Keyboard interrupt received, exiting listen mode.")
+            self._log.debug(
+                "Keyboard interrupt received, exiting listen mode.")
 
     def send_receive(self):
         try:
@@ -45,61 +53,66 @@ def send_receive(self):
             ===============================
             """
             )
-            if cmd_selection not in ["1", "2", "3"]:
-                self._log.warning("Invalid command selection. Please try again.")
-                return
-
-            message: dict[str, object] = {
-                "password": self._config.super_secret_code,
-            }
-
-            if cmd_selection == "1":
-                message["command"] = self._cdh.command_reset
-            elif cmd_selection == "2":
-                message["command"] = self._cdh.command_change_radio_modulation
-                modulation = input("Enter new radio modulation [FSK | LoRa]: ")
-                message["args"] = [modulation]
-            elif cmd_selection == "3":
-                message["command"] = self._cdh.command_send_joke
 
-            while True:
-                # Turn on the radio so that it captures any received packets to buffer
-                self._packet_manager.listen(1)
+            self.handle_input(cmd_selection)
 
-                # Send the message
-                self._log.info(
-                    "Sending command",
-                    cmd=message["command"],
-                    args=message.get("args", []),
-                )
-                self._packet_manager.send(json.dumps(message).encode("utf-8"))
+        except KeyboardInterrupt:
+            self._log.debug("Keyboard interrupt received, exiting send mode.")
 
-                # Listen for ACK response
-                b = self._packet_manager.listen(1)
-                if b is None:
-                    self._log.info("No response received, retrying...")
-                    continue
+    def handle_input(self, cmd_selection):
+        if cmd_selection not in ["1", "2", "3"]:
+            self._log.warning(
+                "Invalid command selection. Please try again.")
+            return
+
+        message: dict[str, object] = {
+            "password": self._config.super_secret_code,
+        }
+
+        if cmd_selection == "1":
+            message["command"] = self._cdh.command_reset
+        elif cmd_selection == "2":
+            message["command"] = self._cdh.command_change_radio_modulation
+            modulation = input("Enter new radio modulation [FSK | LoRa]: ")
+            message["args"] = [modulation]
+        elif cmd_selection == "3":
+            message["command"] = self._cdh.command_send_joke
 
-                if b != b"ACK":
-                    self._log.info(
-                        "No ACK response received, retrying...",
-                        response=b.decode("utf-8"),
-                    )
-                    continue
+        while True:
+            # Turn on the radio so that it captures any received packets to buffer
+            self._packet_manager.listen(1)
+
+            # Send the message
+            self._log.info(
+                "Sending command",
+                cmd=message["command"],
+                args=message.get("args", []),
+            )
+            self._packet_manager.send(json.dumps(message).encode("utf-8"))
 
-                self._log.info("Received ACK")
+            # Listen for ACK response
+            b = self._packet_manager.listen(1)
+            if b is None:
+                self._log.info("No response received, retrying...")
+                continue
 
-                # Now listen for the actual 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 response", response=b.decode("utf-8"))
-                break
+            self._log.info("Received ACK")
 
-        except KeyboardInterrupt:
-            self._log.debug("Keyboard interrupt received, exiting send mode.")
+            # 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 response", response=b.decode("utf-8"))
+            break
 
     def run(self):
         while True: