Add option to log to file mount SD Card (#299)

Co-authored-by: Nate Gay <[email protected]>
Co-authored-by: Nate Gay <[email protected]>

Author: asiemsen

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
         language: pygrep
         types: [python]
         files: ^circuitpython-workspaces/flight-software
-        exclude: ^circuitpython-workspaces/flight-software/src/pysquared/beacon\.py|^circuitpython-workspaces/flight-software/src/pysquared/logger\.py|^circuitpython-workspaces/flight-software/src/pysquared/rtc/manager/microcontroller\.py
+        exclude: ^circuitpython-workspaces/flight-software/src/pysquared/beacon\.py|^circuitpython-workspaces/flight-software/src/pysquared/logger\.py|^circuitpython-workspaces/flight-software/src/pysquared/rtc/manager/microcontroller\.py|^circuitpython-workspaces/flight-software/src/pysquared/hardware/sd_card/manager/sd_card\.py
 
   - repo: local
     hooks:
@@ -32,7 +32,7 @@ repos:
         entry: '# pyright:? *.*false'
         language: pygrep
         types: [python]
-        files: ^circuitpython-workspaces/flight-software
+        files: ^circuitpython-workspaces/flight-software|^cpython-workspace/
 
   - repo: https://github.com/codespell-project/codespell
     rev: v2.4.1

circuitpython-workspaces/flight-software/src/pysquared/boot/__init__.py

@@ -0,0 +1,3 @@
+"""
+This module provides utilities that can run during the boot process by adding them to boot.py.
+"""

circuitpython-workspaces/flight-software/src/pysquared/boot/filesystem.py

@@ -0,0 +1,49 @@
+"""File includes utilities for managing the filesystem during the boot process."""
+
+import os
+import time
+
+import storage
+
+
+def mkdir(
+    path: str,
+    storage_action_delay: float = 0.02,
+) -> None:
+    """
+    Create directories on internal storage during boot.
+
+    In CircuitPython the internal storage is not writable by default. In order to mount
+    any external storage (such as an SD Card) the drive must be remounted in read/write mode.
+    This function handles the necessary steps to safely create a directory on the internal
+    storage during boot.
+
+    Args:
+        mount_point: Path to mount point
+        storage_action_delay: Delay after storage actions to ensure stability
+
+    Usage:
+        ```python
+        from pysquared.boot.filesystem import mkdir
+        mkdir("/sd")
+        ```
+    """
+    try:
+        storage.disable_usb_drive()
+        time.sleep(storage_action_delay)
+        print("Disabled USB drive")
+
+        storage.remount("/", False)
+        time.sleep(storage_action_delay)
+        print("Remounted root filesystem")
+
+        try:
+            os.mkdir(path)
+            print(f"Mount point {path} created.")
+        except OSError:
+            print(f"Mount point {path} already exists.")
+
+    finally:
+        storage.enable_usb_drive()
+        time.sleep(storage_action_delay)
+        print("Enabled USB drive")

circuitpython-workspaces/flight-software/src/pysquared/hardware/sd_card/manager/__init__.py

@@ -0,0 +1,3 @@
+"""
+This module provides an interface for controlling SD cards.
+"""

circuitpython-workspaces/flight-software/src/pysquared/hardware/sd_card/manager/sd_card.py

@@ -0,0 +1,28 @@
+"""This module provides a SD Card class to manipulate the sd card filesystem"""
+
+import sdcardio
+import storage
+from busio import SPI
+from microcontroller import Pin
+
+from ...exception import HardwareInitializationError
+
+
+class SDCardManager:
+    """Class providing various functionalities related to USB and SD card operations."""
+
+    """Initializing class, remounting storage, and initializing SD card"""
+
+    def __init__(
+        self,
+        spi_bus: SPI,
+        chip_select: Pin,
+        baudrate: int = 400000,
+        mount_path: str = "/sd",
+    ) -> None:
+        try:
+            sd = sdcardio.SDCard(spi_bus, chip_select, baudrate)
+            vfs = storage.VfsFat(sd)  # type: ignore # Issue: https://github.com/adafruit/Adafruit_CircuitPython_Typing/issues/51
+            storage.mount(vfs, mount_path)
+        except Exception as e:
+            raise HardwareInitializationError("Failed to initialize SD Card") from e

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

@@ -13,6 +13,7 @@
 """
 
 import json
+import os
 import time
 import traceback
 from collections import OrderedDict
@@ -76,6 +77,8 @@ class LogLevel:
 class Logger:
     """Handles logging messages with different severity levels."""
 
+    _log_dir: str | None = None
+
     def __init__(
         self,
         error_counter: Counter,
@@ -86,13 +89,13 @@ def __init__(
         Initializes the Logger instance.
 
         Args:
-            error_counter (Counter): Counter for error occurrences.
-            log_level (int): Initial log level.
-            colorized (bool): Whether to colorize output.
+            error_counter: Counter for error occurrences.
+            log_level: Initial log level.
+            colorized: Whether to colorize output.
         """
         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:
         """
@@ -151,24 +154,19 @@ def _log(self, level: str, level_value: int, message: str, **kwargs) -> None:
 
         json_order.update(kwargs)
 
-        try:
-            json_output = json.dumps(json_order)
-        except TypeError as e:
-            json_output = json.dumps(
-                OrderedDict(
-                    [
-                        ("time", asctime),
-                        ("level", "ERROR"),
-                        ("msg", f"Failed to serialize log message: {e}"),
-                    ]
-                ),
-            )
+        json_output = json.dumps(json_order)
 
         if self._can_print_this_level(level_value):
-            if self.colorized:
+            if self._log_dir is not None:
+                file = self._log_dir + os.sep + "activity.log"
+                with open(file, "a") as f:
+                    f.write(json_output + "\n")
+
+            if self._colorized:
                 json_output = json_output.replace(
                     f'"level": "{level}"', f'"level": "{LogColors[level]}"'
                 )
+
             print(json_output)
 
     def debug(self, message: str, **kwargs: object) -> None:
@@ -235,3 +233,26 @@ def get_error_count(self) -> int:
             int: The number of errors logged.
         """
         return self._error_counter.get()
+
+    def set_log_dir(self, log_dir: str) -> None:
+        """
+        Sets the log directory for file logging.
+
+        Args:
+            log_dir (str): Directory to save log files.
+
+        Raises:
+            ValueError: If the provided path is not a valid directory.
+        """
+        try:
+            # Octal number 0o040000 is the stat mode indicating the file being stat'd is a directory
+            directory_mode: int = 0o040000
+            st_mode = os.stat(log_dir)[0]
+            if st_mode != directory_mode:
+                raise ValueError(
+                    f"Logging path must be a directory, received {st_mode}."
+                )
+        except OSError as e:
+            raise ValueError("Invalid logging path.") from e
+
+        self._log_dir = log_dir

cpython-workspaces/flight-software-unit-tests/src/unit-tests/boot/test_filesystem.py

@@ -0,0 +1,177 @@
+"""Unit tests for the filesystem class.
+
+This file contains unit tests for the `filesystem` class, which provides utilities
+for managing the filesystem during the boot process. The tests focus on verifying that
+the correct sequence of filesystem operations.
+"""
+
+import os
+import shutil
+import sys
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.modules["storage"] = MagicMock()
+from pysquared.boot.filesystem import mkdir  # noqa: E402
+
+
+@pytest.fixture(autouse=True)
+def test_dir():
+    """Sets up a temporary directory for testing and cleans it up afterwards."""
+    temp_dir = tempfile.mkdtemp()
+
+    test_dir = os.path.join(temp_dir, "mytestdir")
+    yield test_dir
+
+    shutil.rmtree(temp_dir, ignore_errors=True)
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+def test_mkdir_success(
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Test the mkdir function for successful directory creation."""
+    mkdir(test_dir, 0.02)
+
+    # Verify each expected call was made
+    mock_storage.disable_usb_drive.assert_called_once()
+    mock_storage.remount.assert_called_once_with("/", False)
+    mock_storage.enable_usb_drive.assert_called_once()
+
+    assert mock_sleep.call_count == 3
+    mock_sleep.assert_called_with(0.02)
+
+    # Verify correct print messages
+    captured = capsys.readouterr()
+    assert "Disabled USB drive" in captured.out
+    assert "Remounted root filesystem" in captured.out
+    assert f"Mount point {test_dir} created." in captured.out
+    assert "Enabled USB drive" in captured.out
+
+    # directory exists
+    assert os.path.exists(test_dir)
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+@patch("pysquared.boot.filesystem.os.mkdir")
+def test_mkdir_directory_already_exists(
+    mock_os_mkdir: MagicMock,
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Test the mkdir function when directory already exists."""
+    mock_os_mkdir.side_effect = OSError("Directory already exists")
+
+    mkdir(test_dir, 0.02)
+
+    # Verify all storage operations still performed
+    mock_storage.disable_usb_drive.assert_called_once()
+    mock_storage.remount.assert_called_once_with("/", False)
+    mock_os_mkdir.assert_called_once_with(test_dir)
+    mock_storage.enable_usb_drive.assert_called_once()
+
+    # Verify correct print messages
+    captured = capsys.readouterr()
+    assert "Disabled USB drive" in captured.out
+    assert "Remounted root filesystem" in captured.out
+    assert f"Mount point {test_dir} already exists." in captured.out
+    assert "Enabled USB drive" in captured.out
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+@patch("pysquared.boot.filesystem.os.mkdir")
+def test_mkdir_custom_delay(
+    mock_os_mkdir: MagicMock,
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+) -> None:
+    """Test the mkdir function with custom storage action delay."""
+    custom_delay = 0.05
+    mkdir(test_dir, custom_delay)
+
+    # Verify sleep was called with custom delay
+    assert mock_sleep.call_count == 3
+    mock_sleep.assert_called_with(custom_delay)
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+@patch("pysquared.boot.filesystem.os.mkdir")
+def test_mkdir_default_delay(
+    mock_os_mkdir: MagicMock,
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+) -> None:
+    """Test the mkdir function with default storage action delay."""
+    mkdir(test_dir)  # No delay parameter provided
+
+    # Verify sleep was called with default delay
+    assert mock_sleep.call_count == 3
+    mock_sleep.assert_called_with(0.02)
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+@patch("pysquared.boot.filesystem.os.mkdir")
+def test_mkdir_storage_disable_exception(
+    mock_os_mkdir: MagicMock,
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+) -> None:
+    """Test mkdir function when storage.disable_usb_drive() raises an exception."""
+    mock_storage.disable_usb_drive.side_effect = Exception("USB disable failed")
+
+    with pytest.raises(Exception, match="USB disable failed"):
+        mkdir(test_dir, 0.02)
+
+    # Verify that enable_usb_drive is still called in finally block
+    mock_storage.enable_usb_drive.assert_called_once()
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+def test_mkdir_storage_remount_exception(
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+) -> None:
+    """Test mkdir function when storage.remount() raises an exception."""
+    mock_storage.remount.side_effect = Exception("Remount failed")
+
+    with pytest.raises(Exception, match="Remount failed"):
+        mkdir(test_dir, 0.02)
+
+    # Verify that enable_usb_drive is still called in finally block
+    mock_storage.enable_usb_drive.assert_called_once()
+
+
+@patch("pysquared.boot.filesystem.storage")
+@patch("pysquared.boot.filesystem.time.sleep")
+@patch("pysquared.boot.filesystem.os.mkdir")
+def test_mkdir_mkdir_exception_not_oserror(
+    mock_os_mkdir: MagicMock,
+    mock_sleep: MagicMock,
+    mock_storage: MagicMock,
+    test_dir: str,
+) -> None:
+    """Test mkdir function when os.mkdir() raises a non-OSError exception."""
+    mock_os_mkdir.side_effect = ValueError("Invalid path")
+
+    with pytest.raises(ValueError, match="Invalid path"):
+        mkdir(test_dir, 0.02)
+
+    # Verify that enable_usb_drive is still called in finally block
+    mock_storage.enable_usb_drive.assert_called_once()