Coverage for src/lilbee/cli/tui/widgets/thinking_header.py: 100%

1"""Animated Knight-Rider scanner-bar header for the assistant bubble.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from pathlib import Path 

7from typing import ClassVar 

8 

9from textual.content import Content 

10from textual.timer import Timer 

11from textual.widgets import Static 

12 

13_CSS_FILE = Path(__file__).parent / "thinking_header.tcss" 

14 

15_BLOCK_FILLED = "▰" 

16_BLOCK_EMPTY = "▱" 

17_TRACK_CELLS = 9 

18"""Visible width of the bouncing track. Wider than the old 5-cell snake 

19so the back-and-forth motion reads as deliberate scanning.""" 

20 

21_FRAME_INTERVAL = 0.1 

22"""100 ms per frame: motion without renderer thrash on large chat logs.""" 

23 

24_DIM_STYLE = "$text-muted" 

25_FILL_STYLE = "bold $success" 

26 

27 

28def _bounce_position(frame: int) -> int: 

29 """Return the lit cell index for *frame* in a Knight-Rider bounce. 

30 

31 Cycle length is ``2 * (cells - 1)``: forward sweep 0..cells-1, then 

32 backward sweep cells-2..1, repeating. 

33 """ 

34 cycle = 2 * (_TRACK_CELLS - 1) 

35 step = frame % cycle 

36 if step < _TRACK_CELLS: 

37 return step 

38 return cycle - step 

39 

40 

41def _frame_content(frame: int, detail: str = "") -> Content: 

42 """Render the bouncing-block track for *frame*, with an optional detail suffix.""" 

43 pos = _bounce_position(frame) 

44 parts: list[Content] = [] 

45 for i in range(_TRACK_CELLS): 

46 if i == pos: 

47 parts.append(Content.styled(_BLOCK_FILLED, _FILL_STYLE)) 

48 else: 

49 parts.append(Content.styled(_BLOCK_EMPTY, _DIM_STYLE)) 

50 if detail: 

51 parts.append(Content.styled(f" {detail}", _DIM_STYLE)) 

52 return Content.assemble(*parts) 

53 

54 

55class ThinkingHeader(Static): 

56 """Single Static that animates a Knight-Rider bouncing block.""" 

57 

58 DEFAULT_CSS: ClassVar[str] = _CSS_FILE.read_text(encoding="utf-8") 

59 

60 def __init__(self) -> None: 

61 super().__init__(_frame_content(0), classes="thinking-header") 

62 self._frame: int = 0 

63 self._detail: str = "" 

64 self._timer: Timer | None = None 

65 self._target: Callable[[Content], None] | None = None 

66 

67 def on_mount(self) -> None: 

68 self._timer = self.set_interval(_FRAME_INTERVAL, self._tick) 

69 

70 def on_unmount(self) -> None: 

71 self.stop() 

72 

73 def stop(self) -> None: 

74 """Stop the animator timer; safe to call repeatedly.""" 

75 if self._timer is not None: 

76 self._timer.stop() 

77 self._timer = None 

78 

79 def redirect_to(self, target: Callable[[Content], None] | None) -> None: 

80 """Send each frame's content to *target* instead of painting self.""" 

81 self._target = target 

82 

83 def set_status(self, detail: str) -> None: 

84 """Show *detail* beside the scanner track from the next frame on.""" 

85 self._detail = detail 

86 

87 def _tick(self) -> None: 

88 self._frame += 1 

89 content = _frame_content(self._frame, self._detail) 

90 if self._target is not None: 

91 self._target(content) 

92 else: 

93 self.update(content)