Merge pull request #2 from NCSU-High-Powered-Rocketry-Club/add-ruff

Add Ruff as a formatter and linter

.github/workflows/ruff.yml

@@ -0,0 +1,20 @@
+name: CI
+on: push
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install ruff
+      # Update output format to enable automatic inline annotations.
+      - name: Run Ruff linter
+        run: ruff check --output-format=github .
+      - name: Run Ruff formatter
+        run: ruff format --check .

Airbrakes/airbrakes.py

@@ -1,7 +1,9 @@
+"""Module which provides a high level interface to the air brakes system on the rocket."""
+
 from Airbrakes.imu import IMU, IMUDataPacket
 from Airbrakes.logger import Logger
 from Airbrakes.servo import Servo
-from state import *
+from Airbrakes.state import StandByState, State
 
 
 class AirbrakesContext:
@@ -13,13 +15,13 @@ class AirbrakesContext:
     """
 
     __slots__ = (
+        "current_extension",
+        "current_imu_data",
+        "imu",
         "logger",
         "servo",
-        "imu",
-        "state",
         "shutdown_requested",
-        "current_extension",
-        "current_imu_data",
+        "state",
     )
 
     def __init__(self, logger: Logger, servo: Servo, imu: IMU):

Airbrakes/imu.py

@@ -1,4 +1,7 @@
+"""Module for interacting with the IMU (Inertial measurement unit) on the rocket."""
+
 import multiprocessing
+
 import mscl
 
 
@@ -9,24 +12,24 @@ class IMUDataPacket:
     """
 
     __slots__ = (
-        "timestamp",
         "acceleration",
-        "velocity",
         "altitude",
-        "yaw",
         "pitch",
         "roll",
+        "timestamp",
+        "velocity",
+        "yaw",
     )
 
     def __init__(
         self,
         timestamp: float,
-        acceleration: float = None,
-        velocity: float = None,
-        altitude: float = None,
-        yaw: float = None,
-        pitch: float = None,
-        roll: float = None,
+        acceleration: float | None = None,
+        velocity: float | None = None,
+        altitude: float | None = None,
+        yaw: float | None = None,
+        pitch: float | None = None,
+        roll: float | None = None,
     ):
         self.timestamp = timestamp
         self.acceleration = acceleration
@@ -59,13 +62,10 @@ class IMU:
     RAW_DESCRIPTOR_SET = 128
 
     __slots__ = (
-        "port",
-        "frequency",
-        "upside_down",
         "connection",
+        "data_fetch_process",
         "latest_data",
         "running",
-        "data_fetch_process",
     )
 
     def __init__(self, port: str, frequency: int, upside_down: bool):
@@ -74,11 +74,12 @@ def __init__(self, port: str, frequency: int, upside_down: bool):
         self.running = multiprocessing.Value("b", True)  # Makes a boolean value that is shared between processes
 
         # Starts the process that fetches data from the IMU
-        self.data_fetch_process = multiprocessing.Process(target=self._fetch_data_loop, args=(port, frequency,
-                                                                                              upside_down))
+        self.data_fetch_process = multiprocessing.Process(
+            target=self._fetch_data_loop, args=(port, frequency, upside_down)
+        )
         self.data_fetch_process.start()
 
-    def _fetch_data_loop(self, port: str, frequency: int, upside_down: bool):
+    def _fetch_data_loop(self, port: str, frequency: int, _: bool):
         """
         This is the loop that fetches data from the IMU. It runs in parallel with the main loop.
         """
@@ -101,7 +102,6 @@ def _fetch_data_loop(self, port: str, frequency: int, upside_down: bool):
                         # 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
                         if packet.descriptorSet() == self.ESTIMATED_DESCRIPTOR_SET:
-                            print(data_point.channelName())
                             match channel:
                                 case "estPressureAlt":
                                     self.latest_data["altitude"] = data_point.as_float()

Airbrakes/logger.py

@@ -1,6 +1,9 @@
+"""Module for logging data to a CSV file in real time."""
+
 import logging
-import os
 import multiprocessing
+from pathlib import Path
+
 from Airbrakes.imu import IMUDataPacket
 
 
@@ -14,21 +17,22 @@ class Logger:
     real time.
     """
 
-    __slots__ = ("csv_headers", "log_path", "log_queue", "running", "log_process")
+    __slots__ = ("csv_headers", "log_path", "log_process", "log_queue", "running")
 
     def __init__(self, csv_headers: list[str]):
         self.csv_headers = csv_headers
 
-        self.log_path = os.path.join("logs", "log1.csv")
+        log_dir = Path("logs")
+        log_dir.mkdir(parents=True, exist_ok=True)
+
+        # Get all existing log files and find the highest suffix number
+        existing_logs = list(log_dir.glob("log_*.csv"))
+        max_suffix = max(int(log.stem.split("_")[-1]) for log in existing_logs) if existing_logs else 0
 
-        # Creates a new CSV file with the specified headers
-        while True:
-            if not os.path.exists(self.log_path):
-                with open(self.log_path, "w", newline="") as file_writer:
-                    file_writer.write(",".join(csv_headers) + "\n")
-                    break
-            # If the file already exists, increment the number at the end of the file name
-            self.log_path = self.log_path[:8] + str(int(self.log_path[8:-4]) + 1) + ".csv"
+        # Create a new log file with the next number in sequence
+        self.log_path = log_dir / f"log_{max_suffix + 1}.csv"
+        with self.log_path.open(mode="w", newline="") as file_writer:
+            file_writer.write(",".join(csv_headers) + "\n")
 
         # Makes a queue to store log messages, basically it's a process-safe list that you add to the back and pop from
         # front, meaning that things will be logged in the order they were added

Airbrakes/servo.py

@@ -1,3 +1,5 @@
+"""Module which contains the Servo class, representing a servo motor that controls the extension of the airbrakes."""
+
 import gpiozero
 
 
@@ -6,7 +8,7 @@ class Servo:
     A custom class that represents a servo motor, controlling the extension of the airbrakes.
     """
 
-    __slots__ = ("min_extension", "max_extension", "servo", "current_extension")
+    __slots__ = ("current_extension", "max_extension", "min_extension", "servo")
 
     def __init__(self, gpio_pin_number: int, min_extension: float, max_extension: float):
         self.min_extension = min_extension

Airbrakes/state.py

@@ -1,3 +1,5 @@
+"""Module for the finite state machine that represents which state of flight we are in."""
+
 from Airbrakes.airbrakes import AirbrakesContext
 
 
@@ -26,14 +28,12 @@ def update(self):
         Called every loop iteration. Uses the context to interact with the hardware and decides when to move to the
         next state.
         """
-        pass
 
     def next_state(self):
         """
         We never expect/want to go back a state e.g. We're never going to go
         from Flight to Motor Burn, so this method just goes to the next state.
         """
-        pass
 
     def get_name(self):
         """

Scripts/test_logger.py

@@ -1,9 +1,9 @@
-import time
+"""Module to test the logger module."""
+
 from Airbrakes.imu import IMUDataPacket
 from Airbrakes.logger import Logger
 
-
-CSV_HEADERS = ["state", "extension"] + list(IMUDataPacket(0.0).__slots__)
+CSV_HEADERS = ["state", "extension", *list(IMUDataPacket(0.0).__slots__)]
 
 
 def main():

Scripts/test_servo.py

@@ -18,4 +18,3 @@
 print("Type (1) to deploy and (0) to retract the airbrakes.")
 while True:
     servo.set_extension(float(input()))
-

main.py

@@ -1,9 +1,11 @@
+"""The main file which will be run on the Raspberry Pi. It will create the AirbrakesContext object and run the main
+loop."""
+
 from Airbrakes.airbrakes import AirbrakesContext
 from Airbrakes.imu import IMU, IMUDataPacket
 from Airbrakes.logger import Logger
 from Airbrakes.servo import Servo
 
-
 # The pin that the servo's data wire is plugged into, in this case the GPIO 12 pin which is used for PWM
 SERVO_PIN = 12
 
@@ -21,7 +23,7 @@
 FREQUENCY = 100
 
 # The headers for the CSV file
-CSV_HEADERS = ["State", "Extension"] + list(IMUDataPacket(0.0).__slots__)
+CSV_HEADERS = ["State", "Extension", *list(IMUDataPacket(0.0).__slots__)]
 
 
 def main():

pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "AirbrakesV2"
+description = "Logic for airbrakes as a part of the NASA Student Launch Competition"
+requires-python = ">=3.10"
+version = "0.1.0"
+
+
+# RUFF:
+[tool.ruff]
+line-length = 120
+show-fixes = true
+
+[tool.ruff.lint]
+preview = true
+explicit-preview-rules = true  # TODO: Drop this when RUF022 and RUF023 are out of preview
+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", 
+          "D100", "D101", "D300", "D418", "D419", "S"]
+
+[tool.ruff.lint.per-file-ignores]
+"Scripts/test_*.py" = ["T20"]