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).

Local AI Meets Social Media: Building a "Technical Journalist" CLI with Ollama and Python

run_ollama: Local AI Web-to-Mastodon Summarizer

A Python-based CLI utility that acts as a personal technical journalist: it digests web content using local LLMs (via Ollama) and helps you share the takeaways to Mastodon with zero friction.

Key Features

• Local-First Intelligence: Powered by Ollama, it lets you choose between any of your locally installed models. It even supports "Thinking" models (like DeepSeek-R1), showing you the AI's reasoning process in real-time.
• Automated Web Scraping: Using BeautifulSoup, the tool strips away headers, footers, and ads from any URL you provide, feeding only the relevant content to the LLM for analysis.
• The "Technical Journalist" Persona: It uses a specialized system prompt to ensure summaries are objective, data-driven, and focused on "the what" and "the why," bypassing marketing jargon.
• Smart Mastodon Integration: Mastodon’s 500-character limit can be tricky. This tool calculates the "weighted" length of your post (accounting for URL weights) and—if the response is too long—it automatically re-prompts the LLM to rewrite a more concise version until it fits perfectly.
• Clipboard & Workflow: Every response is automatically copied to your clipboard, making it easy to use the generated text elsewhere even if you don't post it immediately.

Tech Stack

• Ollama API: For local model orchestration.
• Requests & BeautifulSoup4: For robust web content extraction.
• Mastodon.py: For seamless API interaction.
• Pyperclip: For instant clipboard access.
• Humanize: For readable model management in the CLI.

Setup

1. Clone the repository:

   git clone https://github.com/mainmeister/run_ollama.git
   cd run_ollama

2. Install dependencies: This project uses uv for dependency management:

   uv sync

3. Configure environment variables: Create a .env file based on .env.example:

   cp .env.example .env

   Edit .env and provide your Mastodon credentials:

   MASTODON_BASE_URL=https://your.mastodon.instance
   MASTODON_ACCESS_TOKEN=your_access_token_here
   # Optional: MASTODON_VISIBILITY=public
   # Optional: OLLAMA_HOST=http://remote.host:11434
   # Optional: DISABLE_CLIPBOARD=true
   

4. Run the script:

```bash
uv run main.py
```
*Tip: Use `uv run main.py --no-clipboard` to disable automatic copying, or `uv run main.py --auto` to automatically select defaults and skip confirmation prompts.*

Usage

• Select a Model: Choose from your locally installed Ollama models. The list indicates which models have "Thinking" capabilities.
• Smart Prompting: If your clipboard contains a URL, it is automatically offered as the default prompt—just press Enter to use it.
• Summarize URL: Enter any URL to have the tool fetch, clean, and summarize its content using the "Technical Journalist" persona.
• Post to Mastodon: Review the summary and confirm if you want to post it to Mastodon. The tool handles character limits and automatic rewriting.

CLI Flags

• --no-clipboard: Disables automatic copying of responses to the clipboard.
• --auto: Skips most interactive prompts by selecting defaults (useful for semi-automated workflows).

Privacy & Security

• Local Processing: Your web content and prompts are processed by your own Ollama instance. By default, this is a local server, meaning no data leaves your machine. However, if you configure a remote OLLAMA_HOST, your data will travel over the network to that server.
• SSRF Awareness: The application includes built-in protection and awareness for Server-Side Request Forgery (SSRF). It resolves hostnames to all possible IP addresses (including IPv6) and will warn you (requiring confirmation) before fetching content from private, reserved, or loopback network ranges (e.g., your local router or local services).
• Prompt Injection Mitigation: For added security, the tool uses unique delimiters ([WEBPAGE CONTENT START/END]) and strict instructions to prevent untrusted webpage content from overriding the system's journalistic persona.
• Download Limits: To prevent resource exhaustion, the tool only downloads up to 1MB of content from any provided URL.
• Privacy Controls: Clipboard copying and Mastodon post visibility are configurable via environment variables (DISABLE_CLIPBOARD, MASTODON_VISIBILITY) or CLI flags, putting the user in control of their data.
• Environment Safety: The application includes a built-in check to warn you if your .env file containing credentials is being tracked by Git, helping you avoid accidental leaks.
• Limited Access: For maximum security, use a Mastodon "App" token with limited scopes (write:statuses only) rather than a full-access token. This ensures the application can only post updates and cannot access your private messages or account settings.
• Dependency Security: The project uses pinned dependency versions and is regularly audited for known vulnerabilities (e.g., via pip-audit) to ensure a secure and stable environment for the user.

License

MIT