Coverage for pyWebLayout/layout/page_buffer.py: 85%

110 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-08 20:34 +0000

1""" 

2Page caching for ereader navigation. 

3 

4`PageBuffer` is an LRU cache of rendered pages plus the position links between 

5them; `BufferedPageRenderer` wraps it around a `BidirectionalLayouter`. 

6 

7This module used to render pages ahead of time in a `ProcessPoolExecutor`. That 

8never worked and has been removed — see S12 in docs/LAYOUT_REMEDIATION_SPEC.md 

9and R1/R2 in docs/ARCHITECTURE_REVIEW.md. In short: the worker returned 

10`pickle.dumps(page)`, and a Page holds a live PIL canvas, which is not 

11picklable, so every job failed and the result was discarded. The cost — four 

12interpreter copies and the whole block list shipped per job — was paid in full 

13for no benefit. On Python 3.14, where the default start method became 

14`forkserver`, submitting from module-level code raised outright. 

15 

16Prefetch is not needed. Measured on the tests/data Wikipedia fixture (411 

17blocks) with the text caches warm, one page render costs: 

18 

19 800x600 p50 8.8 ms p95 15.4 ms 

20 1072x1448 p50 13.8 ms p95 56.1 ms 

21 

22A page turn is cheaper than the IPC that was meant to hide it. If a slower 

23target device ever changes that, the fallback is a synchronous `readahead()` 

24method on this class, or a single worker *thread* — layout is PIL-bound and PIL 

25releases the GIL — not a process pool. Making the concrete tree picklable 

26(Page -> Line -> Text -> Font -> FreeTypeFont) is a large surface area to 

27maintain for a cache. 

28""" 

29 

30from __future__ import annotations 

31from typing import Dict, Optional, List, Tuple, Any 

32from collections import OrderedDict 

33 

34from .ereader_layout import RenderingPosition, BidirectionalLayouter, FontFamilyOverride 

35from pyWebLayout.concrete.page import Page 

36from pyWebLayout.abstract.block import Block 

37from pyWebLayout.style.page_style import PageStyle 

38from pyWebLayout.style.fonts import BundledFont 

39 

40 

41class PageBuffer: 

42 """ 

43 LRU cache of rendered pages, with separate forward and backward buffers and 

44 the position links between adjacent pages. 

45 """ 

46 

47 def __init__(self, buffer_size: int = 5): 

48 """ 

49 Initialize the page buffer. 

50 

51 Args: 

52 buffer_size: Number of pages to cache in each direction 

53 """ 

54 self.buffer_size = buffer_size 

55 

56 # LRU caches for forward and backward pages 

57 self.forward_buffer: OrderedDict[RenderingPosition, Page] = OrderedDict() 

58 self.backward_buffer: OrderedDict[RenderingPosition, Page] = OrderedDict() 

59 

60 # Position tracking for next/previous positions 

61 self.position_map: Dict[RenderingPosition, 

62 RenderingPosition] = {} # current -> next 

63 self.reverse_position_map: Dict[RenderingPosition, 

64 RenderingPosition] = {} # current -> previous 

65 

66 # Document state 

67 self.blocks: Optional[List[Block]] = None 

68 self.page_style: Optional[PageStyle] = None 

69 self.current_font_scale: float = 1.0 

70 self.current_font_family: Optional[BundledFont] = None 

71 

72 def initialize( 

73 self, 

74 blocks: List[Block], 

75 page_style: PageStyle, 

76 font_scale: float = 1.0, 

77 font_family: Optional[BundledFont] = None): 

78 """ 

79 Initialize the buffer with document blocks and page style. 

80 

81 Args: 

82 blocks: Document blocks to render 

83 page_style: Page styling configuration 

84 font_scale: Current font scaling factor 

85 font_family: Optional font family override 

86 """ 

87 self.blocks = blocks 

88 self.page_style = page_style 

89 self.current_font_scale = font_scale 

90 self.current_font_family = font_family 

91 

92 def get_page(self, position: RenderingPosition) -> Optional[Page]: 

93 """ 

94 Get a cached page if available. 

95 

96 Args: 

97 position: Position to get page for 

98 

99 Returns: 

100 Cached page or None if not available 

101 """ 

102 # Check forward buffer first 

103 if position in self.forward_buffer: 

104 # Move to end (most recently used) 

105 page = self.forward_buffer.pop(position) 

106 self.forward_buffer[position] = page 

107 return page 

108 

109 # Check backward buffer 

110 if position in self.backward_buffer: 

111 # Move to end (most recently used) 

112 page = self.backward_buffer.pop(position) 

113 self.backward_buffer[position] = page 

114 return page 

115 

116 return None 

117 

118 def cache_page( 

119 self, 

120 position: RenderingPosition, 

121 page: Page, 

122 next_position: Optional[RenderingPosition] = None, 

123 is_backward: bool = False): 

124 """ 

125 Cache a rendered page with LRU eviction. 

126 

127 Args: 

128 position: Position of the page 

129 page: Rendered page to cache 

130 next_position: Position of the next page (for forward navigation) 

131 is_backward: Whether this is a backward-rendered page 

132 """ 

133 target_buffer = self.backward_buffer if is_backward else self.forward_buffer 

134 

135 # Add to cache 

136 target_buffer[position] = page 

137 

138 # Track position relationships 

139 if next_position: 

140 if is_backward: 

141 self.reverse_position_map[next_position] = position 

142 else: 

143 self.position_map[position] = next_position 

144 

145 # Evict oldest if buffer is full 

146 if len(target_buffer) > self.buffer_size: 

147 oldest_pos, _ = target_buffer.popitem(last=False) 

148 # Clean up position maps 

149 self.position_map.pop(oldest_pos, None) 

150 self.reverse_position_map.pop(oldest_pos, None) 

151 

152 def invalidate_all(self): 

153 """Clear all cached pages""" 

154 self.forward_buffer.clear() 

155 self.backward_buffer.clear() 

156 self.position_map.clear() 

157 self.reverse_position_map.clear() 

158 

159 def set_font_scale(self, font_scale: float): 

160 """ 

161 Update font scale and invalidate cache. 

162 

163 Args: 

164 font_scale: New font scaling factor 

165 """ 

166 if font_scale != self.current_font_scale: 

167 self.current_font_scale = font_scale 

168 self.invalidate_all() 

169 

170 def set_font_family(self, font_family: Optional[BundledFont]): 

171 """ 

172 Update font family and invalidate cache. 

173 

174 Args: 

175 font_family: New font family (None = use original fonts) 

176 """ 

177 if font_family != self.current_font_family: 

178 self.current_font_family = font_family 

179 self.invalidate_all() 

180 

181 def get_cache_stats(self) -> Dict[str, Any]: 

182 """Get cache statistics for debugging/monitoring""" 

183 return { 

184 'forward_buffer_size': len(self.forward_buffer), 

185 'backward_buffer_size': len(self.backward_buffer), 

186 'position_mappings': len(self.position_map), 

187 'reverse_position_mappings': len(self.reverse_position_map), 

188 'current_font_scale': self.current_font_scale, 

189 'current_font_family': self.current_font_family.value if self.current_font_family else None 

190 } 

191 

192 def shutdown(self): 

193 """ 

194 Release cached pages. 

195 

196 Cheap and idempotent. There is deliberately no __del__ calling this: 

197 blocking work in a finaliser is what deadlocked the interpreter at exit 

198 while the process pool existed. 

199 """ 

200 self.invalidate_all() 

201 

202 

203class BufferedPageRenderer: 

204 """ 

205 High-level interface for page rendering with an LRU cache in front of the 

206 layouter. 

207 """ 

208 

209 def __init__(self, 

210 blocks: List[Block], 

211 page_style: PageStyle, 

212 buffer_size: int = 5, 

213 page_size: Tuple[int, 

214 int] = (800, 

215 600), 

216 font_family: Optional[BundledFont] = None): 

217 """ 

218 Initialize the buffered renderer. 

219 

220 Args: 

221 blocks: Document blocks to render 

222 page_style: Page styling configuration 

223 buffer_size: Number of pages to cache in each direction 

224 page_size: Page size (width, height) in pixels 

225 font_family: Optional font family override 

226 """ 

227 # Create font family override if specified 

228 font_family_override = FontFamilyOverride(font_family) if font_family else None 

229 

230 self.layouter = BidirectionalLayouter(blocks, page_style, page_size, font_family_override=font_family_override) 

231 self.buffer = PageBuffer(buffer_size) 

232 self.buffer.initialize(blocks, page_style, font_family=font_family) 

233 self.page_size = page_size 

234 self.blocks = blocks 

235 self.page_style = page_style 

236 

237 self.current_position = RenderingPosition() 

238 self.font_scale = 1.0 

239 self.font_family = font_family 

240 

241 def render_page(self, position: RenderingPosition, 

242 font_scale: float = 1.0) -> Tuple[Page, RenderingPosition]: 

243 """ 

244 Render a page, serving it from cache when possible. 

245 

246 Args: 

247 position: Position to render from 

248 font_scale: Font scaling factor 

249 

250 Returns: 

251 Tuple of (rendered_page, next_position) 

252 """ 

253 # Update font scale if changed 

254 if font_scale != self.font_scale: 

255 self.font_scale = font_scale 

256 self.buffer.set_font_scale(font_scale) 

257 

258 # Check cache first 

259 cached_page = self.buffer.get_page(position) 

260 if cached_page: 

261 # Only use the cache if we also know where the next page starts; 

262 # otherwise fall through and compute it. 

263 next_pos = self.buffer.position_map.get(position) 

264 if next_pos is not None: 

265 return cached_page, next_pos 

266 

267 # Render the page directly 

268 page, next_pos = self.layouter.render_page_forward(position, font_scale) 

269 

270 # Cache the result 

271 self.buffer.cache_page(position, page, next_pos) 

272 

273 return page, next_pos 

274 

275 def render_page_backward(self, 

276 end_position: RenderingPosition, 

277 font_scale: float = 1.0) -> Tuple[Page, 

278 RenderingPosition]: 

279 """ 

280 Render a page ending at the given position, serving it from cache when 

281 possible. 

282 

283 Args: 

284 end_position: Position where page should end 

285 font_scale: Font scaling factor 

286 

287 Returns: 

288 Tuple of (rendered_page, start_position) 

289 """ 

290 # Update font scale if changed 

291 if font_scale != self.font_scale: 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true

292 self.font_scale = font_scale 

293 self.buffer.set_font_scale(font_scale) 

294 

295 # Check cache first 

296 cached_page = self.buffer.get_page(end_position) 

297 if cached_page: 297 ↛ 300line 297 didn't jump to line 300 because the condition on line 297 was never true

298 # Only use the cache if we also know where the previous page 

299 # starts; otherwise fall through and compute it. 

300 prev_pos = self.buffer.reverse_position_map.get(end_position) 

301 if prev_pos is not None: 

302 return cached_page, prev_pos 

303 

304 # Render the page directly 

305 page, start_pos = self.layouter.render_page_backward(end_position, font_scale) 

306 

307 # Cache the result 

308 self.buffer.cache_page(start_pos, page, end_position, is_backward=True) 

309 

310 return page, start_pos 

311 

312 def set_font_family(self, font_family: Optional[BundledFont]): 

313 """ 

314 Change the font family and invalidate cache. 

315 

316 Args: 

317 font_family: New font family (None = use original fonts) 

318 """ 

319 if font_family != self.font_family: 

320 self.font_family = font_family 

321 

322 # Update buffer 

323 self.buffer.set_font_family(font_family) 

324 

325 # Recreate layouter with new font family override 

326 font_family_override = FontFamilyOverride(font_family) if font_family else None 

327 self.layouter = BidirectionalLayouter( 

328 self.blocks, 

329 self.page_style, 

330 self.page_size, 

331 font_family_override=font_family_override 

332 ) 

333 

334 def get_font_family(self) -> Optional[BundledFont]: 

335 """Get the current font family override""" 

336 return self.font_family 

337 

338 def get_cache_stats(self) -> Dict[str, Any]: 

339 """Get cache statistics""" 

340 return self.buffer.get_cache_stats() 

341 

342 def shutdown(self): 

343 """Release cached pages""" 

344 self.buffer.shutdown()