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):
- Reading Events: It reads raw binary data from the
fanotify file descriptor. This data contains a series of metadata structures describing what happened. - 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). - Logic Switch:
- If the
mask contains FAN_MODIFY, it calls blink(write_led). - If the
mask contains FAN_ACCESS, it calls blink(read_led).
- 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
- Startup: Initialize GPIO pins and parse the target mount point.
- Setup: Tell the Linux kernel: "Notify me whenever anything on this drive is read or changed."
- Monitor: Wait for the kernel to send data.
- Action: When data arrives, identify if it's a read or write and pulse the corresponding LED.
- 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).
