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) -> Content: 

42 """Render the bouncing-block track for *frame* as styled content.""" 

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 return Content.assemble(*parts) 

51 

52 

53class ThinkingHeader(Static): 

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

55 

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

57 

58 def __init__(self) -> None: 

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

60 self._frame: int = 0 

61 self._timer: Timer | None = None 

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

63 

64 def on_mount(self) -> None: 

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

66 

67 def on_unmount(self) -> None: 

68 self.stop() 

69 

70 def stop(self) -> None: 

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

72 if self._timer is not None: 

73 self._timer.stop() 

74 self._timer = None 

75 

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

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

78 self._target = target 

79 

80 def _tick(self) -> None: 

81 self._frame += 1 

82 content = _frame_content(self._frame) 

83 if self._target is not None: 

84 self._target(content) 

85 else: 

86 self.update(content)