407 lines
13 KiB
Python
407 lines
13 KiB
Python
"""
|
|
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"
|
|
SETTINGS = "settings"
|
|
BOOKMARKS = "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
|