@@ -0,0 +1,392 @@
|
||||
"""
|
||||
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"""
|
||||
font_scale: float = 1.0
|
||||
line_spacing: int = 5
|
||||
inter_block_spacing: int = 15
|
||||
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),
|
||||
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 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
|
||||
Reference in New Issue
Block a user