big update with ok rendering
Python CI / test (push) Failing after 3m55s

This commit is contained in:
2025-08-27 22:22:54 +02:00
parent 36281be77a
commit 65ab46556f
54 changed files with 6157 additions and 438 deletions
+68 -34
View File
@@ -1,16 +1,15 @@
from __future__ import annotations
from typing import List, Tuple, Optional
from typing import List, Tuple, Optional, Union
from pyWebLayout.concrete import Page, Line, Text
from pyWebLayout.abstract import Paragraph, Word, Link
from pyWebLayout.style.concrete_style import ConcreteStyleRegistry
def paragraph_layouter(paragraph: Paragraph, page: Page, start_word: int = 0, pretext: Optional[Text] = None) -> Tuple[bool, Optional[int], Optional[Text]]:
"""
Layout a paragraph of text within a given page.
This function extracts word spacing constraints from the style system
and uses them to create properly spaced lines of text.
@@ -19,7 +18,7 @@ def paragraph_layouter(paragraph: Paragraph, page: Page, start_word: int = 0, pr
page: The page to layout the paragraph on
start_word: Index of the first word to process (for continuation)
pretext: Optional pretext from a previous hyphenated word
Returns:
Tuple of:
- bool: True if paragraph was completely laid out, False if page ran out of space
@@ -28,29 +27,36 @@ def paragraph_layouter(paragraph: Paragraph, page: Page, start_word: int = 0, pr
"""
if not paragraph.words:
return True, None, None
# Validate inputs
if start_word >= len(paragraph.words):
return True, None, None
# Get the concrete style with resolved word spacing constraints
style_registry = ConcreteStyleRegistry(page.style_resolver)
concrete_style = style_registry.get_concrete_style(paragraph.style)
# Extract word spacing constraints (min, max) for Line constructor
word_spacing_constraints = (
int(concrete_style.word_spacing_min),
int(concrete_style.word_spacing_max)
)
def create_new_line() -> Optional[Line]:
def create_new_line(word: Optional[Union[Word, Text]] = None) -> Optional[Line]:
"""Helper function to create a new line, returns None if page is full."""
if not page.can_fit_line(paragraph.line_height):
return None
y_cursor = page._current_y_offset
x_cursor = page.border_size
# Create a temporary Text object to calculate word width
if word:
temp_text = Text.from_word(word, page.draw)
word_width = temp_text.width
else:
word_width = 0
return Line(
spacing=word_spacing_constraints,
origin=(x_cursor, y_cursor),
@@ -59,38 +65,67 @@ def paragraph_layouter(paragraph: Paragraph, page: Page, start_word: int = 0, pr
font=concrete_style.create_font(),
halign=concrete_style.text_align
)
# Create initial line
current_line = create_new_line()
if not current_line:
return False, start_word, pretext
page.add_child(current_line)
page._current_y_offset += paragraph.line_height
# Track current position in paragraph
current_pretext = pretext
# Process words starting from start_word
for i, word in enumerate(paragraph.words[start_word:], start=start_word):
if current_pretext:
print(current_pretext.text)
success, overflow_text = current_line.add_word(word, current_pretext)
if success:
# Word fit successfully
current_pretext = None # Clear pretext after successful placement
if overflow_text is not None:
# If there's overflow text, we need to start a new line with it
current_pretext = overflow_text
current_line = create_new_line(overflow_text)
if not current_line:
# If we can't create a new line, return with the current state
return False, i, overflow_text
page.add_child(current_line)
page._current_y_offset += paragraph.line_height
# Continue to the next word
continue
else:
# No overflow, clear pretext
current_pretext = None
else:
# Word didn't fit, need a new line
current_line = create_new_line()
current_line = create_new_line(word)
if not current_line:
# Page is full, return current position
return False, i, overflow_text
# Check if the word will fit on the new line before adding it
temp_text = Text.from_word(word, page.draw)
if temp_text.width > current_line.size[0]:
# Word is too wide for the line, we need to hyphenate it
if len(word.text) >= 6:
# Try to hyphenate the word
splits = [(Text(pair[0], word.style, page.draw, line=current_line, source=word), Text(pair[1], word.style, page.draw, line=current_line, source=word)) for pair in word.possible_hyphenation()]
if len(splits) > 0:
# Use the first hyphenation point
first_part, second_part = splits[0]
current_line.add_word(word, first_part)
current_pretext = second_part
continue
page.add_child(current_line)
page._current_y_offset += paragraph.line_height
# Try to add the word to the new line
success, overflow_text = current_line.add_word(word, current_pretext)
if not success:
# Word still doesn't fit even on a new line
# This might happen with very long words or narrow pages
@@ -104,59 +139,58 @@ def paragraph_layouter(paragraph: Paragraph, page: Page, start_word: int = 0, pr
return False, i, None
else:
current_pretext = overflow_text # May be None or hyphenated remainder
# All words processed successfully
return True, None, None
class DocumentLayouter:
"""
Class-based document layouter for more complex layout operations.
"""
def __init__(self, page: Page):
"""Initialize the layouter with a page."""
self.page = page
self.style_registry = ConcreteStyleRegistry(page.style_resolver)
def layout_paragraph(self, paragraph: Paragraph, start_word: int = 0, pretext: Optional[Text] = None) -> Tuple[bool, Optional[int], Optional[Text]]:
"""
Layout a paragraph using the class-based approach.
This method provides the same functionality as the standalone function
but with better state management and reusability.
"""
return paragraph_layouter(paragraph, self.page, start_word, pretext)
def layout_document(self, paragraphs: List[Paragraph]) -> bool:
"""
Layout multiple paragraphs in sequence.
Args:
paragraphs: List of paragraphs to layout
Returns:
True if all paragraphs were laid out successfully, False otherwise
"""
for paragraph in paragraphs:
start_word = 0
pretext = None
while True:
complete, next_word, remaining_pretext = self.layout_paragraph(
paragraph, start_word, pretext
)
if complete:
# Paragraph finished
break
if next_word is None:
# Error condition
return False
# Continue on next page or handle page break
# For now, we'll just return False indicating we need more space
return False
return True
+450
View File
@@ -0,0 +1,450 @@
"""
Enhanced ereader layout system with position tracking, font scaling, and multi-page support.
This module provides the core infrastructure for building high-performance ereader applications
with features like:
- Precise position tracking tied to abstract document structure
- Font scaling support
- Bidirectional page rendering (forward/backward)
- Chapter navigation based on HTML headings
- Multi-process page buffering
- Sub-second page rendering performance
"""
from __future__ import annotations
from dataclasses import dataclass, asdict
from typing import List, Dict, Tuple, Optional, Union, Generator, Any
from enum import Enum
import json
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, as_completed
import threading
import time
from pyWebLayout.abstract.block import Block, Paragraph, Heading, HeadingLevel, Table, HList
from pyWebLayout.abstract.inline import Word
from pyWebLayout.concrete.page import Page
from pyWebLayout.concrete.text import Line, Text
from pyWebLayout.style.page_style import PageStyle
from pyWebLayout.style import Font
@dataclass
class RenderingPosition:
"""
Complete state for resuming rendering at any point in a document.
Position is tied to abstract document structure for stability across font changes.
"""
chapter_index: int = 0 # Which chapter (based on headings)
block_index: int = 0 # Which block within chapter
word_index: int = 0 # Which word within block (for paragraphs)
table_row: int = 0 # Which row for tables
table_col: int = 0 # Which column for tables
list_item_index: int = 0 # Which item for lists
remaining_pretext: Optional[str] = None # Hyphenated word continuation
page_y_offset: int = 0 # Vertical position on page
def to_dict(self) -> Dict[str, Any]:
"""Serialize position for saving to file/database"""
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'RenderingPosition':
"""Deserialize position from saved state"""
return cls(**data)
def copy(self) -> 'RenderingPosition':
"""Create a copy of this position"""
return RenderingPosition(**asdict(self))
def __eq__(self, other) -> bool:
"""Check if two positions are equal"""
if not isinstance(other, RenderingPosition):
return False
return asdict(self) == asdict(other)
def __hash__(self) -> int:
"""Make position hashable for use as dict key"""
return hash(tuple(asdict(self).values()))
class ChapterInfo:
"""Information about a chapter/section in the document"""
def __init__(self, title: str, level: HeadingLevel, position: RenderingPosition, block_index: int):
self.title = title
self.level = level
self.position = position
self.block_index = block_index
class ChapterNavigator:
"""
Handles chapter/section navigation based on HTML heading structure (H1-H6).
Builds a table of contents and provides navigation capabilities.
"""
def __init__(self, blocks: List[Block]):
self.blocks = blocks
self.chapters: List[ChapterInfo] = []
self._build_chapter_map()
def _build_chapter_map(self):
"""Scan blocks for headings and build chapter navigation map"""
current_chapter_index = 0
for block_index, block in enumerate(self.blocks):
if isinstance(block, Heading):
# Create position for this heading
position = RenderingPosition(
chapter_index=current_chapter_index,
block_index=0, # Heading is first block in its chapter
word_index=0,
table_row=0,
table_col=0,
list_item_index=0
)
# Extract heading text
heading_text = self._extract_heading_text(block)
chapter_info = ChapterInfo(
title=heading_text,
level=block.level,
position=position,
block_index=block_index
)
self.chapters.append(chapter_info)
# Only increment chapter index for top-level headings (H1)
if block.level == HeadingLevel.H1:
current_chapter_index += 1
def _extract_heading_text(self, heading: Heading) -> str:
"""Extract text content from a heading block"""
words = []
for word in heading.words():
if isinstance(word, Word):
words.append(word.text)
return " ".join(words)
def get_table_of_contents(self) -> List[Tuple[str, HeadingLevel, RenderingPosition]]:
"""Generate table of contents from heading structure"""
return [(chapter.title, chapter.level, chapter.position) for chapter in self.chapters]
def get_chapter_position(self, chapter_title: str) -> Optional[RenderingPosition]:
"""Get rendering position for a chapter by title"""
for chapter in self.chapters:
if chapter.title.lower() == chapter_title.lower():
return chapter.position
return None
def get_current_chapter(self, position: RenderingPosition) -> Optional[ChapterInfo]:
"""Determine which chapter contains the current position"""
if not self.chapters:
return None
# Find the chapter that contains this position
for i, chapter in enumerate(self.chapters):
# Check if this is the last chapter or if position is before next chapter
if i == len(self.chapters) - 1:
return chapter
next_chapter = self.chapters[i + 1]
if position.chapter_index < next_chapter.position.chapter_index:
return chapter
return self.chapters[0] if self.chapters else None
class FontScaler:
"""
Handles font scaling operations for ereader font size adjustments.
Applies scaling at layout/render time while preserving original font objects.
"""
@staticmethod
def scale_font(font: Font, scale_factor: float) -> Font:
"""
Create a scaled version of a font for layout calculations.
Args:
font: Original font object
scale_factor: Scaling factor (1.0 = no change, 2.0 = double size, etc.)
Returns:
New Font object with scaled size
"""
if scale_factor == 1.0:
return font
scaled_size = max(1, int(font.font_size * scale_factor))
return Font(
font_path=font._font_path,
font_size=scaled_size,
colour=font.colour,
weight=font.weight,
style=font.style,
decoration=font.decoration,
background=font.background,
language=font.language,
min_hyphenation_width=font.min_hyphenation_width
)
@staticmethod
def scale_word_spacing(spacing: Tuple[int, int], scale_factor: float) -> Tuple[int, int]:
"""Scale word spacing constraints proportionally"""
if scale_factor == 1.0:
return spacing
min_spacing, max_spacing = spacing
return (
max(1, int(min_spacing * scale_factor)),
max(2, int(max_spacing * scale_factor))
)
class BidirectionalLayouter:
"""
Core layout engine supporting both forward and backward page rendering.
Handles font scaling and maintains position state.
"""
def __init__(self, blocks: List[Block], page_style: PageStyle, page_size: Tuple[int, int] = (800, 600)):
self.blocks = blocks
self.page_style = page_style
self.page_size = page_size
self.chapter_navigator = ChapterNavigator(blocks)
def render_page_forward(self, position: RenderingPosition, font_scale: float = 1.0) -> Tuple[Page, RenderingPosition]:
"""
Render a page starting from the given position, moving forward through the document.
Args:
position: Starting position in document
font_scale: Font scaling factor
Returns:
Tuple of (rendered_page, next_position)
"""
page = Page(size=self.page_size, style=self.page_style)
current_pos = position.copy()
# Start laying out blocks from the current position
while current_pos.chapter_index < len(self.blocks) and page.free_space()[1] > 0:
block = self.blocks[current_pos.block_index]
# Apply font scaling to the block
scaled_block = self._scale_block_fonts(block, font_scale)
# Try to fit the block on the current page
success, new_pos = self._layout_block_on_page(scaled_block, page, current_pos, font_scale)
if not success:
# Block doesn't fit, we're done with this page
break
current_pos = new_pos
return page, current_pos
def render_page_backward(self, end_position: RenderingPosition, font_scale: float = 1.0) -> Tuple[Page, RenderingPosition]:
"""
Render a page that ends at the given position, filling backward.
Critical for "previous page" navigation.
Args:
end_position: Position where page should end
font_scale: Font scaling factor
Returns:
Tuple of (rendered_page, start_position)
"""
# This is a complex operation that requires iterative refinement
# We'll start with an estimated start position and refine it
estimated_start = self._estimate_page_start(end_position, font_scale)
# Render forward from estimated start and see if we reach the target
page, actual_end = self.render_page_forward(estimated_start, font_scale)
# If we overshot or undershot, adjust and try again
# This is a simplified implementation - a full version would be more sophisticated
if self._position_compare(actual_end, end_position) != 0:
# Adjust estimate and try again (simplified)
estimated_start = self._adjust_start_estimate(estimated_start, end_position, actual_end)
page, actual_end = self.render_page_forward(estimated_start, font_scale)
return page, estimated_start
def _scale_block_fonts(self, block: Block, font_scale: float) -> Block:
"""Apply font scaling to all fonts in a block"""
if font_scale == 1.0:
return block
# This is a simplified implementation
# In practice, we'd need to handle each block type appropriately
if isinstance(block, Paragraph):
scaled_block = Paragraph(FontScaler.scale_font(block.style, font_scale))
for word in block.words():
if isinstance(word, Word):
scaled_word = Word(word.text, FontScaler.scale_font(word.style, font_scale))
scaled_block.add_word(scaled_word)
return scaled_block
return block
def _layout_block_on_page(self, block: Block, page: Page, position: RenderingPosition, font_scale: float) -> Tuple[bool, RenderingPosition]:
"""
Try to layout a block on the page starting from the given position.
Returns:
Tuple of (success, new_position)
"""
if isinstance(block, Paragraph):
return self._layout_paragraph_on_page(block, page, position, font_scale)
elif isinstance(block, Heading):
return self._layout_heading_on_page(block, page, position, font_scale)
elif isinstance(block, Table):
return self._layout_table_on_page(block, page, position, font_scale)
elif isinstance(block, HList):
return self._layout_list_on_page(block, page, position, font_scale)
else:
# Skip unknown block types
new_pos = position.copy()
new_pos.block_index += 1
return True, new_pos
def _layout_paragraph_on_page(self, paragraph: Paragraph, page: Page, position: RenderingPosition, font_scale: float) -> Tuple[bool, RenderingPosition]:
"""Layout a paragraph on the page with font scaling support"""
# This would integrate with the existing paragraph_layouter but with font scaling
# For now, this is a placeholder implementation
# Calculate scaled line height
line_height = int(paragraph.style.font_size * font_scale * 1.2) # 1.2 is line spacing factor
if not page.can_fit_line(line_height):
return False, position
# Create a line and try to fit words
y_cursor = page._current_y_offset
x_cursor = page.border_size
# Scale word spacing constraints
word_spacing = FontScaler.scale_word_spacing((5, 15), font_scale) # Default spacing
line = Line(
spacing=word_spacing,
origin=(x_cursor, y_cursor),
size=(page.available_width, line_height),
draw=page.draw,
font=FontScaler.scale_font(paragraph.style, font_scale)
)
# Add words starting from position.word_index
words_added = 0
for i, word in enumerate(paragraph.words[position.word_index:], start=position.word_index):
success, overflow = line.add_word(word)
if not success:
break
words_added += 1
if words_added > 0:
page.add_child(line)
page._current_y_offset += line_height
new_pos = position.copy()
new_pos.word_index += words_added
# If we finished the paragraph, move to next block
if new_pos.word_index >= len(paragraph.words):
new_pos.block_index += 1
new_pos.word_index = 0
return True, new_pos
return False, position
def _layout_heading_on_page(self, heading: Heading, page: Page, position: RenderingPosition, font_scale: float) -> Tuple[bool, RenderingPosition]:
"""Layout a heading on the page"""
# Similar to paragraph but with heading-specific styling
return self._layout_paragraph_on_page(heading, page, position, font_scale)
def _layout_table_on_page(self, table: Table, page: Page, position: RenderingPosition, font_scale: float) -> Tuple[bool, RenderingPosition]:
"""Layout a table on the page with column fitting and row continuation"""
# This is a complex operation that would need full table layout logic
# For now, skip tables
new_pos = position.copy()
new_pos.block_index += 1
new_pos.table_row = 0
new_pos.table_col = 0
return True, new_pos
def _layout_list_on_page(self, hlist: HList, page: Page, position: RenderingPosition, font_scale: float) -> Tuple[bool, RenderingPosition]:
"""Layout a list on the page"""
# This would need list-specific layout logic
# For now, skip lists
new_pos = position.copy()
new_pos.block_index += 1
new_pos.list_item_index = 0
return True, new_pos
def _estimate_page_start(self, end_position: RenderingPosition, font_scale: float) -> RenderingPosition:
"""Estimate where a page should start to end at the given position"""
# This is a simplified heuristic - a full implementation would be more sophisticated
estimated_start = end_position.copy()
# Move back by an estimated number of blocks that would fit on a page
estimated_blocks_per_page = max(1, int(10 / font_scale)) # Rough estimate
estimated_start.block_index = max(0, end_position.block_index - estimated_blocks_per_page)
estimated_start.word_index = 0
return estimated_start
def _adjust_start_estimate(self, current_start: RenderingPosition, target_end: RenderingPosition, actual_end: RenderingPosition) -> RenderingPosition:
"""Adjust start position estimate based on overshoot/undershoot"""
# Simplified adjustment logic
adjusted = current_start.copy()
comparison = self._position_compare(actual_end, target_end)
if comparison > 0: # Overshot
adjusted.block_index = max(0, adjusted.block_index + 1)
elif comparison < 0: # Undershot
adjusted.block_index = max(0, adjusted.block_index - 1)
return adjusted
def _position_compare(self, pos1: RenderingPosition, pos2: RenderingPosition) -> int:
"""Compare two positions (-1: pos1 < pos2, 0: equal, 1: pos1 > pos2)"""
if pos1.chapter_index != pos2.chapter_index:
return 1 if pos1.chapter_index > pos2.chapter_index else -1
if pos1.block_index != pos2.block_index:
return 1 if pos1.block_index > pos2.block_index else -1
if pos1.word_index != pos2.word_index:
return 1 if pos1.word_index > pos2.word_index else -1
return 0
# Add can_fit_line method to Page class if it doesn't exist
def _add_page_methods():
"""Add missing methods to Page class"""
if not hasattr(Page, 'can_fit_line'):
def can_fit_line(self, line_height: int) -> bool:
"""Check if a line of given height can fit on the page"""
available_height = self.content_size[1] - self._current_y_offset
return available_height >= line_height
Page.can_fit_line = can_fit_line
if not hasattr(Page, 'available_width'):
@property
def available_width(self) -> int:
"""Get available width for content"""
return self.content_size[0]
Page.available_width = available_width
# Apply the page methods
_add_page_methods()
+493
View File
@@ -0,0 +1,493 @@
"""
High-performance ereader layout manager with sub-second page rendering.
This module provides the main interface for ereader applications, combining
position tracking, font scaling, chapter navigation, and intelligent page buffering
into a unified, easy-to-use API.
"""
from __future__ import annotations
from typing import List, Dict, Optional, Tuple, Any, Callable
import json
import os
from pathlib import Path
from .ereader_layout import RenderingPosition, ChapterNavigator, ChapterInfo
from .page_buffer import BufferedPageRenderer
from pyWebLayout.abstract.block import Block, HeadingLevel
from pyWebLayout.concrete.page import Page
from pyWebLayout.style.page_style import PageStyle
class BookmarkManager:
"""
Manages bookmarks and reading position persistence for ereader applications.
"""
def __init__(self, document_id: str, bookmarks_dir: str = "bookmarks"):
"""
Initialize bookmark manager.
Args:
document_id: Unique identifier for the document
bookmarks_dir: Directory to store bookmark files
"""
self.document_id = document_id
self.bookmarks_dir = Path(bookmarks_dir)
self.bookmarks_dir.mkdir(exist_ok=True)
self.bookmarks_file = self.bookmarks_dir / f"{document_id}_bookmarks.json"
self.position_file = self.bookmarks_dir / f"{document_id}_position.json"
self._bookmarks: Dict[str, RenderingPosition] = {}
self._load_bookmarks()
def _load_bookmarks(self):
"""Load bookmarks from file"""
if self.bookmarks_file.exists():
try:
with open(self.bookmarks_file, 'r') as f:
data = json.load(f)
self._bookmarks = {
name: RenderingPosition.from_dict(pos_data)
for name, pos_data in data.items()
}
except Exception as e:
print(f"Failed to load bookmarks: {e}")
self._bookmarks = {}
def _save_bookmarks(self):
"""Save bookmarks to file"""
try:
data = {
name: position.to_dict()
for name, position in self._bookmarks.items()
}
with open(self.bookmarks_file, 'w') as f:
json.dump(data, f, indent=2)
except Exception as e:
print(f"Failed to save bookmarks: {e}")
def add_bookmark(self, name: str, position: RenderingPosition):
"""
Add a bookmark at the given position.
Args:
name: Bookmark name
position: Position to bookmark
"""
self._bookmarks[name] = position
self._save_bookmarks()
def remove_bookmark(self, name: str) -> bool:
"""
Remove a bookmark.
Args:
name: Bookmark name to remove
Returns:
True if bookmark was removed, False if not found
"""
if name in self._bookmarks:
del self._bookmarks[name]
self._save_bookmarks()
return True
return False
def get_bookmark(self, name: str) -> Optional[RenderingPosition]:
"""
Get a bookmark position.
Args:
name: Bookmark name
Returns:
Bookmark position or None if not found
"""
return self._bookmarks.get(name)
def list_bookmarks(self) -> List[Tuple[str, RenderingPosition]]:
"""
Get all bookmarks.
Returns:
List of (name, position) tuples
"""
return list(self._bookmarks.items())
def save_reading_position(self, position: RenderingPosition):
"""
Save the current reading position.
Args:
position: Current reading position
"""
try:
with open(self.position_file, 'w') as f:
json.dump(position.to_dict(), f, indent=2)
except Exception as e:
print(f"Failed to save reading position: {e}")
def load_reading_position(self) -> Optional[RenderingPosition]:
"""
Load the last reading position.
Returns:
Last reading position or None if not found
"""
if self.position_file.exists():
try:
with open(self.position_file, 'r') as f:
data = json.load(f)
return RenderingPosition.from_dict(data)
except Exception as e:
print(f"Failed to load reading position: {e}")
return None
class EreaderLayoutManager:
"""
High-level ereader layout manager providing a complete interface for ereader applications.
Features:
- Sub-second page rendering with intelligent buffering
- Font scaling support
- Chapter navigation
- Bookmark management
- Position persistence
- Progress tracking
"""
def __init__(self,
blocks: List[Block],
page_size: Tuple[int, int],
document_id: str = "default",
buffer_size: int = 5,
page_style: Optional[PageStyle] = None):
"""
Initialize the ereader layout manager.
Args:
blocks: Document blocks to render
page_size: Page size (width, height) in pixels
document_id: Unique identifier for the document (for bookmarks/position)
buffer_size: Number of pages to cache in each direction
page_style: Custom page styling (uses default if None)
"""
self.blocks = blocks
self.page_size = page_size
self.document_id = document_id
# Initialize page style
if page_style is None:
page_style = PageStyle()
self.page_style = page_style
# Initialize core components
self.renderer = BufferedPageRenderer(blocks, page_style, buffer_size, page_size)
self.chapter_navigator = ChapterNavigator(blocks)
self.bookmark_manager = BookmarkManager(document_id)
# Current state
self.current_position = RenderingPosition()
self.font_scale = 1.0
# Load last reading position if available
saved_position = self.bookmark_manager.load_reading_position()
if saved_position:
self.current_position = saved_position
# Callbacks for UI updates
self.position_changed_callback: Optional[Callable[[RenderingPosition], None]] = None
self.chapter_changed_callback: Optional[Callable[[Optional[ChapterInfo]], None]] = None
def set_position_changed_callback(self, callback: Callable[[RenderingPosition], None]):
"""Set callback for position changes"""
self.position_changed_callback = callback
def set_chapter_changed_callback(self, callback: Callable[[Optional[ChapterInfo]], None]):
"""Set callback for chapter changes"""
self.chapter_changed_callback = callback
def _notify_position_changed(self):
"""Notify UI of position change"""
if self.position_changed_callback:
self.position_changed_callback(self.current_position)
# Check if chapter changed
current_chapter = self.chapter_navigator.get_current_chapter(self.current_position)
if self.chapter_changed_callback:
self.chapter_changed_callback(current_chapter)
# Auto-save reading position
self.bookmark_manager.save_reading_position(self.current_position)
def get_current_page(self) -> Page:
"""
Get the page at the current reading position.
Returns:
Rendered page
"""
page, _ = self.renderer.render_page(self.current_position, self.font_scale)
return page
def next_page(self) -> Optional[Page]:
"""
Advance to the next page.
Returns:
Next page or None if at end of document
"""
page, next_position = self.renderer.render_page(self.current_position, self.font_scale)
# Check if we made progress
if next_position != self.current_position:
self.current_position = next_position
self._notify_position_changed()
return self.get_current_page()
return None # At end of document
def previous_page(self) -> Optional[Page]:
"""
Go to the previous page.
Returns:
Previous page or None if at beginning of document
"""
if self._is_at_beginning():
return None
# Use backward rendering to find the previous page
page, start_position = self.renderer.render_page_backward(self.current_position, self.font_scale)
if start_position != self.current_position:
self.current_position = start_position
self._notify_position_changed()
return page
return None # At beginning of document
def _is_at_beginning(self) -> bool:
"""Check if we're at the beginning of the document"""
return (self.current_position.chapter_index == 0 and
self.current_position.block_index == 0 and
self.current_position.word_index == 0)
def jump_to_position(self, position: RenderingPosition) -> Page:
"""
Jump to a specific position in the document.
Args:
position: Position to jump to
Returns:
Page at the new position
"""
self.current_position = position
self._notify_position_changed()
return self.get_current_page()
def jump_to_chapter(self, chapter_title: str) -> Optional[Page]:
"""
Jump to a specific chapter by title.
Args:
chapter_title: Title of the chapter to jump to
Returns:
Page at chapter start or None if chapter not found
"""
position = self.chapter_navigator.get_chapter_position(chapter_title)
if position:
return self.jump_to_position(position)
return None
def jump_to_chapter_index(self, chapter_index: int) -> Optional[Page]:
"""
Jump to a chapter by index.
Args:
chapter_index: Index of the chapter (0-based)
Returns:
Page at chapter start or None if index invalid
"""
chapters = self.chapter_navigator.chapters
if 0 <= chapter_index < len(chapters):
return self.jump_to_position(chapters[chapter_index].position)
return None
def set_font_scale(self, scale: float) -> Page:
"""
Change the font scale and re-render current page.
Args:
scale: Font scaling factor (1.0 = normal, 2.0 = double size, etc.)
Returns:
Re-rendered page with new font scale
"""
if scale != self.font_scale:
self.font_scale = scale
# The renderer will handle cache invalidation
return self.get_current_page()
def get_font_scale(self) -> float:
"""Get the current font scale"""
return self.font_scale
def get_table_of_contents(self) -> List[Tuple[str, HeadingLevel, RenderingPosition]]:
"""
Get the table of contents.
Returns:
List of (title, level, position) tuples
"""
return self.chapter_navigator.get_table_of_contents()
def get_current_chapter(self) -> Optional[ChapterInfo]:
"""
Get information about the current chapter.
Returns:
Current chapter info or None if no chapters
"""
return self.chapter_navigator.get_current_chapter(self.current_position)
def add_bookmark(self, name: str) -> bool:
"""
Add a bookmark at the current position.
Args:
name: Bookmark name
Returns:
True if bookmark was added successfully
"""
try:
self.bookmark_manager.add_bookmark(name, self.current_position)
return True
except Exception:
return False
def remove_bookmark(self, name: str) -> bool:
"""
Remove a bookmark.
Args:
name: Bookmark name
Returns:
True if bookmark was removed
"""
return self.bookmark_manager.remove_bookmark(name)
def jump_to_bookmark(self, name: str) -> Optional[Page]:
"""
Jump to a bookmark.
Args:
name: Bookmark name
Returns:
Page at bookmark position or None if bookmark not found
"""
position = self.bookmark_manager.get_bookmark(name)
if position:
return self.jump_to_position(position)
return None
def list_bookmarks(self) -> List[Tuple[str, RenderingPosition]]:
"""
Get all bookmarks.
Returns:
List of (name, position) tuples
"""
return self.bookmark_manager.list_bookmarks()
def get_reading_progress(self) -> float:
"""
Get reading progress as a percentage.
Returns:
Progress from 0.0 to 1.0
"""
if not self.blocks:
return 0.0
# Simple progress calculation based on block index
# A more sophisticated version would consider word positions
total_blocks = len(self.blocks)
current_block = min(self.current_position.block_index, total_blocks - 1)
return current_block / max(1, total_blocks - 1)
def get_position_info(self) -> Dict[str, Any]:
"""
Get detailed information about the current position.
Returns:
Dictionary with position details
"""
current_chapter = self.get_current_chapter()
return {
'position': self.current_position.to_dict(),
'chapter': {
'title': current_chapter.title if current_chapter else None,
'level': current_chapter.level if current_chapter else None,
'index': current_chapter.block_index if current_chapter else None
},
'progress': self.get_reading_progress(),
'font_scale': self.font_scale,
'page_size': self.page_size
}
def get_cache_stats(self) -> Dict[str, Any]:
"""
Get cache statistics for debugging/monitoring.
Returns:
Dictionary with cache statistics
"""
return self.renderer.get_cache_stats()
def shutdown(self):
"""
Shutdown the ereader manager and clean up resources.
Call this when the application is closing.
"""
# Save current position
self.bookmark_manager.save_reading_position(self.current_position)
# Shutdown renderer and buffer
self.renderer.shutdown()
def __del__(self):
"""Cleanup on destruction"""
self.shutdown()
# Convenience function for quick setup
def create_ereader_manager(blocks: List[Block],
page_size: Tuple[int, int],
document_id: str = "default",
**kwargs) -> EreaderLayoutManager:
"""
Convenience function to create an ereader manager with sensible defaults.
Args:
blocks: Document blocks to render
page_size: Page size (width, height) in pixels
document_id: Unique identifier for the document
**kwargs: Additional arguments passed to EreaderLayoutManager
Returns:
Configured EreaderLayoutManager instance
"""
return EreaderLayoutManager(blocks, page_size, document_id, **kwargs)
+411
View File
@@ -0,0 +1,411 @@
"""
Multi-process page buffering system for high-performance ereader navigation.
This module provides intelligent page caching with background rendering using
multiprocessing to achieve sub-second page navigation performance.
"""
from __future__ import annotations
from typing import Dict, Optional, List, Tuple, Any
from collections import OrderedDict
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, as_completed, Future
import threading
import time
import pickle
from dataclasses import asdict
from .ereader_layout import RenderingPosition, BidirectionalLayouter
from pyWebLayout.concrete.page import Page
from pyWebLayout.abstract.block import Block
from pyWebLayout.style.page_style import PageStyle
def _render_page_worker(args: Tuple[List[Block], PageStyle, RenderingPosition, float, bool]) -> Tuple[RenderingPosition, bytes, RenderingPosition]:
"""
Worker function for multiprocess page rendering.
Args:
args: Tuple of (blocks, page_style, position, font_scale, is_backward)
Returns:
Tuple of (original_position, pickled_page, next_position)
"""
blocks, page_style, position, font_scale, is_backward = args
layouter = BidirectionalLayouter(blocks, page_style)
if is_backward:
page, next_pos = layouter.render_page_backward(position, font_scale)
else:
page, next_pos = layouter.render_page_forward(position, font_scale)
# Serialize the page for inter-process communication
pickled_page = pickle.dumps(page)
return position, pickled_page, next_pos
class PageBuffer:
"""
Intelligent page caching system with LRU eviction and background rendering.
Maintains separate forward and backward buffers for optimal navigation performance.
"""
def __init__(self, buffer_size: int = 5, max_workers: int = 4):
"""
Initialize the page buffer.
Args:
buffer_size: Number of pages to cache in each direction
max_workers: Maximum number of worker processes for background rendering
"""
self.buffer_size = buffer_size
self.max_workers = max_workers
# LRU caches for forward and backward pages
self.forward_buffer: OrderedDict[RenderingPosition, Page] = OrderedDict()
self.backward_buffer: OrderedDict[RenderingPosition, Page] = OrderedDict()
# Position tracking for next/previous positions
self.position_map: Dict[RenderingPosition, RenderingPosition] = {} # current -> next
self.reverse_position_map: Dict[RenderingPosition, RenderingPosition] = {} # current -> previous
# Background rendering
self.executor: Optional[ProcessPoolExecutor] = None
self.pending_renders: Dict[RenderingPosition, Future] = {}
self.render_lock = threading.Lock()
# Document state
self.blocks: Optional[List[Block]] = None
self.page_style: Optional[PageStyle] = None
self.current_font_scale: float = 1.0
def initialize(self, blocks: List[Block], page_style: PageStyle, font_scale: float = 1.0):
"""
Initialize the buffer with document blocks and page style.
Args:
blocks: Document blocks to render
page_style: Page styling configuration
font_scale: Current font scaling factor
"""
self.blocks = blocks
self.page_style = page_style
self.current_font_scale = font_scale
# Start the process pool
if self.executor is None:
self.executor = ProcessPoolExecutor(max_workers=self.max_workers)
def get_page(self, position: RenderingPosition) -> Optional[Page]:
"""
Get a cached page if available.
Args:
position: Position to get page for
Returns:
Cached page or None if not available
"""
# Check forward buffer first
if position in self.forward_buffer:
# Move to end (most recently used)
page = self.forward_buffer.pop(position)
self.forward_buffer[position] = page
return page
# Check backward buffer
if position in self.backward_buffer:
# Move to end (most recently used)
page = self.backward_buffer.pop(position)
self.backward_buffer[position] = page
return page
return None
def cache_page(self, position: RenderingPosition, page: Page, next_position: Optional[RenderingPosition] = None, is_backward: bool = False):
"""
Cache a rendered page with LRU eviction.
Args:
position: Position of the page
page: Rendered page to cache
next_position: Position of the next page (for forward navigation)
is_backward: Whether this is a backward-rendered page
"""
target_buffer = self.backward_buffer if is_backward else self.forward_buffer
# Add to cache
target_buffer[position] = page
# Track position relationships
if next_position:
if is_backward:
self.reverse_position_map[next_position] = position
else:
self.position_map[position] = next_position
# Evict oldest if buffer is full
if len(target_buffer) > self.buffer_size:
oldest_pos, _ = target_buffer.popitem(last=False)
# Clean up position maps
self.position_map.pop(oldest_pos, None)
self.reverse_position_map.pop(oldest_pos, None)
def start_background_rendering(self, current_position: RenderingPosition, direction: str = 'forward'):
"""
Start background rendering of upcoming pages.
Args:
current_position: Current reading position
direction: 'forward', 'backward', or 'both'
"""
if not self.blocks or not self.page_style or not self.executor:
return
with self.render_lock:
if direction in ['forward', 'both']:
self._queue_forward_renders(current_position)
if direction in ['backward', 'both']:
self._queue_backward_renders(current_position)
def _queue_forward_renders(self, start_position: RenderingPosition):
"""Queue forward page renders starting from the given position"""
current_pos = start_position
for i in range(self.buffer_size):
# Skip if already cached or being rendered
if current_pos in self.forward_buffer or current_pos in self.pending_renders:
# Try to get next position from cache
current_pos = self.position_map.get(current_pos)
if not current_pos:
break
continue
# Queue render job
args = (self.blocks, self.page_style, current_pos, self.current_font_scale, False)
future = self.executor.submit(_render_page_worker, args)
self.pending_renders[current_pos] = future
# We don't know the next position yet, so we'll update it when the render completes
break
def _queue_backward_renders(self, start_position: RenderingPosition):
"""Queue backward page renders ending at the given position"""
current_pos = start_position
for i in range(self.buffer_size):
# Skip if already cached or being rendered
if current_pos in self.backward_buffer or current_pos in self.pending_renders:
# Try to get previous position from cache
current_pos = self.reverse_position_map.get(current_pos)
if not current_pos:
break
continue
# Queue render job
args = (self.blocks, self.page_style, current_pos, self.current_font_scale, True)
future = self.executor.submit(_render_page_worker, args)
self.pending_renders[current_pos] = future
# We don't know the previous position yet, so we'll update it when the render completes
break
def check_completed_renders(self):
"""Check for completed background renders and cache the results"""
if not self.pending_renders:
return
completed = []
with self.render_lock:
for position, future in self.pending_renders.items():
if future.done():
try:
original_pos, pickled_page, next_pos = future.result()
# Deserialize the page
page = pickle.loads(pickled_page)
# Cache the page
self.cache_page(original_pos, page, next_pos, is_backward=False)
completed.append(position)
except Exception as e:
print(f"Background render failed for position {position}: {e}")
completed.append(position)
# Remove completed renders
for pos in completed:
self.pending_renders.pop(pos, None)
def invalidate_all(self):
"""Clear all cached pages and cancel pending renders"""
with self.render_lock:
# Cancel pending renders
for future in self.pending_renders.values():
future.cancel()
self.pending_renders.clear()
# Clear caches
self.forward_buffer.clear()
self.backward_buffer.clear()
self.position_map.clear()
self.reverse_position_map.clear()
def set_font_scale(self, font_scale: float):
"""
Update font scale and invalidate cache.
Args:
font_scale: New font scaling factor
"""
if font_scale != self.current_font_scale:
self.current_font_scale = font_scale
self.invalidate_all()
def get_cache_stats(self) -> Dict[str, Any]:
"""Get cache statistics for debugging/monitoring"""
return {
'forward_buffer_size': len(self.forward_buffer),
'backward_buffer_size': len(self.backward_buffer),
'pending_renders': len(self.pending_renders),
'position_mappings': len(self.position_map),
'reverse_position_mappings': len(self.reverse_position_map),
'current_font_scale': self.current_font_scale
}
def shutdown(self):
"""Shutdown the page buffer and clean up resources"""
if self.executor:
# Cancel pending renders
with self.render_lock:
for future in self.pending_renders.values():
future.cancel()
# Shutdown executor
self.executor.shutdown(wait=True)
self.executor = None
# Clear all caches
self.invalidate_all()
def __del__(self):
"""Cleanup on destruction"""
self.shutdown()
class BufferedPageRenderer:
"""
High-level interface for buffered page rendering with automatic background caching.
"""
def __init__(self, blocks: List[Block], page_style: PageStyle, buffer_size: int = 5, page_size: Tuple[int, int] = (800, 600)):
"""
Initialize the buffered renderer.
Args:
blocks: Document blocks to render
page_style: Page styling configuration
buffer_size: Number of pages to cache in each direction
page_size: Page size (width, height) in pixels
"""
self.layouter = BidirectionalLayouter(blocks, page_style, page_size)
self.buffer = PageBuffer(buffer_size)
self.buffer.initialize(blocks, page_style)
self.current_position = RenderingPosition()
self.font_scale = 1.0
def render_page(self, position: RenderingPosition, font_scale: float = 1.0) -> Tuple[Page, RenderingPosition]:
"""
Render a page with intelligent caching.
Args:
position: Position to render from
font_scale: Font scaling factor
Returns:
Tuple of (rendered_page, next_position)
"""
# Update font scale if changed
if font_scale != self.font_scale:
self.font_scale = font_scale
self.buffer.set_font_scale(font_scale)
# Check cache first
cached_page = self.buffer.get_page(position)
if cached_page:
# Get next position from position map
next_pos = self.buffer.position_map.get(position, position)
# Start background rendering for upcoming pages
self.buffer.start_background_rendering(position, 'forward')
return cached_page, next_pos
# Render the page directly
page, next_pos = self.layouter.render_page_forward(position, font_scale)
# Cache the result
self.buffer.cache_page(position, page, next_pos)
# Start background rendering
self.buffer.start_background_rendering(position, 'both')
# Check for completed background renders
self.buffer.check_completed_renders()
return page, next_pos
def render_page_backward(self, end_position: RenderingPosition, font_scale: float = 1.0) -> Tuple[Page, RenderingPosition]:
"""
Render a page ending at the given position with intelligent caching.
Args:
end_position: Position where page should end
font_scale: Font scaling factor
Returns:
Tuple of (rendered_page, start_position)
"""
# Update font scale if changed
if font_scale != self.font_scale:
self.font_scale = font_scale
self.buffer.set_font_scale(font_scale)
# Check cache first
cached_page = self.buffer.get_page(end_position)
if cached_page:
# Get previous position from reverse position map
prev_pos = self.buffer.reverse_position_map.get(end_position, end_position)
# Start background rendering for previous pages
self.buffer.start_background_rendering(end_position, 'backward')
return cached_page, prev_pos
# Render the page directly
page, start_pos = self.layouter.render_page_backward(end_position, font_scale)
# Cache the result
self.buffer.cache_page(start_pos, page, end_position, is_backward=True)
# Start background rendering
self.buffer.start_background_rendering(end_position, 'both')
# Check for completed background renders
self.buffer.check_completed_renders()
return page, start_pos
def get_cache_stats(self) -> Dict[str, Any]:
"""Get cache statistics"""
return self.buffer.get_cache_stats()
def shutdown(self):
"""Shutdown the renderer and clean up resources"""
self.buffer.shutdown()
+481
View File
@@ -0,0 +1,481 @@
"""
Recursive location index system for dynamic content positioning.
This module provides a flexible, hierarchical position tracking system that can
reference any type of content (words, images, table cells, list items, etc.)
in a nested document structure.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Union, Tuple
from enum import Enum
import json
import pickle
import shelve
from pathlib import Path
class ContentType(Enum):
"""Types of content that can be referenced in the position index"""
DOCUMENT = "document"
CHAPTER = "chapter"
BLOCK = "block"
PARAGRAPH = "paragraph"
HEADING = "heading"
TABLE = "table"
TABLE_ROW = "table_row"
TABLE_CELL = "table_cell"
LIST = "list"
LIST_ITEM = "list_item"
WORD = "word"
IMAGE = "image"
LINK = "link"
BUTTON = "button"
FORM_FIELD = "form_field"
LINE = "line" # Rendered line of text
PAGE = "page" # Rendered page
@dataclass
class LocationNode:
"""
A single node in the recursive location index.
Each node represents a position within a specific content type.
"""
content_type: ContentType
index: int = 0 # Position within this content type
offset: int = 0 # Offset within the indexed item (e.g., character offset in word)
metadata: Dict[str, Any] = field(default_factory=dict) # Additional context
def to_dict(self) -> Dict[str, Any]:
"""Serialize node to dictionary"""
return {
'content_type': self.content_type.value,
'index': self.index,
'offset': self.offset,
'metadata': self.metadata
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'LocationNode':
"""Deserialize node from dictionary"""
return cls(
content_type=ContentType(data['content_type']),
index=data['index'],
offset=data['offset'],
metadata=data.get('metadata', {})
)
def __str__(self) -> str:
"""Human-readable representation"""
if self.offset > 0:
return f"{self.content_type.value}[{self.index}]+{self.offset}"
return f"{self.content_type.value}[{self.index}]"
@dataclass
class RecursivePosition:
"""
Hierarchical position that can reference any nested content structure.
The path represents a traversal from document root to the specific location:
- Document -> Chapter[2] -> Block[5] -> Paragraph -> Word[12] -> Character[3]
- Document -> Chapter[1] -> Block[3] -> Table -> Row[2] -> Cell[1] -> Word[0]
- Document -> Chapter[0] -> Block[1] -> Image
"""
path: List[LocationNode] = field(default_factory=list)
rendering_metadata: Dict[str, Any] = field(default_factory=dict) # Font scale, page size, etc.
def __post_init__(self):
"""Ensure we always have at least a document root"""
if not self.path:
self.path = [LocationNode(ContentType.DOCUMENT)]
def copy(self) -> 'RecursivePosition':
"""Create a deep copy of this position"""
return RecursivePosition(
path=[LocationNode(node.content_type, node.index, node.offset, node.metadata.copy())
for node in self.path],
rendering_metadata=self.rendering_metadata.copy()
)
def get_node(self, content_type: ContentType) -> Optional[LocationNode]:
"""Get the first node of a specific content type in the path"""
for node in self.path:
if node.content_type == content_type:
return node
return None
def get_nodes(self, content_type: ContentType) -> List[LocationNode]:
"""Get all nodes of a specific content type in the path"""
return [node for node in self.path if node.content_type == content_type]
def add_node(self, node: LocationNode) -> 'RecursivePosition':
"""Add a node to the path (returns self for chaining)"""
self.path.append(node)
return self
def pop_node(self) -> Optional[LocationNode]:
"""Remove and return the last node in the path"""
if len(self.path) > 1: # Keep at least document root
return self.path.pop()
return None
def get_depth(self) -> int:
"""Get the depth of the position (number of nodes)"""
return len(self.path)
def get_leaf_node(self) -> LocationNode:
"""Get the deepest (most specific) node in the path"""
return self.path[-1] if self.path else LocationNode(ContentType.DOCUMENT)
def truncate_to_type(self, content_type: ContentType) -> 'RecursivePosition':
"""Truncate path to end at the first occurrence of the given content type"""
for i, node in enumerate(self.path):
if node.content_type == content_type:
self.path = self.path[:i+1]
break
return self
def is_ancestor_of(self, other: 'RecursivePosition') -> bool:
"""Check if this position is an ancestor of another position"""
if len(self.path) >= len(other.path):
return False
for i, node in enumerate(self.path):
if i >= len(other.path):
return False
other_node = other.path[i]
if (node.content_type != other_node.content_type or
node.index != other_node.index):
return False
return True
def is_descendant_of(self, other: 'RecursivePosition') -> bool:
"""Check if this position is a descendant of another position"""
return other.is_ancestor_of(self)
def get_common_ancestor(self, other: 'RecursivePosition') -> 'RecursivePosition':
"""Find the deepest common ancestor with another position"""
common_path = []
min_length = min(len(self.path), len(other.path))
for i in range(min_length):
if (self.path[i].content_type == other.path[i].content_type and
self.path[i].index == other.path[i].index):
common_path.append(self.path[i])
else:
break
return RecursivePosition(path=common_path)
def to_dict(self) -> Dict[str, Any]:
"""Serialize position to dictionary for JSON storage"""
return {
'path': [node.to_dict() for node in self.path],
'rendering_metadata': self.rendering_metadata
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'RecursivePosition':
"""Deserialize position from dictionary"""
return cls(
path=[LocationNode.from_dict(node_data) for node_data in data['path']],
rendering_metadata=data.get('rendering_metadata', {})
)
def to_json(self) -> str:
"""Serialize to JSON string"""
return json.dumps(self.to_dict(), indent=2)
@classmethod
def from_json(cls, json_str: str) -> 'RecursivePosition':
"""Deserialize from JSON string"""
return cls.from_dict(json.loads(json_str))
def __str__(self) -> str:
"""Human-readable path representation"""
return " -> ".join(str(node) for node in self.path)
def __eq__(self, other) -> bool:
"""Check equality with another position"""
if not isinstance(other, RecursivePosition):
return False
return (self.path == other.path and
self.rendering_metadata == other.rendering_metadata)
def __hash__(self) -> int:
"""Make position hashable for use as dict key"""
path_tuple = tuple((node.content_type, node.index, node.offset) for node in self.path)
return hash(path_tuple)
class PositionBuilder:
"""
Builder class for constructing RecursivePosition objects fluently.
Example usage:
position = (PositionBuilder()
.chapter(2)
.block(5)
.paragraph()
.word(12, offset=3)
.build())
"""
def __init__(self):
self._position = RecursivePosition()
def document(self, index: int = 0, **metadata) -> 'PositionBuilder':
"""Add document node"""
self._position.add_node(LocationNode(ContentType.DOCUMENT, index, metadata=metadata))
return self
def chapter(self, index: int, **metadata) -> 'PositionBuilder':
"""Add chapter node"""
self._position.add_node(LocationNode(ContentType.CHAPTER, index, metadata=metadata))
return self
def block(self, index: int, **metadata) -> 'PositionBuilder':
"""Add block node"""
self._position.add_node(LocationNode(ContentType.BLOCK, index, metadata=metadata))
return self
def paragraph(self, index: int = 0, **metadata) -> 'PositionBuilder':
"""Add paragraph node"""
self._position.add_node(LocationNode(ContentType.PARAGRAPH, index, metadata=metadata))
return self
def heading(self, index: int = 0, **metadata) -> 'PositionBuilder':
"""Add heading node"""
self._position.add_node(LocationNode(ContentType.HEADING, index, metadata=metadata))
return self
def table(self, index: int = 0, **metadata) -> 'PositionBuilder':
"""Add table node"""
self._position.add_node(LocationNode(ContentType.TABLE, index, metadata=metadata))
return self
def table_row(self, index: int, **metadata) -> 'PositionBuilder':
"""Add table row node"""
self._position.add_node(LocationNode(ContentType.TABLE_ROW, index, metadata=metadata))
return self
def table_cell(self, index: int, **metadata) -> 'PositionBuilder':
"""Add table cell node"""
self._position.add_node(LocationNode(ContentType.TABLE_CELL, index, metadata=metadata))
return self
def list(self, index: int = 0, **metadata) -> 'PositionBuilder':
"""Add list node"""
self._position.add_node(LocationNode(ContentType.LIST, index, metadata=metadata))
return self
def list_item(self, index: int, **metadata) -> 'PositionBuilder':
"""Add list item node"""
self._position.add_node(LocationNode(ContentType.LIST_ITEM, index, metadata=metadata))
return self
def word(self, index: int, offset: int = 0, **metadata) -> 'PositionBuilder':
"""Add word node"""
self._position.add_node(LocationNode(ContentType.WORD, index, offset, metadata=metadata))
return self
def image(self, index: int = 0, **metadata) -> 'PositionBuilder':
"""Add image node"""
self._position.add_node(LocationNode(ContentType.IMAGE, index, metadata=metadata))
return self
def link(self, index: int, **metadata) -> 'PositionBuilder':
"""Add link node"""
self._position.add_node(LocationNode(ContentType.LINK, index, metadata=metadata))
return self
def button(self, index: int, **metadata) -> 'PositionBuilder':
"""Add button node"""
self._position.add_node(LocationNode(ContentType.BUTTON, index, metadata=metadata))
return self
def form_field(self, index: int, **metadata) -> 'PositionBuilder':
"""Add form field node"""
self._position.add_node(LocationNode(ContentType.FORM_FIELD, index, metadata=metadata))
return self
def line(self, index: int, **metadata) -> 'PositionBuilder':
"""Add rendered line node"""
self._position.add_node(LocationNode(ContentType.LINE, index, metadata=metadata))
return self
def page(self, index: int, **metadata) -> 'PositionBuilder':
"""Add page node"""
self._position.add_node(LocationNode(ContentType.PAGE, index, metadata=metadata))
return self
def with_rendering_metadata(self, **metadata) -> 'PositionBuilder':
"""Add rendering metadata (font scale, page size, etc.)"""
self._position.rendering_metadata.update(metadata)
return self
def build(self) -> RecursivePosition:
"""Build and return the final position"""
return self._position
class PositionStorage:
"""
Storage manager for recursive positions supporting both JSON and shelf formats.
"""
def __init__(self, storage_dir: str = "positions", use_shelf: bool = False):
"""
Initialize position storage.
Args:
storage_dir: Directory to store position files
use_shelf: If True, use Python shelf format; if False, use JSON
"""
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(exist_ok=True)
self.use_shelf = use_shelf
def save_position(self, document_id: str, position_name: str, position: RecursivePosition):
"""Save a position to storage"""
if self.use_shelf:
self._save_to_shelf(document_id, position_name, position)
else:
self._save_to_json(document_id, position_name, position)
def load_position(self, document_id: str, position_name: str) -> Optional[RecursivePosition]:
"""Load a position from storage"""
if self.use_shelf:
return self._load_from_shelf(document_id, position_name)
else:
return self._load_from_json(document_id, position_name)
def list_positions(self, document_id: str) -> List[str]:
"""List all saved positions for a document"""
if self.use_shelf:
return self._list_shelf_positions(document_id)
else:
return self._list_json_positions(document_id)
def delete_position(self, document_id: str, position_name: str) -> bool:
"""Delete a position from storage"""
if self.use_shelf:
return self._delete_from_shelf(document_id, position_name)
else:
return self._delete_from_json(document_id, position_name)
def _save_to_json(self, document_id: str, position_name: str, position: RecursivePosition):
"""Save position as JSON file"""
file_path = self.storage_dir / f"{document_id}_{position_name}.json"
with open(file_path, 'w') as f:
json.dump(position.to_dict(), f, indent=2)
def _load_from_json(self, document_id: str, position_name: str) -> Optional[RecursivePosition]:
"""Load position from JSON file"""
file_path = self.storage_dir / f"{document_id}_{position_name}.json"
if not file_path.exists():
return None
try:
with open(file_path, 'r') as f:
data = json.load(f)
return RecursivePosition.from_dict(data)
except Exception:
return None
def _list_json_positions(self, document_id: str) -> List[str]:
"""List JSON position files for a document"""
pattern = f"{document_id}_*.json"
files = list(self.storage_dir.glob(pattern))
return [f.stem.replace(f"{document_id}_", "") for f in files]
def _delete_from_json(self, document_id: str, position_name: str) -> bool:
"""Delete JSON position file"""
file_path = self.storage_dir / f"{document_id}_{position_name}.json"
if file_path.exists():
file_path.unlink()
return True
return False
def _save_to_shelf(self, document_id: str, position_name: str, position: RecursivePosition):
"""Save position to shelf database"""
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
with shelve.open(shelf_path) as shelf:
shelf[position_name] = position
def _load_from_shelf(self, document_id: str, position_name: str) -> Optional[RecursivePosition]:
"""Load position from shelf database"""
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
try:
with shelve.open(shelf_path) as shelf:
return shelf.get(position_name)
except Exception:
return None
def _list_shelf_positions(self, document_id: str) -> List[str]:
"""List positions in shelf database"""
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
try:
with shelve.open(shelf_path) as shelf:
return list(shelf.keys())
except Exception:
return []
def _delete_from_shelf(self, document_id: str, position_name: str) -> bool:
"""Delete position from shelf database"""
shelf_path = str(self.storage_dir / f"{document_id}.shelf")
try:
with shelve.open(shelf_path) as shelf:
if position_name in shelf:
del shelf[position_name]
return True
except Exception:
pass
return False
# Convenience functions for common position patterns
def create_word_position(chapter: int, block: int, word: int, char_offset: int = 0) -> RecursivePosition:
"""Create a position pointing to a specific word and character"""
return (PositionBuilder()
.chapter(chapter)
.block(block)
.paragraph()
.word(word, offset=char_offset)
.build())
def create_image_position(chapter: int, block: int, image_index: int = 0) -> RecursivePosition:
"""Create a position pointing to an image"""
return (PositionBuilder()
.chapter(chapter)
.block(block)
.image(image_index)
.build())
def create_table_cell_position(chapter: int, block: int, row: int, col: int, word: int = 0) -> RecursivePosition:
"""Create a position pointing to content in a table cell"""
return (PositionBuilder()
.chapter(chapter)
.block(block)
.table()
.table_row(row)
.table_cell(col)
.word(word)
.build())
def create_list_item_position(chapter: int, block: int, item: int, word: int = 0) -> RecursivePosition:
"""Create a position pointing to content in a list item"""
return (PositionBuilder()
.chapter(chapter)
.block(block)
.list()
.list_item(item)
.word(word)
.build())