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 fd from argv 

4and animates until the pipe signals EOF (parent closed its write end, or 

5parent died). This guarantees no orphan/zombie animation processes. 

6""" 

7 

8from __future__ import annotations 

9 

10import contextlib 

11import os 

12import select 

13import signal 

14import sys 

15import time 

16from collections.abc import Callable 

17from enum import IntEnum 

18 

19from lilbee.runtime.bee_logo import ( 

20 BEE_LINES, 

21 LOGO_WIDTH, 

22 ROSE_BRIGHT_XTERM, 

23 ROSE_DIM_XTERM, 

24 ROSE_MID_XTERM, 

25 xterm_fg, 

26) 

27 

28HIDE_CURSOR = "\033[?25l" 

29SHOW_CURSOR = "\033[?25h" 

30CLEAR_LINE = "\033[2K" 

31MOVE_UP = "\033[A" 

32 

33ROSE_BRIGHT = xterm_fg(ROSE_BRIGHT_XTERM) 

34ROSE_MID = xterm_fg(ROSE_MID_XTERM) 

35ROSE_DIM = xterm_fg(ROSE_DIM_XTERM) 

36RESET = "\033[0m" 

37 

38FRAME_INTERVAL = 0.15 

39STARTUP_DELAY = 0.08 

40POLL_INTERVAL = 0.01 

41 

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

43_BAR_FALLOFF_DENSE = 1 

44_BAR_FALLOFF_LIGHT = 2 

45 

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

47_EXPECTED_ARGV_LEN = 2 

48 

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

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

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

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

53TAKEOVER_BYTE = b"T" 

54 

55 

56class PipeSignal(IntEnum): 

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

58 

59 OPEN = 0 

60 CLOSED = 1 

61 TAKEOVER = 2 

62 

63 

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

65 

66 

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

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

69 if not line.strip(): 

70 return line 

71 return color + line + RESET 

72 

73 

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

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

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

77 

78 

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

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

81 frames: list[str] = [] 

82 sweep_range = LOGO_WIDTH - 1 

83 total_frames = sweep_range * 2 

84 

85 for pos in range(total_frames): 

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

87 

88 bar = "" 

89 for i in range(LOGO_WIDTH): 

90 dist = abs(i - head_pos) 

91 if dist == 0: 

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

93 elif dist == _BAR_FALLOFF_DENSE: 

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

95 elif dist == _BAR_FALLOFF_LIGHT: 

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

97 else: 

98 bar += " " 

99 frames.append(bar) 

100 

101 return frames 

102 

103 

104def left_pad() -> int: 

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

106 

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

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

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

110 """ 

111 try: 

112 columns = os.get_terminal_size(2).columns 

113 except OSError: 

114 return 0 

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

116 

117 

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

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

120 margin = " " * pad 

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

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

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

124 

125 

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

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

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

129 

130 

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

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

133 

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

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

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

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

138 """ 

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

140 

141 

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

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

144 try: 

145 data = os.read(pipe_fd, 1) 

146 except OSError: 

147 return PipeSignal.CLOSED 

148 if data == TAKEOVER_BYTE: 

149 return PipeSignal.TAKEOVER 

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

151 

152 

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

154 """Win32 pipe poll using PeekNamedPipe.""" 

155 import ctypes 

156 import msvcrt 

157 

158 try: 

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

160 except OSError: 

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

162 avail = ctypes.c_ulong(0) 

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

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

165 ): 

166 return PipeSignal.CLOSED 

167 if avail.value == 0: 

168 return PipeSignal.OPEN 

169 return _read_signal(pipe_fd) 

170 

171 

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

173 """POSIX pipe poll using select.""" 

174 try: 

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

176 except (ValueError, OSError): 

177 return PipeSignal.CLOSED 

178 if not readable: 

179 return PipeSignal.OPEN 

180 return _read_signal(pipe_fd) 

181 

182 

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

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

185 if sys.platform == "win32": 

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

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

188 

189 

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

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

192 

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

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

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

196 writing a single byte more. 

197 """ 

198 fd = 2 # stderr 

199 

200 logo_frames = build_logo_frames() 

201 knight_frames = build_knight_rider_frames() 

202 pad = left_pad() 

203 frame_height = len(BEE_LINES) + 2 

204 

205 got_signal = False 

206 pipe_signal = PipeSignal.OPEN 

207 

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

209 

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

211 nonlocal got_signal 

212 got_signal = True 

213 

214 signal.signal(signal.SIGTERM, handle_term) 

215 

216 def should_stop() -> bool: 

217 nonlocal pipe_signal 

218 if pipe_signal is PipeSignal.OPEN: 

219 pipe_signal = poll_pipe(pipe_fd) 

220 return got_signal or pipe_signal is not PipeSignal.OPEN 

221 

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

223 if should_stop(): 

224 return # nothing drawn yet, nothing to clean up 

225 time.sleep(POLL_INTERVAL) 

226 

227 try: 

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

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

230 except OSError: 

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

232 finally: 

233 if pipe_signal is not PipeSignal.TAKEOVER: 

234 with contextlib.suppress(OSError): 

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

236 

237 

238def _animate_frames( 

239 fd: int, 

240 logo_frames: list[list[str]], 

241 knight_frames: list[str], 

242 pad: int, 

243 frame_height: int, 

244 should_stop: Callable[[], bool], 

245) -> None: 

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

247 frame_idx = 0 

248 while not should_stop(): 

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

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

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

252 

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

254 if should_stop(): 

255 break 

256 time.sleep(POLL_INTERVAL) 

257 

258 if not should_stop(): 

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

260 

261 frame_idx += 1 

262 

263 

264def main() -> None: 

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

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

267 sys.exit(1) 

268 

269 try: 

270 pipe_fd = int(sys.argv[1]) 

271 except ValueError: 

272 sys.exit(1) 

273 

274 try: 

275 animation_loop(pipe_fd) 

276 finally: 

277 with contextlib.suppress(OSError): 

278 os.close(pipe_fd) 

279 

280 

281if __name__ == "__main__": 

282 main()