Feedback on SD Card Logging

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
         language: pygrep
         types: [python]
         files: ^pysquared/
-        exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py
+        exclude: ^pysquared/beacon\.py|^pysquared/logger\.py|^pysquared/rtc/manager/microcontroller\.py|pysquared/hardware/sd_card/manager/sd_card\.py
 
   - repo: local
     hooks:

mocks/circuitpython/storage.py

@@ -0,0 +1,26 @@
+"""Mock for the CircuitPython storage module.
+
+This module provides a mock implementation of the CircuitPython storage module
+for testing purposes. It allows for simulating the behavior of the storage
+module without the need for actual hardware.
+"""
+
+
+def disable_usb_drive() -> None:
+    """A mock function to disable the USB drive."""
+    pass
+
+
+def enable_usb_drive() -> None:
+    """A mock function to enable the USB drive."""
+    pass
+
+
+def remount(path: str, readonly: bool) -> None:
+    """A mock function to remount the filesystem.
+
+    Args:
+        path: The path to remount.
+        readonly: Whether to mount as read-only.
+    """
+    pass

pyproject.toml

@@ -45,26 +45,29 @@ docs = [
 [tool.setuptools]
 packages = [
     "pysquared",
+    "pysquared.boot",
     "pysquared.config",
-    "pysquared.hardware",
-    "pysquared.hardware.imu",
+    "pysquared.hardware.burnwire.manager",
+    "pysquared.hardware.burnwire",
     "pysquared.hardware.imu.manager",
-    "pysquared.hardware.magnetometer",
+    "pysquared.hardware.imu",
+    "pysquared.hardware.light_sensor.manager",
     "pysquared.hardware.magnetometer.manager",
-    "pysquared.hardware.radio",
+    "pysquared.hardware.magnetometer",
+    "pysquared.hardware.power_monitor.manager",
+    "pysquared.hardware.power_monitor",
     "pysquared.hardware.radio.manager",
     "pysquared.hardware.radio.packetizer",
-    "pysquared.hardware.power_monitor",
-    "pysquared.hardware.power_monitor.manager",
-    "pysquared.hardware.temperature_sensor",
+    "pysquared.hardware.radio",
+    "pysquared.hardware.sd_card.manager",
+    "pysquared.hardware.sd_card",
     "pysquared.hardware.temperature_sensor.manager",
-    "pysquared.hardware.burnwire",
-    "pysquared.hardware.burnwire.manager",
-    "pysquared.hardware.light_sensor.manager",
+    "pysquared.hardware.temperature_sensor",
+    "pysquared.hardware",
     "pysquared.nvm",
     "pysquared.protos",
-    "pysquared.rtc",
     "pysquared.rtc.manager",
+    "pysquared.rtc",
     "pysquared.sensor_reading"
 ]
 

pysquared/beacon.py

@@ -19,9 +19,9 @@
 from collections import OrderedDict
 
 try:
-    from mocks.circuitpython.microcontroller import Processor
-except ImportError:
     from microcontroller import Processor
+except ImportError:
+    from mocks.circuitpython.microcontroller import Processor
 
 from .hardware.radio.packetizer.packet_manager import PacketManager
 from .logger import Logger

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.
+"""

pysquared/boot/filesystem.py

@@ -0,0 +1,52 @@
+"""File includes utilities for managing the filesystem during the boot process."""
+
+import os
+import time
+
+try:
+    import storage
+except ImportError:
+    import mocks.circuitpython.storage as 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")

pysquared/hardware/sd_card/manager/__init__.py

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

pysquared/sd_card.py → pysquared/hardware/sd_card/manager/sd_card.py

@@ -1,17 +1,11 @@
 """This module provides a SD Card class to manipulate the sd card filesystem"""
 
-# import os
-
 import sdcardio
 import storage
+from busio import SPI
+from microcontroller import Pin
 
-from .hardware.exception import HardwareInitializationError
-
-try:
-    from busio import SPI
-    from microcontroller import Pin
-except ImportError:
-    pass
+from ...exception import HardwareInitializationError
 
 
 class SDCardManager:
@@ -26,13 +20,9 @@ def __init__(
         baudrate: int = 400000,
         mount_path: str = "/sd",
     ) -> None:
-        self.mounted = False
-        self.mount_path = mount_path
-
         try:
             sd = sdcardio.SDCard(spi_bus, chip_select, baudrate)
-            vfs = storage.VfsFat(sd)
-            storage.mount(vfs, self.mount_path)
-            self.mounted = True
-        except (OSError, ValueError) as e:
+            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

pysquared/logger.py

@@ -18,8 +18,6 @@
 import traceback
 from collections import OrderedDict
 
-import adafruit_pathlib
-
 from .nvm.counter import Counter
 
 
@@ -79,31 +77,25 @@ class LogLevel:
 class Logger:
     """Handles logging messages with different severity levels."""
 
+    _log_dir: str | None = None
+
     def __init__(
         self,
         error_counter: Counter,
-        sd_path: adafruit_pathlib.Path = None,
-        # sd_card: SDCardManager = None,
         log_level: int = LogLevel.NOTSET,
         colorized: bool = False,
     ) -> None:
         """
         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.sd_path: adafruit_pathlib.Path = sd_path
         self._log_level: int = log_level
-        self.colorized: bool = colorized
-
-        try:
-            self.sd_path = self.sd_path / "logs"
-        except TypeError as e:
-            print(f"path not set: {e}")
+        self._colorized: bool = colorized
 
     def _can_print_this_level(self, level_value: int) -> bool:
         """
@@ -162,30 +154,15 @@ 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):
-            # Write to sd card if mounted
-            if self.path:
-                if "logs" not in self.sd_path.iterdir():
-                    print("/sd/logs does not exist, creating...")
-                    os.mkdir("/sd/logs")
-
-                with open("/sd/logs/activity.log", "a") as f:
+            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:
+            if self._colorized:
                 json_output = json_output.replace(
                     f'"level": "{level}"', f'"level": "{LogColors[level]}"'
                 )
@@ -256,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