Coverage for src/lilbee/runtime/_splash_runner.py: 100%

1"""Standalone splash animation process: stdlib plus the shared wordmark. 

2 

3Launched as a subprocess by ``splash.start()``. Reads a pipe reference from 

4argv (the fd on POSIX, the pipe's OS handle on Windows) and animates until 

5the pipe signals EOF (parent closed its write end, or parent died). This 

6guarantees no orphan/zombie animation processes. 

7""" 

8 

9from __future__ import annotations 

10 

11import contextlib 

12import os 

13import select 

14import signal 

15import sys 

16import time 

17from collections.abc import Callable 

18from enum import IntEnum 

19 

20from lilbee.runtime.bee_logo import ( 

21 BEE_LINES, 

22 LOGO_WIDTH, 

23 ROSE_BRIGHT_XTERM, 

24 ROSE_DIM_XTERM, 

25 ROSE_MID_XTERM, 

26 xterm_fg, 

27) 

28 

29HIDE_CURSOR = "\033[?25l" 

30SHOW_CURSOR = "\033[?25h" 

31CLEAR_LINE = "\033[2K" 

32MOVE_UP = "\033[A" 

33 

34ROSE_BRIGHT = xterm_fg(ROSE_BRIGHT_XTERM) 

35ROSE_MID = xterm_fg(ROSE_MID_XTERM) 

36ROSE_DIM = xterm_fg(ROSE_DIM_XTERM) 

37RESET = "\033[0m" 

38 

39FRAME_INTERVAL = 0.15 

40STARTUP_DELAY = 0.08 

41POLL_INTERVAL = 0.01 

42 

43# Knight-rider bar uses three falloff steps after the bright head. 

44_BAR_FALLOFF_DENSE = 1 

45_BAR_FALLOFF_LIGHT = 2 

46 

47# Subprocess entry point expects exactly ``python -m ... <pipe_ref>`` (script name + 1 arg). 

48_EXPECTED_ARGV_LEN = 2 

49 

50# Console-mode bits for the legacy Windows console (Windows Terminal has VT on 

51# by default, conhost does not). 

52_STD_ERROR_HANDLE = -12 

53_ENABLE_VT_PROCESSING = 0x0004 

54_UTF8_CODEPAGE = 65001 

55 

56# Sent down the pipe by ``splash.dismiss()`` when the TUI takes over the 

57# terminal: the child must exit without writing anything (no frame clear, no 

58# cursor-show), because every byte would land on Textual's alt-screen and the 

59# cursor-show would leave a visible cursor for the whole TUI session. 

60TAKEOVER_BYTE = b"T" 

61 

62 

63class PipeSignal(IntEnum): 

64 """What the control pipe currently says the child should do.""" 

65 

66 OPEN = 0 

67 CLOSED = 1 

68 TAKEOVER = 2 

69 

70 

71COLOR_SEQUENCE = [ROSE_BRIGHT, ROSE_MID, ROSE_DIM, ROSE_MID] 

72 

73 

74def apply_color(line: str, color: str) -> str: 

75 """Apply color to non-empty parts of a line.""" 

76 if not line.strip(): 

77 return line 

78 return color + line + RESET 

79 

80 

81def build_logo_frames() -> list[list[str]]: 

82 """Pre-create 4 color-pulsed versions of the logo.""" 

83 return [[apply_color(line, color) for line in BEE_LINES] for color in COLOR_SEQUENCE] 

84 

85 

86def build_knight_rider_frames() -> list[str]: 

87 """Build a Knight Rider bar sweeping the full logo width and back.""" 

88 frames: list[str] = [] 

89 sweep_range = LOGO_WIDTH - 1 

90 total_frames = sweep_range * 2 

91 

92 for pos in range(total_frames): 

93 head_pos = pos if pos < sweep_range else (total_frames - pos) 

94 

95 bar = "" 

96 for i in range(LOGO_WIDTH): 

97 dist = abs(i - head_pos) 

98 if dist == 0: 

99 bar += ROSE_BRIGHT + "\u2593" + RESET 

100 elif dist == _BAR_FALLOFF_DENSE: 

101 bar += ROSE_DIM + "\u2592" + RESET 

102 elif dist == _BAR_FALLOFF_LIGHT: 

103 bar += ROSE_DIM + "\u2591" + RESET 

104 else: 

105 bar += " " 

106 frames.append(bar) 

107 

108 return frames 

109 

110 

111def left_pad() -> int: 

112 """Columns needed to centre the wordmark, matching the C bootstrap's formula. 

113 

114 The bootstrap frame this animation repaints in place is centred with 

115 ``(columns - LILBEE_LOGO_WIDTH) / 2``; diverging here would draw the two 

116 stages at different offsets and break the one-continuous-logo illusion. 

117 """ 

118 try: 

119 columns = os.get_terminal_size(2).columns 

120 except OSError: 

121 return 0 

122 return max((columns - LOGO_WIDTH) // 2, 0) 

123 

124 

125def render_frame(logo_lines: list[str], loading_bar: str, pad: int = 0) -> bytes: 

126 """Build a single frame as raw bytes for os.write().""" 

127 margin = " " * pad 

128 all_lines = [margin + line for line in logo_lines] 

129 all_lines += ["", f"{margin} {loading_bar}"] 

130 return ("\n".join(all_lines) + "\n").encode() 

131 

132 

133def move_up_and_clear(n: int) -> bytes: 

134 """ANSI sequence to move cursor up n lines and clear each one.""" 

135 return ((MOVE_UP + CLEAR_LINE) * n).encode() 

136 

137 

138def clear_screen(frame_height: int) -> bytes: 

139 """Erase the splash frame area and restore the cursor to the top. 

140 

141 Uses line-by-line clear (move-up + erase) instead of ``\\033[2J\\033[H`` 

142 so the subprocess never writes a cursor-home escape. A cursor-home 

143 would land on the Textual alt-screen if the TUI starts before the 

144 subprocess has finished, leaving a stuck cursor artifact at (0,0). 

145 """ 

146 return move_up_and_clear(frame_height) + SHOW_CURSOR.encode() 

147 

148 

149def _read_signal(pipe_fd: int) -> PipeSignal: 

150 """Read one byte: EOF/error means CLOSED, the takeover byte means TAKEOVER.""" 

151 try: 

152 data = os.read(pipe_fd, 1) 

153 except OSError: 

154 return PipeSignal.CLOSED 

155 if data == TAKEOVER_BYTE: 

156 return PipeSignal.TAKEOVER 

157 return PipeSignal.CLOSED if len(data) == 0 else PipeSignal.OPEN 

158 

159 

160def _poll_pipe_win32(pipe_fd: int) -> PipeSignal: # pragma: no cover Windows-only 

161 """Win32 pipe poll using PeekNamedPipe.""" 

162 import ctypes 

163 import msvcrt 

164 

165 try: 

166 handle = msvcrt.get_osfhandle(pipe_fd) # type: ignore[attr-defined] 

167 except OSError: 

168 return PipeSignal.CLOSED # bad fd, pipe is gone 

169 avail = ctypes.c_ulong(0) 

170 if not ctypes.windll.kernel32.PeekNamedPipe( # type: ignore[attr-defined] 

171 handle, None, 0, None, ctypes.byref(avail), None 

172 ): 

173 return PipeSignal.CLOSED 

174 if avail.value == 0: 

175 return PipeSignal.OPEN 

176 return _read_signal(pipe_fd) 

177 

178 

179def _poll_pipe_posix(pipe_fd: int) -> PipeSignal: # pragma: no cover POSIX-only 

180 """POSIX pipe poll using select.""" 

181 try: 

182 readable, _, _ = select.select([pipe_fd], [], [], 0) 

183 except (ValueError, OSError): 

184 return PipeSignal.CLOSED 

185 if not readable: 

186 return PipeSignal.OPEN 

187 return _read_signal(pipe_fd) 

188 

189 

190def poll_pipe(pipe_fd: int) -> PipeSignal: 

191 """Check the control pipe without blocking.""" 

192 if sys.platform == "win32": 

193 return _poll_pipe_win32(pipe_fd) # pragma: no cover Windows-only 

194 return _poll_pipe_posix(pipe_fd) # pragma: no cover POSIX-only 

195 

196 

197def _open_pipe_ref(ref: int) -> int: 

198 """Turn the argv pipe reference into a readable fd. 

199 

200 POSIX passes the fd itself (pass_fds keeps the number). Windows cannot 

201 pass fds, so the parent sends the pipe's OS handle number and the child 

202 reopens it as a fd here. 

203 """ 

204 if sys.platform == "win32": # pragma: no cover Windows-only 

205 import msvcrt 

206 

207 return msvcrt.open_osfhandle(ref, os.O_RDONLY) 

208 return ref 

209 

210 

211def _setup_console_win32() -> int: # pragma: no cover Windows-only 

212 """Make the console render this process's output: VT on, UTF-8 out. 

213 

214 ANSI processing is off on the legacy console (Windows Terminal has it 

215 on), and the default output codepage mangles the loading bar's block 

216 characters, which the frames carry as UTF-8 bytes. Returns the previous 

217 codepage so the caller can restore it; 0 when there is no console 

218 (GetConsoleMode fails -- the parent only starts the splash on a tty). 

219 """ 

220 if sys.platform != "win32": 

221 return 0 

222 import ctypes 

223 

224 kernel32 = ctypes.windll.kernel32 

225 handle = kernel32.GetStdHandle(_STD_ERROR_HANDLE) 

226 mode = ctypes.c_ulong(0) 

227 if not kernel32.GetConsoleMode(handle, ctypes.byref(mode)): 

228 return 0 

229 kernel32.SetConsoleMode(handle, mode.value | _ENABLE_VT_PROCESSING) 

230 previous_cp = int(kernel32.GetConsoleOutputCP()) 

231 kernel32.SetConsoleOutputCP(_UTF8_CODEPAGE) 

232 return previous_cp 

233 

234 

235def _restore_console_win32(previous_cp: int) -> None: # pragma: no cover Windows-only 

236 """Put the console codepage back so the parent shell keeps its own.""" 

237 if sys.platform != "win32" or not previous_cp: 

238 return 

239 import ctypes 

240 

241 ctypes.windll.kernel32.SetConsoleOutputCP(previous_cp) 

242 

243 

244def animation_loop(pipe_fd: int) -> None: 

245 """Run the animation, exiting when the pipe signals EOF or takeover. 

246 

247 A plain EOF (``splash.stop()``, parent death) clears the frame and 

248 restores the cursor so the shell gets a clean terminal back. A takeover 

249 byte (``splash.dismiss()``) means Textual owns the terminal: exit without 

250 writing a single byte more. 

251 """ 

252 fd = 2 # stderr 

253 

254 logo_frames = build_logo_frames() 

255 knight_frames = build_knight_rider_frames() 

256 pad = left_pad() 

257 frame_height = len(BEE_LINES) + 2 

258 

259 got_signal = False 

260 pipe_signal = PipeSignal.OPEN 

261 

262 if sys.platform != "win32": # pragma: no cover - POSIX-only SIGTERM handler 

263 

264 def handle_term(signum: int, frame: object) -> None: 

265 nonlocal got_signal 

266 got_signal = True 

267 

268 signal.signal(signal.SIGTERM, handle_term) 

269 

270 def should_stop() -> bool: 

271 nonlocal pipe_signal 

272 if pipe_signal is PipeSignal.OPEN: 

273 pipe_signal = poll_pipe(pipe_fd) 

274 return got_signal or pipe_signal is not PipeSignal.OPEN 

275 

276 if _stopped_during_startup_delay(should_stop): 

277 return # nothing drawn yet, nothing to clean up 

278 

279 previous_cp = 0 

280 if sys.platform == "win32": # pragma: no cover - Windows-only console setup 

281 previous_cp = _setup_console_win32() 

282 try: 

283 os.write(fd, HIDE_CURSOR.encode()) 

284 _animate_frames(fd, logo_frames, knight_frames, pad, frame_height, should_stop) 

285 except OSError: 

286 pass # parent closed the splash pipe; just stop drawing 

287 finally: 

288 if pipe_signal is not PipeSignal.TAKEOVER: 

289 with contextlib.suppress(OSError): 

290 os.write(fd, clear_screen(frame_height)) 

291 if sys.platform == "win32": # pragma: no cover - Windows-only console restore 

292 _restore_console_win32(previous_cp) 

293 

294 

295def _stopped_during_startup_delay(should_stop: Callable[[], bool]) -> bool: 

296 """Poll through the startup delay; True when the splash should not draw.""" 

297 for _ in range(int(STARTUP_DELAY / POLL_INTERVAL)): 

298 if should_stop(): 

299 return True 

300 time.sleep(POLL_INTERVAL) 

301 return False 

302 

303 

304def _animate_frames( 

305 fd: int, 

306 logo_frames: list[list[str]], 

307 knight_frames: list[str], 

308 pad: int, 

309 frame_height: int, 

310 should_stop: Callable[[], bool], 

311) -> None: 

312 """Draw pulse/sweep frames until *should_stop* reports a stop condition.""" 

313 frame_idx = 0 

314 while not should_stop(): 

315 logo = logo_frames[frame_idx % len(logo_frames)] 

316 knight = knight_frames[frame_idx % len(knight_frames)] 

317 os.write(fd, render_frame(logo, knight, pad)) 

318 

319 for _ in range(int(FRAME_INTERVAL / POLL_INTERVAL)): 

320 if should_stop(): 

321 break 

322 time.sleep(POLL_INTERVAL) 

323 

324 if not should_stop(): 

325 os.write(fd, move_up_and_clear(frame_height)) # pragma: no cover 

326 

327 frame_idx += 1 

328 

329 

330def main() -> None: 

331 """Entry point when run as ``python -m lilbee.runtime._splash_runner <pipe_ref>``.""" 

332 if len(sys.argv) != _EXPECTED_ARGV_LEN: 

333 sys.exit(1) 

334 

335 try: 

336 pipe_fd = _open_pipe_ref(int(sys.argv[1])) 

337 except (ValueError, OSError): 

338 sys.exit(1) 

339 

340 try: 

341 animation_loop(pipe_fd) 

342 finally: 

343 with contextlib.suppress(OSError): 

344 os.close(pipe_fd) 

345 

346 

347if __name__ == "__main__": 

348 main()