Merge master and fix conflicts

airbrakes/airbrakes.py

@@ -66,6 +66,10 @@ def update(self) -> None:
         # behind on processing
         data_packets: collections.deque[IMUDataPacket] = self.imu.get_imu_data_packets()
 
+        # This should never happen, but if it does, we want to not error out and wait for packets
+        if not data_packets:
+            return
+
         # Update the processed data with the new data packets. We only care about EstimatedDataPackets
         self.data_processor.update_data(
             [data_packet for data_packet in data_packets if isinstance(data_packet, EstimatedDataPacket)]

airbrakes/constants.py

@@ -29,6 +29,9 @@
 ESTIMATED_DESCRIPTOR_SET = 130
 RAW_DESCRIPTOR_SET = 128
 
+# The maximum size of the data queue for the packets, so we don't run into memory issues
+MAX_QUEUE_SIZE = 100000
+
 # -------------------------------------------------------
 # Orientation Configuration
 # -------------------------------------------------------

airbrakes/imu/data_processor.py

@@ -15,13 +15,18 @@ class IMUDataProcessor:
     :param data_points: A sequence of EstimatedDataPacket objects to process.
     """
 
-    __slots__ = ("_avg_accel", "_avg_accel_mag", "_max_altitude", "data_points")
+    __slots__ = ("_avg_accel", "_avg_accel_mag", "_data_points", "_max_altitude")
 
     def __init__(self, data_points: Sequence[EstimatedDataPacket]):
-        self.data_points: Sequence[EstimatedDataPacket] = data_points
         self._avg_accel: tuple[float, float, float] = (0.0, 0.0, 0.0)
         self._avg_accel_mag: float = 0.0
         self._max_altitude: float = 0.0
+        self._data_points: Sequence[EstimatedDataPacket]
+
+        if data_points:  # actually update the data on init
+            self.update_data(data_points)
+        else:
+            self._data_points: Sequence[EstimatedDataPacket] = data_points
 
     def __str__(self) -> str:
         return (
@@ -37,11 +42,13 @@ def update_data(self, data_points: Sequence[EstimatedDataPacket]) -> None:
         altitude.
         :param data_points: A sequence of EstimatedDataPacket objects to process.
         """
+        if not data_points:  # Data packets may not be EstimatedDataPacket in the beginning
+            return
         self._data_points = data_points
         a_x, a_y, a_z = self._compute_averages()
         self._avg_accel = (a_x, a_y, a_z)
         self._avg_accel_mag = (self._avg_accel[0] ** 2 + self._avg_accel[1] ** 2 + self._avg_accel[2] ** 2) ** 0.5
-        self._max_altitude = max(*(data_point.estPressureAlt for data_point in self.data_points), self._max_altitude)
+        self._max_altitude = max(*(data_point.estPressureAlt for data_point in self._data_points), self._max_altitude)
 
     def _compute_averages(self) -> tuple[float, float, float]:
         """
@@ -50,9 +57,9 @@ def _compute_averages(self) -> tuple[float, float, float]:
         """
         # calculate the average acceleration in the x, y, and z directions
         # TODO: Test what these accel values actually look like
-        x_accel = stats.fmean(data_point.estCompensatedAccelX for data_point in self.data_points)
-        y_accel = stats.fmean(data_point.estCompensatedAccelY for data_point in self.data_points)
-        z_accel = stats.fmean(data_point.estCompensatedAccelZ for data_point in self.data_points)
+        x_accel = stats.fmean(data_point.estCompensatedAccelX for data_point in self._data_points)
+        y_accel = stats.fmean(data_point.estCompensatedAccelY for data_point in self._data_points)
+        z_accel = stats.fmean(data_point.estCompensatedAccelZ for data_point in self._data_points)
         # TODO: Calculate avg velocity if that's also available
         return x_accel, y_accel, z_accel
 

airbrakes/imu/imu.py

@@ -2,10 +2,18 @@
 
 import collections
 import multiprocessing
+import warnings
+
+try:
+    import mscl
+except ImportError:
+    warnings.warn(
+        "Could not import MSCL, IMU will not work. Please see installation instructions"
+        "here: https://github.com/LORD-MicroStrain/MSCL/tree/master",
+        stacklevel=2,
+    )
 
-import mscl
-
-from airbrakes.constants import ESTIMATED_DESCRIPTOR_SET, RAW_DESCRIPTOR_SET
+from airbrakes.constants import ESTIMATED_DESCRIPTOR_SET, MAX_QUEUE_SIZE, RAW_DESCRIPTOR_SET
 from airbrakes.imu.imu_data_packet import EstimatedDataPacket, IMUDataPacket, RawDataPacket
 
 
@@ -27,8 +35,10 @@ class IMU:
     )
 
     def __init__(self, port: str, frequency: int, upside_down: bool):
-        # Shared Queue which contains the latest data from the IMU
-        self._data_queue: multiprocessing.Queue[IMUDataPacket] = multiprocessing.Queue()
+        # Shared Queue which contains the latest data from the IMU. The MAX_QUEUE_SIZE is there
+        # to prevent memory issues. Realistically, the queue size never exceeds 50 packets when
+        # it's being logged.
+        self._data_queue: multiprocessing.Queue[IMUDataPacket] = multiprocessing.Queue(MAX_QUEUE_SIZE)
         self._running = multiprocessing.Value("b", False)  # Makes a boolean value that is shared between processes
 
         # Starts the process that fetches data from the IMU
@@ -92,20 +102,19 @@ def _fetch_data_loop(self, port: str, frequency: int, _: bool):
                         channel = data_point.channelName()
                         # This cpp file was the only place I was able to find all the channel names
                         # https://github.com/LORD-MicroStrain/MSCL/blob/master/MSCL/source/mscl/MicroStrain/MIP/MipTypes.cpp
-                        # Check if the imu_data_packet has an attribute with the name of the channel
-                        if hasattr(imu_data_packet, channel):
+                        # Check if the channel name is one we want to save
+                        if hasattr(imu_data_packet, channel) or "Quaternion" in channel:
                             # First checks if the data point needs special handling, if not, just set the attribute
                             match channel:
                                 # These specific data points are matrix's rather than doubles
                                 case "estAttitudeUncertQuaternion" | "estOrientQuaternion":
+                                    # This makes a 4x1 matrix from the data point with the data as [[x], [y], [z], [w]]
                                     matrix = data_point.as_Matrix()
-                                    # Converts the [4x1] matrix to the X, Y, Z, and W of the quaternion
-                                    quaternion_tuple = tuple(matrix[i, 0] for i in range(matrix.rows()))
                                     # Sets the X, Y, Z, and W of the quaternion to the data packet object
-                                    setattr(imu_data_packet, f"{channel}X", quaternion_tuple[0])
-                                    setattr(imu_data_packet, f"{channel}Y", quaternion_tuple[1])
-                                    setattr(imu_data_packet, f"{channel}Z", quaternion_tuple[2])
-                                    setattr(imu_data_packet, f"{channel}W", quaternion_tuple[3])
+                                    setattr(imu_data_packet, f"{channel}X", matrix.as_floatAt(0, 0))
+                                    setattr(imu_data_packet, f"{channel}Y", matrix.as_floatAt(0, 1))
+                                    setattr(imu_data_packet, f"{channel}Z", matrix.as_floatAt(0, 2))
+                                    setattr(imu_data_packet, f"{channel}W", matrix.as_floatAt(0, 3))
                                 case _:
                                     # Because the attribute names in our data packet classes are the same as the channel
                                     # names, we can just set the attribute to the value of the data point.

airbrakes/logger.py

@@ -43,7 +43,7 @@ def __init__(self, log_dir: Path):
         self._log_queue: multiprocessing.Queue[dict[str, str] | str] = multiprocessing.Queue()
 
         # Start the logging process
-        self._log_process = multiprocessing.Process(target=self._logging_loop)
+        self._log_process = multiprocessing.Process(target=self._logging_loop, name="Logger")
 
     @property
     def is_running(self) -> bool:

airbrakes/servo.py

@@ -10,14 +10,17 @@ class Servo:
 
     __slots__ = ("current_extension", "max_extension", "min_extension", "servo")
 
-    def __init__(self, gpio_pin_number: int, min_extension: float, max_extension: float):
+    def __init__(self, gpio_pin_number: int, min_extension: float, max_extension: float, pin_factory=None):
         self.min_extension = min_extension
         self.max_extension = max_extension
         self.current_extension = 0.0
 
         # Sets up the servo with the specified GPIO pin number
         # For this to work, you have to run the pigpio daemon on the Raspberry Pi (sudo pigpiod)
-        gpiozero.Device.pin_factory = gpiozero.pins.pigpio.PiGPIOFactory()
+        if pin_factory is None:
+            gpiozero.Device.pin_factory = gpiozero.pins.pigpio.PiGPIOFactory()
+        else:
+            gpiozero.Device.pin_factory = pin_factory
         self.servo = gpiozero.Servo(gpio_pin_number)
 
     def set_extension(self, extension: float):

airbrakes/state.py

@@ -1,9 +1,10 @@
 """Module for the finite state machine that represents which state of flight we are in."""
 
 from abc import ABC, abstractmethod
-from typing import override
+from typing import TYPE_CHECKING, override
 
-from airbrakes.airbrakes import AirbrakesContext
+if TYPE_CHECKING:
+    from airbrakes.airbrakes import AirbrakesContext
 
 
 class State(ABC):
@@ -22,7 +23,7 @@ class State(ABC):
 
     __slots__ = ("context",)
 
-    def __init__(self, context: AirbrakesContext):
+    def __init__(self, context: "AirbrakesContext"):
         """
         :param context: The state context object that will be used to interact with the electronics
         """

pyproject.toml

@@ -4,7 +4,8 @@ description = "Logic for airbrakes as a part of the NASA Student Launch Competit
 requires-python = ">=3.10"
 version = "0.1.0"
 dependencies = [
-    "gpiozero",  # Run sudo pigpiod before running the program
+    "gpiozero",
+    "pigpio",  # Run sudo pigpiod before running the program
     # Installation instructions for the following dependencies can be found in the README:
     # "mscl" https://github.com/LORD-MicroStrain/MSCL/blob/master/BuildScripts/buildReadme_Linux.md
 ]
@@ -31,8 +32,8 @@ explicit-preview-rules = true  # TODO: Drop this when RUF022 and RUF023 are out
 ignore = ["PLR2004", "PLR0911", "PLR0912", "PLR0913", "PLR0915", "PERF203", "ISC001"]
 select = ["E", "F", "I", "PL", "UP", "RUF", "PTH", "C4", "B", "PIE", "SIM", "RET", "RSE",
           "G", "ISC", "PT", "ASYNC", "TCH", "SLOT", "PERF", "PYI", "FLY", "AIR", "RUF022",
-          "RUF023", "Q", "INP", "W", "YTT", "DTZ", "ARG", "T20", "FURB", "DOC", "TRY", 
+          "RUF023", "Q", "INP", "W", "YTT", "DTZ", "ARG", "T20", "FURB", "DOC",
           "D100", "D101", "D300", "D418", "D419", "S"]
 
 [tool.ruff.lint.per-file-ignores]
-"tests/*.py" = ["T20", "S101", "D100"]
+"tests/*.py" = ["T20", "S101", "D100", "ARG001"]

requirements.txt

@@ -1,3 +1,4 @@
 gpiozero
+pigpio
 pytest
 msgspec

scripts/run_imu.py

@@ -73,4 +73,4 @@
 logger.start()
 
 while True:
-    print(imu.get_imu_data_packet())  # noqa: T201
+    print(imu.get_imu_data_packet())

scripts/run_servo.py

@@ -9,6 +9,6 @@
 
 servo = Servo(SERVO_PIN, MIN_EXTENSION, MAX_EXTENSION)
 
-print("Type (1) to deploy and (0) to retract the airbrakes.")  # noqa: T201
+print("Type (1) to deploy and (0) to retract the airbrakes.")
 while True:
     servo.set_extension(float(input()))

tests/conftest.py

@@ -0,0 +1,28 @@
+"""Module where fixtures are shared between all test files."""
+
+from pathlib import Path
+
+import pytest
+from gpiozero.pins.mock import MockFactory, MockPWMPin
+
+from airbrakes.constants import FREQUENCY, MAX_EXTENSION, MIN_EXTENSION, PORT, SERVO_PIN, UPSIDE_DOWN
+from airbrakes.imu.imu import IMU
+from airbrakes.logger import Logger
+from airbrakes.servo import Servo
+
+LOG_PATH = Path("tests/logs")
+
+
+@pytest.fixture
+def logger():
+    return Logger(LOG_PATH)
+
+
+@pytest.fixture
+def imu():
+    return IMU(port=PORT, frequency=FREQUENCY, upside_down=UPSIDE_DOWN)
+
+
+@pytest.fixture
+def servo():
+    return Servo(SERVO_PIN, MIN_EXTENSION, MAX_EXTENSION, pin_factory=MockFactory(pin_class=MockPWMPin))

tests/test_airbrakes.py

@@ -0,0 +1,22 @@
+import pytest
+
+from airbrakes.airbrakes import AirbrakesContext
+
+
+@pytest.fixture
+def airbrakes_context(imu, logger, servo):
+    return AirbrakesContext(logger, servo, imu)
+
+
+class TestAirbrakesContext:
+    """Tests the AirbrakesContext class"""
+
+    def test_slots(self, airbrakes_context):
+        inst = airbrakes_context
+        for attr in inst.__slots__:
+            assert getattr(inst, attr, "err") != "err", f"got extra slot '{attr}'"
+
+    def test_init(self, airbrakes_context, logger, imu, servo):
+        assert airbrakes_context.logger == logger
+        assert airbrakes_context.servo == servo
+        assert airbrakes_context.imu == imu

tests/test_data_processor.py

@@ -0,0 +1,100 @@
+import math
+import random
+
+import pytest
+
+from airbrakes.imu.data_processor import IMUDataProcessor
+from airbrakes.imu.imu_data_packet import EstimatedDataPacket
+
+
+def simulate_altitude_sine_wave(n_points=1000, frequency=0.01, amplitude=100, noise_level=3, base_altitude=20):
+    """Generates a random distribution of altitudes that follow a sine wave pattern, with some
+    noise added to simulate variations in the readings.
+
+    :param n_points: The number of altitude points to generate.
+    :param frequency: The frequency of the sine wave.
+    :param amplitude: The amplitude of the sine wave.
+    :param noise_level: The standard deviation of the Gaussian noise to add.
+    :param base_altitude: The base altitude, i.e. starting altitude from sea level.
+    """
+    altitudes = []
+    for i in range(n_points):
+        # Calculate the sine wave value
+        # sine wave roughly models the altitude of the rocket
+        sine_value = amplitude * math.sin(math.pi * i / (n_points - 1))
+        # Add Gaussian noise
+        noise = random.gauss(0, noise_level)
+        # Calculate the altitude at this point
+        altitude_value = base_altitude + sine_value + noise
+        altitudes.append(altitude_value)
+    return altitudes
+
+
+@pytest.fixture
+def data_processor():
+    # list of randomly increasing altitudes up to 1000 items
+    sample_data = [
+        EstimatedDataPacket(
+            1, estCompensatedAccelX=1, estCompensatedAccelY=2, estCompensatedAccelZ=3, estPressureAlt=20
+        )
+    ]
+    return IMUDataProcessor(sample_data)
+
+
+class TestIMUDataProcessor:
+    """Tests the IMUDataProcessor class"""
+
+    def test_slots(self, data_processor):
+        inst = data_processor
+        for attr in inst.__slots__:
+            assert getattr(inst, attr, "err") != "err", f"got extra slot '{attr}'"
+
+    def test_init(self, data_processor):
+        d = IMUDataProcessor([])
+        assert d._avg_accel == (0.0, 0.0, 0.0)
+        assert d._avg_accel_mag == 0.0
+        assert d._max_altitude == 0.0
+
+        d = data_processor
+        assert d._avg_accel == (1, 2, 3)
+        assert d._avg_accel_mag == math.sqrt(1**2 + 2**2 + 3**2)
+        assert d._max_altitude == 20
+
+    def test_str(self, data_processor):
+        assert (
+            str(data_processor) == "IMUDataProcessor(avg_acceleration=(1.0, 2.0, 3.0), "
+            "avg_acceleration_mag=3.7416573867739413, max_altitude=20)"
+        )
+
+    def test_update_data(self, data_processor):
+        d = data_processor
+        d.update_data(
+            [
+                EstimatedDataPacket(
+                    1, estCompensatedAccelX=1, estCompensatedAccelY=2, estCompensatedAccelZ=3, estPressureAlt=20
+                ),
+                EstimatedDataPacket(
+                    2, estCompensatedAccelX=2, estCompensatedAccelY=3, estCompensatedAccelZ=4, estPressureAlt=30
+                ),
+            ]
+        )
+        assert d._avg_accel == (1.5, 2.5, 3.5) == d.avg_acceleration
+        assert d._avg_accel_mag == math.sqrt(1.5**2 + 2.5**2 + 3.5**2) == d.avg_acceleration_mag
+        assert d.avg_acceleration_z == 3.5
+        assert d._max_altitude == 30 == d.max_altitude
+
+    def test_max_altitude(self, data_processor):
+        """Tests whether the max altitude is correctly calculated even when alititude decreases"""
+        d = data_processor
+        altitudes = simulate_altitude_sine_wave(n_points=1000)
+        # run update_data every 10 packets, to simulate actual data processing in real time:
+        for i in range(0, len(altitudes), 10):
+            d.update_data(
+                [
+                    EstimatedDataPacket(
+                        i, estCompensatedAccelX=1, estCompensatedAccelY=2, estCompensatedAccelZ=3, estPressureAlt=alt
+                    )
+                    for alt in altitudes[i : i + 10]
+                ]
+            )
+        assert d.max_altitude == max(altitudes)