Creating Drive Lights on GPIO - LINUX ONLY!

Github repository

Drive_Lights

Description

If you are using a Raspberry PI or have a USB add on for your PC with GPIO pins (MCP2221A USB to Gpio Adapter Board) you can have LEDs to flash on disk drive read and write. I will use python for this.

This uses GPIOZERO to talk to the LEDs. It uses Fanotify to monitor reads and writes on a mount point. You can use two separate LEDs, one for read and one for write, or a dual color LED which can show two colors in a single body. I recommend two separate LEDs. I've tried a dual color LED but the amount of time the read is on verses the write, the write color gets overwhelmed.

This will need to run as root because the fanotify API is a system level call. I used the fanotify to monitor the '/' mount point. If there are other devices mounted under the root they will not be included. You could have multiple fanotify's running, one for each mount point. You could either provide a separate set of LEDs for each mount point or you could multiplex them all to use the same two LEDs.

I envision a small device with two LEDs, a green and a red, and a USB connector. The software could be provided on the device as a USB FAT32 file system. Of course the user would have to have python installed which is usually the case for most Linux distributions. I would provide a shell script the user would run which would install GPIOZERO if required and then run the python script. The python script could display a list of mount points and allow selection of one or more to monitor, All reads and writes would be multiplexed on the single pair of LEDs. But that's for another time.

The GPIOZERO is on PyPI so you can PIP install. The fanotify is a C library (libc.so.6) included in Linux. You access it in python using the ctypes.CDLL interface.

Mounting a solderless breadboard to a Raspberry PI is made very easy with a GPIO Breakout Kit.This allows you to experiment with the GPIO pins easily on a solderless breadboard. My only complaint is the length of the included ribbon cable I find a little short. I bought a 18 inch one and it seems to work fine. If you are going to use higher speed signals, like SPI or I2C then you shouldn't exceed the recommended length.

Protocol / Task	Recommended Max Length	Why?
High-Speed SPI	15 cm (6 inches)	High clock speeds (MHz) are extremely sensitive to signal "rounding" caused by cable capacitance.
I2C Communication	20 cm (8 inches)	I2C uses pull-up resistors. Long cables add capacitance that prevents the signal from returning to "high" quickly enough.
PWM (Servos/LEDs)	30 cm (12 inches)	High-frequency switching can cause electromagnetic interference (EMI) that leaks into adjacent wires in the ribbon.
Simple Digital I/O	50 cm (20 inches)	Reading buttons or blinking LEDs is less timing-sensitive, though "debouncing" becomes more critical.

I arbitrarily choose to use pin 20 for the read LED and pin 21 for the write LED. This was chosen strictly for convenience as they appear on the bottom right corner of the breadboard adaptor.

NOTE:

I want to point out that this program monitors software activity on a mount point. The actual hardware mounted on this mount point may or may not show the same activity on their own hardware activity LEDs due to buffering and other factors. If you are monitoring an SSD you might be alarmed at times by the number of writes being shown on the write LED. This is only showing you the operating systems calls to the device drivers write method. It is up to the device driver when or if a physical write takes place on the physical drive.

Curcuit

Curcuit diagram

The code

In order to address the GPIO pins we need to import from the GPIOZERO library. We will use three items from the library.

#-------------------------------------------------------------------------------------

#   Title:      Drive_Lights

#   Author:     Wiilliam Main

#   Created:    2021-05-20

#   Synopsys:   When there are GPIO pins available, flash a LED for reads and writes

#   Inputs:     Mount point to monitor default '/'

#-------------------------------------------------------------------------------------

from gpiozero import LED, Device

from gpiozero.pins.lgpio import LGPIOFactory

import os

import ctypes

import struct

import argparse

import sys

from typing import NoReturn

# Initialize the pin factory

try:

    Device.pin_factory = LGPIOFactory()

except ImportError:

    # If LGPIOFactory is not available, let gpiozero choose the best one

    pass

# Initialize LEDs once

write_led = LED(21)

read_led = LED(20)

def blink(led_device: LED):

    """

    Flash the LED for a short duration.

    Reusing the LED object avoids frequent thread creation/destruction issues.

    """

    try:

        # on_time=0.01: High for 0.01 second

        # off_time=0: No low time needed after the pulse

        # n=1: Do this only once

        # background=True: Script continues running immediately

        led_device.blink(on_time=0.01, off_time=0, n=1, background=True)

    except Exception:

        pass

# Fanotify constants from <sys/fanotify.h>

FAN_CLASS_NOTIF = 0x00000000

FAN_MARK_ADD = 0x00000001

FAN_MARK_MOUNT = 0x00000010

FAN_ACCESS = 0x00000001

FAN_MODIFY = 0x00000002

FAN_EVENT_METADATA_LEN = 24  # Size of fanotify_event_metadata

libc = ctypes.CDLL("libc.so.6")

class FanotifyMonitor:

    """Monitors a mount point for read/write events using fanotify."""

    def __init__(self, mount_path: str) -> None:

        self.mount_path: str = mount_path

        self.fd: int = -1

    def _initialize_fanotify(self) -> None:

        """Initialize the fanotify group and mark the mount point."""

        self.fd = libc.fanotify_init(FAN_CLASS_NOTIF, os.O_RDONLY)

        if self.fd < 0:

            raise OSError("Failed to initialize fanotify. Are you root?")

        mask: int = FAN_ACCESS | FAN_MODIFY

        result: int = libc.fanotify_mark(

            self.fd, FAN_MARK_ADD | FAN_MARK_MOUNT, mask, -1,

            self.mount_path.encode('utf-8')

        )

        if result < 0:

            raise OSError(f"Failed to mark mount point: {self.mount_path}")

    def run(self) -> NoReturn:

        """Read and process events in a loop."""

        self._initialize_fanotify()

        #print(f"Monitoring {self.mount_path}... Press Ctrl+C to stop.")

        try:

            while True:

                # Read event metadata from the file descriptor

                data = os.read(self.fd, 4096)

                offset = 0

                while offset + FAN_EVENT_METADATA_LEN <= len(data):

                    # Unpack header: event_len (I), vers (B), reserved (B),

                    # metadata_len (H), mask (Q), fd (i), pid (i)

                    header = struct.unpack_from("IBBHQii", data, offset)

                    event_len, _, _, _, mask, event_fd, pid = header

                    if event_fd >= 0:

                        if mask & FAN_ACCESS or mask & FAN_MODIFY:

                            if mask & FAN_MODIFY:

                                blink(write_led)           #flash write led

                            else:

                                blink(read_led)           #flash read led

                        os.close(event_fd)

                    offset += event_len

        except KeyboardInterrupt:

            sys.exit(0)

        finally:

            if self.fd >= 0:

                os.close(self.fd)

    @staticmethod

    def _get_path_from_fd(fd: int) -> str:

        """Retrieve the file path from its file descriptor via /proc."""

        try:

            return os.readlink(f"/proc/self/fd/{fd}")

        except FileNotFoundError:

            return "Unknown"

if __name__ == "__main__":

    parser = argparse.ArgumentParser(description="Monitor a mount point for read/write events using fanotify.")

    parser.add_argument(

        "mount_point",

        nargs="?",

        default="/",

        help="The mount point to monitor (default: /)"

    )

    args = parser.parse_args()

    # Initialize the FanotifyMonitor with the specified mount point

    monitor = FanotifyMonitor(args.mount_point)

    # Start the monitor

    monitor.run()

The main.py script is the core of the Drive_Lights project. Its purpose is to monitor a filesystem (like your SD card or an external drive) for read and write operations and flash physical LEDs connected to the Raspberry Pi's GPIO pins to provide a visual indicator of disk activity.

Here is a breakdown of how the code works:

---

1. Hardware Control (GPIO)

The script uses the gpiozero library to control the LEDs.

• Pin Factory: It specifically attempts to use LGPIOFactory (lines 18-23), which is the recommended backend for newer Raspberry Pi hardware (like the Pi 5) to ensure reliable pin control.
• LED Initialization: It defines two LED objects (lines 26-27):
  • Read LED: Connected to GPIO 20.
  • Write LED: Connected to GPIO 21.
• The blink function: This function (lines 29-41) triggers a very short pulse (0.01 seconds). It uses background=True so that the main monitoring loop isn't paused while the LED is flashing.

2. Kernel-Level Monitoring (Fanotify)

Instead of constantly polling files (which would be slow and resource-intensive), the script uses fanotify, a powerful Linux kernel feature.

• ctypes & libc: Since Python doesn't have a built-in high-level wrapper for fanotify, the script uses ctypes to talk directly to the C standard library (libc) (line 51).
• Initialization: The _initialize_fanotify method (lines 60-72) sets up a "fanotify group" and marks a specific mount point (like /) to be watched for two types of events:
  • FAN_ACCESS: Triggered when a file is read.
  • FAN_MODIFY: Triggered when a file is written to or modified.

3. The Event Loop (run method)

The heart of the script is the while True loop inside the FanotifyMonitor.run method (lines 80-98):

1. Reading Events: It reads raw binary data from the fanotify file descriptor. This data contains a series of metadata structures describing what happened.
2. Unpacking Metadata: It uses the struct module (line 87) to "unpack" the binary data into readable Python variables (like the event length, the event mask, and the file descriptor of the file being accessed).
3. Logic Switch:
   • If the mask contains FAN_MODIFY, it calls blink(write_led).
   • If the mask contains FAN_ACCESS, it calls blink(read_led).
4. Cleanup: It immediately closes the file descriptor (event_fd) created by the kernel for that specific event (line 96) to prevent the system from running out of available file handles.

4. Command Line Interface

The script uses argparse (lines 114-121) to allow flexibility:

• You can run it simply as sudo python main.py to monitor the root filesystem.
• Or specify a mount point, such as sudo python main.py /media/external_drive.

Summary of Execution Flow

1. Startup: Initialize GPIO pins and parse the target mount point.
2. Setup: Tell the Linux kernel: "Notify me whenever anything on this drive is read or changed."
3. Monitor: Wait for the kernel to send data.
4. Action: When data arrives, identify if it's a read or write and pulse the corresponding LED.
5. Repeat: Continue until the user stops the script with Ctrl+C.

Note: Because fanotify interacts directly with the kernel, the script must be run with root privileges (e.g., using sudo).