""" State management for dreader application. Handles application state persistence with asyncio-based auto-save functionality. State is saved to a JSON file and includes current mode, book position, settings, etc. """ from __future__ import annotations import asyncio import json import os from dataclasses import dataclass, asdict, field from datetime import datetime from enum import Enum from pathlib import Path from typing import Optional, Dict, Any, List import tempfile import shutil class EreaderMode(Enum): """Application mode states""" LIBRARY = "library" READING = "reading" class OverlayState(Enum): """Overlay states within READING mode""" NONE = "none" TOC = "toc" # Deprecated: use NAVIGATION instead SETTINGS = "settings" BOOKMARKS = "bookmarks" # Deprecated: use NAVIGATION instead NAVIGATION = "navigation" # Unified overlay for TOC and Bookmarks @dataclass class BookState: """State for currently open book - just the path and metadata""" path: str title: str = "" author: str = "" last_read_timestamp: str = "" def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for JSON serialization""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'BookState': """Create from dictionary""" return cls( path=data['path'], title=data.get('title', ''), author=data.get('author', ''), last_read_timestamp=data.get('last_read_timestamp', '') ) @dataclass class LibraryState: """State for library view""" books_path: str = "" last_selected_index: int = 0 scan_cache: List[Dict[str, Any]] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for JSON serialization""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'LibraryState': """Create from dictionary""" return cls( books_path=data.get('books_path', ''), last_selected_index=data.get('last_selected_index', 0), scan_cache=data.get('scan_cache', []) ) @dataclass class Settings: """User settings for rendering and display""" font_scale: float = 1.0 line_spacing: int = 5 inter_block_spacing: int = 15 word_spacing: int = 0 # Default word spacing brightness: int = 8 theme: str = "day" def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for JSON serialization""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Settings': """Create from dictionary""" return cls( font_scale=data.get('font_scale', 1.0), line_spacing=data.get('line_spacing', 5), inter_block_spacing=data.get('inter_block_spacing', 15), word_spacing=data.get('word_spacing', 0), brightness=data.get('brightness', 8), theme=data.get('theme', 'day') ) @dataclass class AppState: """Complete application state""" version: str = "1.0" mode: EreaderMode = EreaderMode.LIBRARY overlay: OverlayState = OverlayState.NONE current_book: Optional[BookState] = None library: LibraryState = field(default_factory=LibraryState) settings: Settings = field(default_factory=Settings) bookmarks: Dict[str, List[str]] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for JSON serialization""" return { 'version': self.version, 'mode': self.mode.value, 'overlay': self.overlay.value, 'current_book': self.current_book.to_dict() if self.current_book else None, 'library': self.library.to_dict(), 'settings': self.settings.to_dict(), 'bookmarks': self.bookmarks } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'AppState': """Create from dictionary""" current_book = None if data.get('current_book'): current_book = BookState.from_dict(data['current_book']) return cls( version=data.get('version', '1.0'), mode=EreaderMode(data.get('mode', 'library')), overlay=OverlayState(data.get('overlay', 'none')), current_book=current_book, library=LibraryState.from_dict(data.get('library', {})), settings=Settings.from_dict(data.get('settings', {})), bookmarks=data.get('bookmarks', {}) ) class StateManager: """ Manages application state with persistence and auto-save. Features: - Load/save state to JSON file - Asyncio-based auto-save timer (every 60 seconds) - Atomic writes (write to temp file, then rename) - Backup of previous state on corruption - Thread-safe state updates """ def __init__(self, state_file: Optional[str] = None, auto_save_interval: int = 60): """ Initialize state manager. Args: state_file: Path to state file. If None, uses default location. auto_save_interval: Auto-save interval in seconds (default: 60) """ if state_file: self.state_file = Path(state_file) else: self.state_file = self._get_default_state_file() self.auto_save_interval = auto_save_interval self.state = AppState() self._dirty = False self._save_task: Optional[asyncio.Task] = None self._lock = asyncio.Lock() # Ensure state directory exists self.state_file.parent.mkdir(parents=True, exist_ok=True) @staticmethod def _get_default_state_file() -> Path: """Get default state file location based on platform""" if os.name == 'nt': # Windows config_dir = Path(os.environ.get('APPDATA', '~/.config')) else: # Linux/Mac config_dir = Path.home() / '.config' return config_dir / 'dreader' / 'state.json' def load_state(self) -> AppState: """ Load state from file. Returns: Loaded AppState, or default AppState if file doesn't exist or is corrupt """ if not self.state_file.exists(): print(f"No state file found at {self.state_file}, using defaults") return AppState() try: with open(self.state_file, 'r') as f: data = json.load(f) self.state = AppState.from_dict(data) self._dirty = False print(f"State loaded from {self.state_file}") # Clear overlay state on boot (always start without overlays) if self.state.overlay != OverlayState.NONE: print("Clearing overlay state on boot") self.state.overlay = OverlayState.NONE self._dirty = True return self.state except Exception as e: print(f"Error loading state from {self.state_file}: {e}") # Backup corrupt file backup_path = self.state_file.with_suffix('.json.backup') try: shutil.copy2(self.state_file, backup_path) print(f"Backed up corrupt state to {backup_path}") except Exception as backup_error: print(f"Failed to backup corrupt state: {backup_error}") # Return default state self.state = AppState() self._dirty = True return self.state def save_state(self, force: bool = False) -> bool: """ Save state to file (synchronous). Args: force: Save even if state is not dirty Returns: True if saved successfully, False otherwise """ if not force and not self._dirty: return True try: # Atomic write: write to temp file, then rename temp_fd, temp_path = tempfile.mkstemp( dir=self.state_file.parent, prefix='.state_', suffix='.json.tmp' ) try: with os.fdopen(temp_fd, 'w') as f: json.dump(self.state.to_dict(), f, indent=2) # Atomic rename os.replace(temp_path, self.state_file) self._dirty = False print(f"State saved to {self.state_file}") return True except Exception as e: # Clean up temp file on error try: os.unlink(temp_path) except: pass raise e except Exception as e: print(f"Error saving state: {e}") return False async def save_state_async(self, force: bool = False) -> bool: """ Save state to file (async version). Args: force: Save even if state is not dirty Returns: True if saved successfully, False otherwise """ async with self._lock: # Run sync save in executor to avoid blocking loop = asyncio.get_event_loop() return await loop.run_in_executor(None, self.save_state, force) async def _auto_save_loop(self): """Background task for automatic state saving""" while True: try: await asyncio.sleep(self.auto_save_interval) if self._dirty: print(f"Auto-saving state (interval: {self.auto_save_interval}s)") await self.save_state_async() except asyncio.CancelledError: print("Auto-save loop cancelled") break except Exception as e: print(f"Error in auto-save loop: {e}") def start_auto_save(self): """Start the auto-save background task""" if self._save_task is None or self._save_task.done(): self._save_task = asyncio.create_task(self._auto_save_loop()) print(f"Auto-save started (interval: {self.auto_save_interval}s)") async def stop_auto_save(self, save_final: bool = True): """ Stop the auto-save background task. Args: save_final: Whether to perform a final save before stopping """ if self._save_task and not self._save_task.done(): self._save_task.cancel() try: await self._save_task except asyncio.CancelledError: pass if save_final: await self.save_state_async(force=True) print("Final state save completed") # Convenience methods for state access def get_mode(self) -> EreaderMode: """Get current application mode""" return self.state.mode def set_mode(self, mode: EreaderMode): """Set application mode""" if self.state.mode != mode: self.state.mode = mode self._dirty = True def get_overlay(self) -> OverlayState: """Get current overlay state""" return self.state.overlay def set_overlay(self, overlay: OverlayState): """Set overlay state""" if self.state.overlay != overlay: self.state.overlay = overlay self._dirty = True def get_current_book(self) -> Optional[BookState]: """Get current book state""" return self.state.current_book def set_current_book(self, book: Optional[BookState]): """Set current book state""" self.state.current_book = book if book: book.last_read_timestamp = datetime.now().isoformat() self._dirty = True def update_book_timestamp(self): """Update current book's last read timestamp""" if self.state.current_book: self.state.current_book.last_read_timestamp = datetime.now().isoformat() self._dirty = True def get_settings(self) -> Settings: """Get user settings""" return self.state.settings def update_setting(self, key: str, value: Any): """Update a single setting""" if hasattr(self.state.settings, key): setattr(self.state.settings, key, value) self._dirty = True def update_settings(self, settings_dict: Dict[str, Any]): """ Update multiple settings at once. Args: settings_dict: Dictionary with setting keys and values """ for key, value in settings_dict.items(): if hasattr(self.state.settings, key): setattr(self.state.settings, key, value) self._dirty = True def get_library_state(self) -> LibraryState: """Get library state""" return self.state.library def update_library_cache(self, cache: List[Dict[str, Any]]): """Update library scan cache""" self.state.library.scan_cache = cache self._dirty = True def is_dirty(self) -> bool: """Check if state has unsaved changes""" return self._dirty def mark_dirty(self): """Mark state as having unsaved changes""" self._dirty = True