🔍 Code Extractor

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.

Purpose

This class provides automated monitoring of a reMarkable tablet folder through REST API polling. It tracks which files have been processed, detects new files since the last check, downloads them to temporary storage, and executes user-defined callbacks for processing. The watcher runs continuously in an async loop, making it suitable for building automated workflows that respond to new documents added to a reMarkable tablet folder.

Source Code

class RemarkableRestFileWatcher:
    """File watcher using the REST API client"""
    
    def __init__(self, rest_client: RemarkableRestClient, 
                 watch_folder_name: str, poll_interval: int = 60):
        self.rest_client = rest_client
        self.watch_folder_name = watch_folder_name
        self.poll_interval = poll_interval
        self.logger = logging.getLogger(__name__)
        
        self.processed_files: Set[str] = set()
        self.last_check_time = datetime.now()
    
    async def get_new_files(self) -> List[Dict]:
        """Get list of new files since last check"""
        try:
            # Find the watch folder
            folder_id = self.rest_client.find_folder_by_name(self.watch_folder_name)
            if folder_id is None:
                self.logger.warning(f"Watch folder '{self.watch_folder_name}' not found")
                return []
            
            # Get documents in the folder
            all_files = self.rest_client.get_documents_in_folder(folder_id)
            
            # Filter for new files
            new_files = []
            for doc in all_files:
                doc_id = doc.get("ID")
                if doc_id and doc_id not in self.processed_files:
                    new_files.append(doc)
                    self.processed_files.add(doc_id)
            
            return new_files
            
        except Exception as e:
            self.logger.error(f"Error checking for new files: {e}")
            return []
    
    async def start_watching(self, callback):
        """
        Start watching for new files
        
        Args:
            callback: Async function to call with new files
        """
        print(f"👁️  Started watching reMarkable folder: {self.watch_folder_name}")
        print(f"🔄 Checking every {self.poll_interval} seconds...")
        
        # Get initial file list to mark as processed
        initial_files = await self.get_new_files()
        print(f"📁 Tracking {len(self.processed_files)} existing files")
        
        try:
            while True:
                try:
                    new_files = await self.get_new_files()
                    
                    if new_files:
                        print(f"📥 Found {len(new_files)} new file(s)")
                        
                        with tempfile.TemporaryDirectory() as temp_dir:
                            temp_path = Path(temp_dir)
                            
                            for doc in new_files:
                                doc_id = doc.get("ID")
                                doc_name = doc.get("VissibleName", "Unknown")
                                
                                print(f"📄 Processing: {doc_name}")
                                
                                # Download file
                                local_file = self.rest_client.download_document(
                                    doc_id, doc_name, temp_path
                                )
                                
                                if local_file:
                                    try:
                                        await callback(doc, local_file)
                                    except Exception as e:
                                        self.logger.error(f"Error in callback for {doc_name}: {e}")
                                        print(f"❌ Error processing {doc_name}: {e}")
                    
                    # Wait before next check
                    await asyncio.sleep(self.poll_interval)
                    
                except Exception as e:
                    self.logger.error(f"Error in watch loop: {e}")
                    print(f"❌ Watch error: {e}")
                    await asyncio.sleep(self.poll_interval)
        
        except KeyboardInterrupt:
            print(f"\n🛑 Stopping reMarkable file watcher...")
        except Exception as e:
            self.logger.error(f"Fatal error in file watcher: {e}")
            print(f"❌ Fatal watcher error: {e}")

Parameters

Parameter Details

rest_client: An instance of RemarkableRestClient that provides the REST API interface for communicating with the reMarkable tablet. This client must be properly authenticated and configured before being passed to the watcher.

watch_folder_name: The name of the folder on the reMarkable tablet to monitor for new files. This is a human-readable folder name (e.g., 'Inbox'), not a folder ID.

poll_interval: The number of seconds to wait between checks for new files. Defaults to 60 seconds. Lower values provide faster detection but increase API calls and resource usage.

Return Value

The constructor returns an instance of RemarkableRestFileWatcher. The get_new_files() method returns a List[Dict] containing document metadata dictionaries for newly detected files, with each dict containing keys like 'ID' and 'VissibleName'. The start_watching() method does not return a value but runs indefinitely until interrupted.

Class Interface

Methods

Attributes

Dependencies

• asyncio
• logging
• tempfile
• datetime
• pathlib
• typing

Required Imports

import asyncio
import logging
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Set

Usage Example

import asyncio
from pathlib import Path
from typing import Dict

# Assume rest_client is already created and authenticated
rest_client = RemarkableRestClient(device_token='your_token')

# Create the watcher
watcher = RemarkableRestFileWatcher(
    rest_client=rest_client,
    watch_folder_name='Inbox',
    poll_interval=30
)

# Define callback for processing new files
async def process_new_file(doc: Dict, local_file: Path):
    doc_name = doc.get('VissibleName', 'Unknown')
    print(f'Processing {doc_name} at {local_file}')
    # Your processing logic here
    with open(local_file, 'rb') as f:
        content = f.read()
        # Do something with the file content

# Start watching (runs indefinitely)
async def main():
    await watcher.start_watching(process_new_file)

if __name__ == '__main__':
    asyncio.run(main())

Best Practices

• Always provide an authenticated RemarkableRestClient instance before creating the watcher
• Ensure the callback function is async and handles exceptions internally to prevent the watch loop from crashing
• The watcher maintains state in processed_files set - creating a new instance will reset tracking and reprocess all files
• Files are downloaded to temporary directories that are automatically cleaned up after callback execution
• Use appropriate poll_interval values to balance responsiveness with API rate limits and resource usage
• The watcher runs indefinitely until KeyboardInterrupt or fatal error - plan for graceful shutdown in production
• Initial call to get_new_files() marks all existing files as processed to avoid reprocessing on startup
• Callback receives both document metadata (dict) and local file path (Path) for flexible processing
• The watcher continues running even if individual file processing fails, logging errors but not stopping
• Consider implementing retry logic in your callback for transient failures

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class RemarkableFileWatcher 85.5% similar

  A unified file watcher class that monitors a reMarkable Cloud folder for new files, supporting both REST API and rmcl client implementations with automatic client type detection.

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

• class RemarkableCloudWatcher 70.3% similar

  Monitors the reMarkable Cloud 'gpt_out' folder for new documents, automatically downloads them, and converts .rm (reMarkable native) files to PDF format.

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

• function main_v21 69.7% similar

  Asynchronous main function that runs a reMarkable tablet file watcher as a separate process, monitoring a specified folder for new documents, processing them, and uploading responses back to the reMarkable Cloud.

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

• class RemarkableAPIClient 59.7% similar

  Asynchronous API client for interacting with the reMarkable Cloud service, providing methods for file management, folder operations, and document synchronization.

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

• function test_remarkable_discovery 59.2% similar

  Asynchronous test function that verifies reMarkable cloud folder discovery functionality by initializing a RemarkableCloudWatcher and attempting to locate the 'gpt_out' folder.

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