Add ground station back to pysquared repo (#311)

Co-authored-by: ineskhou <[email protected]>

Author: nateinaction

circuitpython-workspaces/flight-software/pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     "adafruit-circuitpython-neopixel==6.3.12",
     "adafruit-circuitpython-register==1.10.4",
     "adafruit-circuitpython-rfm==1.0.6",
-    "adafruit-circuitpython-tca9548a @ git+https://github.com/proveskit/Adafruit_CircuitPython_TCA9548A@1.0.0",
+    "adafruit-circuitpython-tca9548a @ git+https://github.com/proveskit/Adafruit_CircuitPython_TCA9548A@1.1.0",
     "adafruit-circuitpython-ticks==1.1.1",
     "adafruit-circuitpython-veml7700==2.1.4",
     "adafruit-circuitpython-hashlib==1.4.19",

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

@@ -95,7 +95,7 @@ def __init__(
         """
         self._error_counter: Counter = error_counter
         self._log_level: int = log_level
-        self._colorized: bool = colorized
+        self.colorized: bool = colorized
 
     def _can_print_this_level(self, level_value: int) -> bool:
         """
@@ -162,7 +162,7 @@ def _log(self, level: str, level_value: int, message: str, **kwargs) -> None:
                 with open(file, "a") as f:
                     f.write(json_output + "\n")
 
-            if self._colorized:
+            if self.colorized:
                 json_output = json_output.replace(
                     f'"level": "{level}"', f'"level": "{LogColors[level]}"'
                 )

circuitpython-workspaces/ground-station/pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "pysquared-ground-station"
+version = "2.0.0"
+description = "Ground Station Software for the PROVES Kit"
+requires-python = ">=3.13"
+dependencies = [
+    "pysquared-flight-software",
+]
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+
+[tool.setuptools.packages.find]
+where = ["src"]
+include = ["ground_station*"]

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

@@ -0,0 +1,3 @@
+"""
+PySquared Ground Station
+"""

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

@@ -0,0 +1,160 @@
+"""
+PySquared Ground Station
+"""
+
+import json
+import time
+
+import supervisor
+from pysquared.cdh import CommandDataHandler
+from pysquared.config.config import Config
+from pysquared.hardware.radio.packetizer.packet_manager import PacketManager
+from pysquared.logger import Logger
+
+
+class GroundStation:
+    """Ground Station class to manage communication with the satellite."""
+
+    def __init__(
+        self,
+        logger: Logger,
+        config: Config,
+        packet_manager: PacketManager,
+        cdh: CommandDataHandler,
+    ):
+        self._log = logger
+        self._log.colorized = True
+        self._config = config
+        self._packet_manager = packet_manager
+        self._cdh = cdh
+
+    def listen(self):
+        """Listen for incoming packets from the satellite."""
+
+        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.")
+
+    def send_receive(self):
+        """Send commands to the satellite and wait for responses."""
+
+        try:
+            cmd_selection = input(
+                """
+            ===============================
+            | Select command to send      |
+            | 1: Reset                    |
+            | 2: Change radio modulation  |
+            | 3: Send joke                |
+            ===============================
+            """
+            )
+
+            self.handle_input(cmd_selection)
+
+        except KeyboardInterrupt:
+            self._log.debug("Keyboard interrupt received, exiting send mode.")
+
+    def handle_input(self, cmd_selection):
+        """
+        Handle user input commands.
+
+        Args:
+            cmd_selection: The command selection input by the user.
+        """
+        if cmd_selection not in ["1", "2", "3"]:
+            self._log.warning("Invalid command selection. Please try again.")
+            return
+
+        message: dict[str, object] = {
+            "name": self._config.cubesat_name,
+            "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)
+
+            # 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"))
+
+            # 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 response", response=b.decode("utf-8"))
+            break
+
+    def run(self):
+        """Run the ground station interface."""
+        while True:
+            print(
+                """
+            =============================
+            |                           |
+            | WELCOME!                  |
+            | PROVESKIT Ground Station  |
+            |                           |
+            =============================
+            | Please Select Your Mode   |
+            | 'A': Listen               |
+            | 'B': Send                 |
+            =============================
+            """
+            )
+
+            device_selection = input().lower()
+
+            if device_selection not in ["a", "b"]:
+                self._log.warning("Invalid Selection. Please try again.")
+                continue
+
+            if device_selection == "a":
+                self.listen()
+            elif device_selection == "b":
+                self.send_receive()
+
+            time.sleep(1)

pyproject.toml

@@ -6,6 +6,7 @@ dependencies = [
     "pysquared-flight-software",
     "pysquared-flight-software-mocks",
     "pysquared-flight-software-unit-tests",
+    "pysquared-ground-station",
 ]
 
 [dependency-groups]
@@ -22,6 +23,7 @@ docs = [
 pysquared-flight-software = { workspace = true }
 pysquared-flight-software-mocks = { workspace = true }
 pysquared-flight-software-unit-tests = { workspace = true }
+pysquared-ground-station = { workspace = true }
 
 [tool.uv.workspace]
 members = [