🔍 Code Extractor

Unified manager for reMarkable Cloud operations that uses REST API as primary method with rmcl library as fallback, handling authentication, file operations, and folder management.

Purpose

This class provides a comprehensive interface for interacting with reMarkable Cloud storage. It abstracts away the complexity of using multiple client libraries (REST API and rmcl) by automatically selecting the best available method and falling back when needed. The class handles authentication, document upload/download, folder creation and navigation, and maintains caching for improved performance. It's designed to be resilient, attempting multiple authentication methods and providing detailed logging and user feedback throughout operations.

Source Code

class RemarkableCloudManager:
    """
    Unified reMarkable Cloud manager that uses REST API as primary method
    with rmcl library as fallback for compatibility
    """
    
    def __init__(self, config_dir: Optional[str] = None, prefer_rest_api: bool = True):
        self.config_dir = Path(config_dir) if config_dir else Path.home() / '.eink-llm'
        self.config_dir.mkdir(exist_ok=True)
        
        self.device_token_file = self.config_dir / 'remarkable_device_token'
        self.config_file = self.config_dir / 'remarkable_config.json'
        
        self.logger = logging.getLogger(__name__)
        self.authenticated = False
        self.last_sync_time = None
        self.prefer_rest_api = prefer_rest_api
        
        # Initialize the appropriate client
        self.rest_client = RemarkableRestClient(config_dir)
        self.rmcl_client = None
        
        # Try to initialize rmcl client if available and requested
        if RMCL_AVAILABLE and not prefer_rest_api:
            try:
                # rmcl initialization would go here
                self.client_type = "rmcl"
                print("🔧 Using rmcl library")
            except Exception as e:
                self.logger.warning(f"rmcl initialization failed: {e}")
                self.client_type = "rest"
                print("🔧 Falling back to REST API")
        else:
            self.client_type = "rest"
            print("🔧 Using REST API client")
        
        # Cache for folder structure and file metadata (REST API only)
        self.folder_cache: Dict[str, Dict] = {}
        self.file_cache: Dict[str, Dict] = {}
        self.cache_timestamp = None
        
    async def authenticate(self, one_time_code: Optional[str] = None) -> bool:
        """
        Authenticate with reMarkable Cloud using the selected client
        
        Args:
            one_time_code: One-time code from reMarkable account (required for first setup)
            
        Returns:
            True if authentication successful, False otherwise
        """
        try:
            # First try REST API (primary method)
            if self.client_type == "rest" or self.prefer_rest_api:
                print("🔧 Attempting authentication with REST API...")
                success = await self.rest_client.authenticate(one_time_code)
                
                if success:
                    self.authenticated = success
                    self.client_type = "rest"
                    return success
                else:
                    print("⚠️  REST API authentication failed")
                    
                    # If REST API failed and rmcl is available, try fallback
                    if RMCL_AVAILABLE:
                        print("🔄 Falling back to rmcl library...")
                        self.client_type = "rmcl"
                    else:
                        print("❌ No fallback available - rmcl library not installed")
                        print("💡 To install rmcl fallback: pip install rmcl")
                        return False
            
            # Try rmcl client (either as primary choice or fallback)
            if self.client_type == "rmcl" and RMCL_AVAILABLE:
                print("🔧 Using rmcl library for authentication...")
                
                # Check if we have a device token already
                if self.device_token_file.exists() and not one_time_code:
                    print("🔑 Using existing device token with rmcl...")
                    try:
                        # Test existing authentication
                        root = Item.get_by_id_s('')  # Get root folder
                        self.authenticated = True
                        print("✅ rmcl authentication successful!")
                        return True
                    except (AuthError, ApiError) as e:
                        print("⚠️  Existing token invalid, need fresh one-time code")
                
                # First time setup or token refresh - need one-time code
                if one_time_code:
                    print("🔐 Registering new device with rmcl...")
                    try:
                        Item.register_device(one_time_code)
                        # Test the registration worked
                        root = Item.get_by_id_s('')  # Get root folder
                        print("✅ rmcl device registered and authenticated!")
                        self.authenticated = True
                        return True
                    except Exception as e:
                        print(f"❌ rmcl registration failed: {e}")
                        return False
                
                print("❌ rmcl requires a one-time code for authentication")
                return False
                
            print("❌ No available authentication method")
            print("📝 Options:")
            print("   1. Generate a one-time code from https://my.remarkable.com/connect/desktop")
            print("   2. Install rmcl library for fallback: pip install rmcl")
            return False
                
        except Exception as e:
            self.logger.error(f"Authentication error: {e}")
            print(f"❌ Authentication error: {e}")
            
            # If we haven't tried rmcl yet and it's available, try it as last resort
            if self.client_type == "rest" and RMCL_AVAILABLE and one_time_code:
                print("🔄 Trying rmcl as last resort...")
                self.client_type = "rmcl"
                return await self.authenticate(one_time_code)
                
            return False
    
    async def get_folder_by_path(self, folder_path: str) -> Optional[Union[Dict, Folder]]:
        """
        Get a folder by its path (e.g., '/My Folder/Subfolder')
        
        Args:
            folder_path: Path to the folder, starting with '/' for root
            
        Returns:
            Folder object/dict if found, None otherwise
        """
        try:
            if self.client_type == "rest":
                # For REST API, we'll work with folder names rather than paths
                # This is a simplified implementation
                if folder_path.strip() in ['/', '']:
                    return {"ID": "", "Type": "CollectionType", "VissibleName": "Root"}
                
                # Find folder by name (simplified - could be enhanced for full path support)
                folder_name = folder_path.strip('/').split('/')[-1]
                folder_id = self.rest_client.find_folder_by_name(folder_name)
                if folder_id:
                    return {"ID": folder_id, "Type": "CollectionType", "VissibleName": folder_name}
                return None
            
            elif self.client_type == "rmcl" and RMCL_AVAILABLE:
                # Original rmcl implementation
                current_folder = Item.get_by_id_s('')  # Root folder has empty ID
                
                if folder_path.strip() in ['/', '']:
                    return current_folder
                
                # Split path and navigate through folders
                path_parts = [part for part in folder_path.strip('/').split('/') if part]
                
                for part in path_parts:
                    found = False
                    for child in current_folder.children:
                        if isinstance(child, Folder) and child.name == part:
                            current_folder = child
                            found = True
                            break
                    
                    if not found:
                        print(f"❌ Folder not found: {part} in path {folder_path}")
                        return None
                
                return current_folder
            
            return None
            
        except Exception as e:
            self.logger.error(f"Error finding folder {folder_path}: {e}")
            return None
    
    async def list_files_in_folder(self, folder_path: str, include_subfolders: bool = True) -> List[Union[Dict, Document]]:
        """
        List all PDF files in a folder and optionally its subfolders
        
        Args:
            folder_path: Path to the folder to scan
            include_subfolders: If True, scan subfolders recursively
            
        Returns:
            List of Document objects/dicts representing PDF files
        """
        try:
            if self.client_type == "rest":
                # For REST API implementation
                if folder_path.strip() in ['/', '']:
                    folder_id = ""
                else:
                    folder_name = folder_path.strip('/').split('/')[-1]
                    folder_id = self.rest_client.find_folder_by_name(folder_name)
                    if not folder_id:
                        print(f"❌ Folder not found: {folder_path}")
                        return []
                
                documents = self.rest_client.get_documents_in_folder(folder_id)
                
                # If include_subfolders is True, we'd need to implement recursive search
                # This is simplified for now
                return documents
            
            elif self.client_type == "rmcl" and RMCL_AVAILABLE:
                # Original rmcl implementation
                folder = await self.get_folder_by_path(folder_path)
                if not folder:
                    return []
                
                documents = []
                
                def collect_documents(current_folder: Folder):
                    for child in current_folder.children:
                        if isinstance(child, Document):
                            # Check if it's a PDF or has content we can process
                            documents.append(child)
                        elif isinstance(child, Folder) and include_subfolders:
                            collect_documents(child)
                
                collect_documents(folder)
                return documents
            
            return []
            
        except Exception as e:
            self.logger.error(f"Error listing files in {folder_path}: {e}")
            return []
    
    async def download_document(self, document: Union[Dict, Document], output_dir: Path) -> Optional[Path]:
        """
        Download a document from reMarkable Cloud
        
        Args:
            document: Document object/dict to download
            output_dir: Directory to save the downloaded file
            
        Returns:
            Path to downloaded file, or None if failed
        """
        try:
            if self.client_type == "rest":
                # REST API implementation
                doc_id = document.get("ID") if isinstance(document, dict) else getattr(document, 'id', None)
                doc_name = document.get("VissibleName") if isinstance(document, dict) else getattr(document, 'name', 'Unknown')
                
                if not doc_id:
                    print("❌ No document ID available")
                    return None
                
                return self.rest_client.download_document(doc_id, doc_name, output_dir)
            
            elif self.client_type == "rmcl" and RMCL_AVAILABLE:
                # Original rmcl implementation
                output_dir.mkdir(parents=True, exist_ok=True)
                
                # Generate safe filename
                safe_name = "".join(c for c in document.name if c.isalnum() or c in (' ', '-', '_')).rstrip()
                if not safe_name:
                    safe_name = f"document_{document.id[:8]}"
                
                output_path = output_dir / f"{safe_name}.pdf"
                
                # Download the document content
                # First try to get the original PDF if it exists
                try:
                    content = document.contents_s()  # Get original PDF/EPUB
                    with open(output_path, 'wb') as f:
                        f.write(content.read())
                    print(f"📥 Downloaded: {document.name} -> {output_path.name}")
                    return output_path
                    
                except Exception:
                    # If no original content, try to get the raw file
                    raw_content = document.raw_s()
                    
                    # Save as a temporary zip and try to extract PDF
                    with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as temp_file:
                        temp_file.write(raw_content.read())
                        temp_zip_path = Path(temp_file.name)
                    
                    # For now, just save the raw content - in a real implementation,
                    # you might want to extract and convert notebook files to PDF
                    output_path = output_dir / f"{safe_name}.rm"
                    temp_zip_path.rename(output_path)
                    
                    print(f"📥 Downloaded raw: {document.name} -> {output_path.name}")
                    return output_path
            
            return None
                
        except Exception as e:
            doc_name = document.get("VissibleName", "Unknown") if isinstance(document, dict) else getattr(document, 'name', 'Unknown')
            self.logger.error(f"Error downloading {doc_name}: {e}")
            print(f"❌ Failed to download {doc_name}: {e}")
            return None
    
    async def upload_file(self, file_path: str, folder_path: str = None) -> Optional[str]:
        """Upload a file to reMarkable Cloud"""
        try:
            if self.client_type == 'rest':
                return await self.rest_client.upload_file(file_path, folder_path)
            else:
                return await self.rmcl_client.upload_file(file_path, folder_path)
        except Exception as e:
            logger.error(f"Failed to upload file {file_path}: {e}")
            return None
    
    async def upload_content(self, content: bytes, filename: str, folder_path: str = None, 
                           file_type: str = "application/pdf") -> Optional[str]:
        """Upload content directly to reMarkable Cloud"""
        try:
            if self.client_type == 'rest':
                return await self.rest_client.upload_content(content, filename, folder_path, file_type)
            else:
                return await self.rmcl_client.upload_content(content, filename, folder_path, file_type)
        except Exception as e:
            logger.error(f"Failed to upload content {filename}: {e}")
            return None
    
    async def create_folder(self, folder_path: str) -> bool:
        """
        Create a folder in reMarkable Cloud
        
        Args:
            folder_path: Full path to the folder to create
            
        Returns:
            True if folder created or already exists, False otherwise
        """
        try:
            # Check if folder already exists
            existing = await self.get_folder_by_path(folder_path)
            if existing:
                print(f"📁 Folder already exists: {folder_path}")
                return True
            
            # Split path to get parent and folder name
            path_parts = [part for part in folder_path.strip('/').split('/') if part]
            if not path_parts:
                print("❌ Invalid folder path")
                return False
            
            folder_name = path_parts[-1]
            parent_path = '/' + '/'.join(path_parts[:-1]) if len(path_parts) > 1 else '/'
            
            # Get parent folder
            parent_folder = await self.get_folder_by_path(parent_path)
            if not parent_folder:
                print(f"❌ Parent folder not found: {parent_path}")
                return False
            
            # Create folder using REST API
            success = await self.rest_client.create_folder(folder_name, parent_folder.id if parent_folder.id != 'root' else None)
            
            if success:
                print(f"✅ Folder created: {folder_path}")
                # Invalidate cache
                self.cache_timestamp = None
                return True
            else:
                print(f"❌ Failed to create folder via REST API")
                return False
            
        except Exception as e:
            self.logger.error(f"Error creating folder {folder_path}: {e}")
            print(f"❌ Failed to create folder: {e}")
            return False
    
    def get_config(self) -> Dict:
        """Load configuration from file"""
        if self.config_file.exists():
            try:
                with open(self.config_file, 'r') as f:
                    return json.load(f)
            except Exception as e:
                self.logger.error(f"Error loading config: {e}")
                return {}
        return {}
    
    def save_config(self, config: Dict) -> None:
        """Save configuration to file"""
        try:
            with open(self.config_file, 'w') as f:
                json.dump(config, f, indent=2)
        except Exception as e:
            self.logger.error(f"Error saving config: {e}")

Parameters

Parameter Details

config_dir: Optional path to configuration directory where authentication tokens and settings are stored. Defaults to '~/.eink-llm' if not provided. The directory will be created if it doesn't exist.

prefer_rest_api: Boolean flag indicating whether to prefer REST API over rmcl library. Defaults to True. When True, REST API is attempted first with rmcl as fallback. When False and rmcl is available, rmcl is used as primary method.

Return Value

Instantiation returns a RemarkableCloudManager object configured with the specified settings. Key method returns: authenticate() returns bool indicating success/failure; get_folder_by_path() returns Dict or Folder object or None; list_files_in_folder() returns List of Document objects/dicts; download_document() returns Path to downloaded file or None; upload_file() and upload_content() return Optional[str] document ID; create_folder() returns bool indicating success.

Class Interface

Methods

Attributes

Dependencies

• asyncio
• os
• tempfile
• time
• uuid
• datetime
• pathlib
• typing
• logging
• json
• remarkable_rest_client
• rmcl

Required Imports

import asyncio
import os
import tempfile
import time
import uuid
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple, Union
import logging
import json
from remarkable_rest_client import RemarkableRestClient, RemarkableRestFileWatcher

Conditional/Optional Imports

These imports are only needed under specific conditions:

from rmcl import Item, Document, Folder

from rmcl.exceptions import ApiError, AuthError, DocumentNotFound, FolderNotFound

Usage Example

import asyncio
from pathlib import Path
from remarkable_cloud_manager import RemarkableCloudManager

async def main():
    # Initialize manager with REST API preference
    manager = RemarkableCloudManager(
        config_dir='~/.eink-llm',
        prefer_rest_api=True
    )
    
    # Authenticate (first time requires one-time code)
    one_time_code = 'your-code-from-remarkable'
    success = await manager.authenticate(one_time_code)
    if not success:
        print('Authentication failed')
        return
    
    # Create a folder
    await manager.create_folder('/My Documents/LLM Output')
    
    # List files in a folder
    documents = await manager.list_files_in_folder('/My Documents', include_subfolders=True)
    print(f'Found {len(documents)} documents')
    
    # Download a document
    if documents:
        output_dir = Path('./downloads')
        file_path = await manager.download_document(documents[0], output_dir)
        if file_path:
            print(f'Downloaded to {file_path}')
    
    # Upload a file
    doc_id = await manager.upload_file('document.pdf', '/My Documents')
    if doc_id:
        print(f'Uploaded with ID: {doc_id}')
    
    # Upload content directly
    with open('content.pdf', 'rb') as f:
        content = f.read()
    doc_id = await manager.upload_content(
        content,
        'my_document.pdf',
        '/My Documents',
        'application/pdf'
    )

asyncio.run(main())

Best Practices

• Always call authenticate() before performing any operations - check the return value to ensure authentication succeeded
• Use async/await pattern for all method calls as most operations are asynchronous
• Handle None returns from methods gracefully - many methods return None on failure
• For first-time setup, obtain a one-time code from https://my.remarkable.com/connect/desktop
• The manager automatically handles fallback between REST API and rmcl - monitor console output for which client is being used
• Cache is invalidated automatically on folder creation - no manual cache management needed
• Document downloads may return .rm files for notebooks without original PDFs - handle both .pdf and .rm extensions
• Folder paths should start with '/' and use forward slashes - root folder is represented as '/' or empty string
• The manager maintains state (authenticated, client_type, caches) - reuse the same instance for multiple operations
• Check self.authenticated property before operations to verify authentication status
• Use include_subfolders=True in list_files_in_folder() for recursive scanning
• Output directories for downloads are created automatically if they don't exist
• Configuration and tokens are persisted to disk - subsequent runs may not need one-time code

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class RemarkableAPIClient 79.1% 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

• class RemarkableUploadManager 75.4% similar

  Manages uploads to reMarkable cloud

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

• class RemarkableUploadManager_v1 74.8% similar

  Manages uploads to reMarkable cloud

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

• class Client 73.4% similar

  API Client for the Remarkable Cloud that handles authentication, communication, and document management with the Remarkable Cloud service.

  From: /tf/active/vicechatdev/rmcl/api.py

• class RemarkableFileWatcher 72.9% 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