Coverage for src / lilbee / catalog / download_progress.py: 100%

1"""Download progress callback plumbing shared by all surfaces. 

2 

3The TUI runs under Textual which owns the terminal; tqdm output to 

4stderr/stdout corrupts the screen. This module provides a tqdm subclass 

5(``_CallbackProgressBar``) that suppresses terminal output and forwards 

6progress to a plain ``Callable[[int, int], None]`` callback. The 

7``_ProgressTracker`` wrapper detects whether progress events actually 

8fired so the TUI can detect a cache-hit (no progress events) and render 

9``"already downloaded"`` instead of leaving the bar at 0%. 

10 

11``make_download_callback`` is the public entry point used by every 

12surface to convert raw bytes-progress into ``DownloadProgress`` events. 

13""" 

14 

15from __future__ import annotations 

16 

17import io 

18import threading 

19import time 

20from collections.abc import Callable 

21from typing import Any 

22 

23from tqdm.auto import tqdm as _base_tqdm 

24 

25from lilbee.catalog.models import DownloadProgress 

26 

27ProgressCallback = Callable[[int, int], None] 

28_BYTES_PER_MB = 1024 * 1024 

29 

30 

31def make_download_callback( 

32 on_update: Callable[[DownloadProgress], None], 

33 *, 

34 throttle_interval: float = 0.1, 

35) -> ProgressCallback: 

36 """Build a download progress callback that converts bytes to human-readable state. 

37 *on_update(progress: DownloadProgress)* is called at most once per 

38 ``throttle_interval`` seconds with a float percentage (0.0 to 100.0), a 

39 ``"<done>/<total> MB"`` detail string, and a cache-hit flag. Both the 

40 catalog and setup screens use this so byte-to-MB conversion and 

41 cache-hit detection aren't duplicated. 

42 """ 

43 last_update_time = 0.0 

44 seen_partial = False 

45 

46 def _on_progress(downloaded: int, total: int) -> None: 

47 nonlocal last_update_time, seen_partial 

48 

49 if total > 0 and downloaded >= total and not seen_partial: 

50 on_update( 

51 DownloadProgress(percent=100.0, detail="already downloaded", is_cache_hit=True) 

52 ) 

53 return 

54 seen_partial = True 

55 

56 now = time.monotonic() 

57 if now - last_update_time < throttle_interval: 

58 return 

59 last_update_time = now 

60 

61 mb_done = downloaded / _BYTES_PER_MB 

62 if total > 0: 

63 pct = min(downloaded * 100.0 / total, 100.0) 

64 mb_total = total / _BYTES_PER_MB 

65 on_update( 

66 DownloadProgress( 

67 percent=pct, 

68 detail=f"{mb_done:.0f}/{mb_total:.0f} MB", 

69 is_cache_hit=False, 

70 ) 

71 ) 

72 else: 

73 on_update(DownloadProgress(percent=0.0, detail=f"{mb_done:.0f} MB", is_cache_hit=False)) 

74 

75 return _on_progress 

76 

77 

78class _CallbackProgressBar(_base_tqdm): 

79 """tqdm subclass that forwards progress to a plain callback. 

80 Fully suppresses terminal output by disabling tqdm rendering and redirecting 

81 its file handle to a devnull sink: prevents ANSI escape sequences from leaking 

82 into Textual's managed terminal. 

83 

84 Overrides ``get_lock`` to return a threading lock instead of tqdm's default 

85 multiprocessing lock. Vanilla tqdm acquires ``self._lock`` even on the 

86 ``disable=True`` path (std.py:988), and the multiprocessing lock's lazy init 

87 raises ``ValueError`` when ``sys.stderr.fileno() == -1`` (Textual, Jupyter, 

88 pytest capture). A thread lock sidesteps that fd handling entirely. 

89 """ 

90 

91 _lock = threading.RLock() 

92 _callback: Any = None 

93 

94 @classmethod 

95 def get_lock(cls) -> threading.RLock: 

96 return cls._lock 

97 

98 def __init__(self, *args: Any, **kwargs: Any): 

99 kwargs["disable"] = True 

100 kwargs["file"] = io.StringIO() # absorb any accidental tqdm output 

101 super().__init__(*args, **kwargs) 

102 self._cumulative = 0 

103 

104 def update(self, n: float = 1) -> bool | None: 

105 self._cumulative += int(n) 

106 if self._callback is not None: 

107 total = self.total if self.total is not None else 0 

108 self._callback(int(self._cumulative), int(total)) 

109 return None 

110 

111 

112class _ProgressTracker: 

113 """Wraps a tqdm_class to detect whether progress updates actually fired.""" 

114 

115 def __init__(self, callback: Any) -> None: 

116 self.was_used = False 

117 self._callback = callback 

118 

119 def make_tqdm_class(self) -> type[_base_tqdm]: 

120 tracker = self 

121 

122 class _Cls(_CallbackProgressBar): 

123 _callback = staticmethod(tracker._callback) 

124 

125 def update(self, n: float = 1) -> bool | None: 

126 tracker.was_used = True 

127 return super().update(n) 

128 

129 return _Cls