🔍 Code Extractor

A file system event handler that monitors a directory for new files and automatically processes them asynchronously when detected.

Purpose

EInkFileHandler extends FileSystemEventHandler to watch for file creation events in a monitored directory. When a new supported file is created (excluding response files), it triggers asynchronous processing after a delay to ensure the file is fully written. It maintains state to prevent duplicate processing of the same file and integrates with an InputProcessor instance to handle the actual file processing logic.

Source Code

class EInkFileHandler(FileSystemEventHandler):
    """File system event handler for processing new files"""
    
    def __init__(self, processor_instance):
        self.processor = processor_instance
        self.processing_files = set()  # Track files currently being processed
        
    def on_created(self, event):
        """Handle new file creation"""
        if not event.is_directory:
            file_path = Path(event.src_path)
            
            # Check if it's a supported file type and not already being processed
            if (InputProcessor.is_supported_file(file_path) and 
                file_path not in self.processing_files and
                not file_path.name.startswith('RESPONSE_')):  # Don't process our own output files
                
                print(f"📁 New file detected: {file_path.name}")
                # Add a small delay to ensure file is fully written
                asyncio.create_task(self._delayed_process(file_path))
    
    async def _delayed_process(self, file_path: Path):
        """Process file with a small delay to ensure it's fully written"""
        await asyncio.sleep(2)  # Wait 2 seconds for file to be fully written
        
        if file_path.exists() and file_path not in self.processing_files:
            self.processing_files.add(file_path)
            try:
                await self.processor.process_file(file_path)
            finally:
                self.processing_files.discard(file_path)

Parameters

Parameter Details

processor_instance: An instance of InputProcessor (or compatible processor) that provides the process_file() coroutine method and is_supported_file() static method. This processor handles the actual file processing logic once a file is detected and validated.

Return Value

The __init__ method returns an instance of EInkFileHandler. The on_created method returns None (implicit). The _delayed_process method is an async coroutine that returns None after processing completes or if the file is already being processed.

Class Interface

Methods

Attributes

Dependencies

• asyncio
• pathlib
• watchdog
• input_processor

Required Imports

import asyncio
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from input_processor import InputProcessor

Usage Example

from pathlib import Path
import asyncio
from watchdog.observers import Observer
from input_processor import InputProcessor
from eink_file_handler import EInkFileHandler

# Create processor instance
processor = InputProcessor()

# Create file handler with processor
file_handler = EInkFileHandler(processor)

# Set up observer to watch a directory
observer = Observer()
watch_path = Path('./watched_folder')
observer.schedule(file_handler, str(watch_path), recursive=False)

# Start observing
observer.start()

try:
    # Keep running (file handler will process files automatically)
    while True:
        asyncio.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

Best Practices

• Always ensure an async event loop is running before using this handler, as it creates async tasks
• The handler includes a 2-second delay before processing to ensure files are fully written - adjust if needed for larger files
• Files starting with 'RESPONSE_' are automatically excluded to prevent processing output files
• The processing_files set prevents duplicate processing - files are tracked during processing and removed when complete
• Must be used with watchdog.observers.Observer to actually monitor the file system
• The processor_instance must implement async process_file(file_path) method and static is_supported_file(file_path) method
• Handle exceptions in the processor's process_file method to prevent handler crashes
• The handler only processes files (not directories) and only supported file types
• Files are only processed if they exist at the time of delayed processing (handles quick deletions)

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class OneDriveProcessor 53.2% similar

  OneDriveProcessor is a class that monitors a OneDrive folder for new files, processes them using an E-Ink LLM Assistant, and uploads the results back to OneDrive.

  From: /tf/active/vicechatdev/e-ink-llm/onedrive_client.py

• class RemarkableEInkProcessor 48.0% similar

  Enhanced E-Ink LLM Processor that extends EInkLLMProcessor with reMarkable Cloud integration, enabling file processing from both local directories and reMarkable Cloud storage.

  From: /tf/active/vicechatdev/e-ink-llm/remarkable_processor.py

• class RemarkableRestFileWatcher 45.6% similar

  A file watcher class that monitors a specific folder on a reMarkable tablet using the REST API, polling for new files at regular intervals and triggering callbacks when new files are detected.

  From: /tf/active/vicechatdev/e-ink-llm/remarkable_rest_client.py

• function run_demo 44.3% similar

  Orchestrates a complete demonstration of an E-Ink LLM Assistant by creating three sample handwritten content files (question, instruction diagram, math problem) and processing each through an AI pipeline.

  From: /tf/active/vicechatdev/e-ink-llm/demo.py

• function main_v68 43.7% similar

  Async entry point for an E-Ink LLM Assistant that processes handwritten/drawn content using AI vision models, supporting local files, reMarkable Cloud, and OneDrive integration.

  From: /tf/active/vicechatdev/e-ink-llm/main.py