Coverage for src / monte_neo / cli / progress.py: 61%

1"""Progress tracking module. 

2 

3Progress bars and ETA estimation. 

4""" 

5 

6from __future__ import annotations 

7 

8import time 

9 

10from rich.progress import ( 

11 BarColumn, 

12 Progress, 

13 SpinnerColumn, 

14 TaskID, 

15 TextColumn, 

16 TimeElapsedColumn, 

17 TimeRemainingColumn, 

18) 

19 

20from monte_neo.utils.console import console 

21 

22 

23class ProgressTracker: 

24 """Track progress with ETA estimation. 

25  

26 V0.0.5 Development Stages: 

27 1. Metal Foundation: C++/Metal bridge and BaseKernel. [DONE] 

28 2. Indicators Library: SMA, EMA, RSI, ATR in MSL. [DONE] 

29 3. Optimization: GPU Grid Search and Walk-Forward. [DONE] 

30 4. Production Gate: Robustness scoring and Certification. [DONE] 

31 """ 

32 

33 def __init__(self) -> None: 

34 """Initialize tracker.""" 

35 self._progress: Progress | None = None 

36 self._task_id: TaskID | None = None 

37 self._start_time: float = 0 

38 self._total: int = 0 

39 

40 def start(self, total: int, description: str = "Processing...") -> None: 

41 """Start progress tracking. 

42 

43 Args: 

44 total: Total number of items. 

45 description: Progress description. 

46 """ 

47 self._total = total 

48 self._start_time = time.time() 

49 

50 self._progress = Progress( 

51 SpinnerColumn(), 

52 TextColumn("[bold blue]{task.description}"), 

53 BarColumn(bar_width=40), 

54 TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), 

55 TextColumn("•"), 

56 TimeElapsedColumn(), 

57 TextColumn("•"), 

58 TimeRemainingColumn(), 

59 TextColumn("[dim]{task.fields[status]}"), 

60 console=console, 

61 transient=True, 

62 ) 

63 

64 self._progress.start() 

65 self._task_id = self._progress.add_task( 

66 description, 

67 total=total, 

68 status="", 

69 ) 

70 

71 def update(self, current: int, total: int, status: str = "") -> None: 

72 """Update progress. 

73 

74 Args: 

75 current: Current position. 

76 total: Total items. 

77 status: Status message. 

78 """ 

79 if self._progress and self._task_id is not None: 

80 # Ensure we don't exceed 100% in display 

81 val = min(current, total) 

82 self._progress.update( 

83 self._task_id, 

84 completed=val, 

85 total=total, 

86 status=status, 

87 refresh=True if val >= total else False, 

88 ) 

89 

90 def stop(self) -> None: 

91 """Stop progress tracking.""" 

92 if self._progress: 

93 self._progress.stop() 

94 self._progress = None 

95 self._task_id = None 

96 

97 def get_eta_minutes(self, current: int) -> float: 

98 """Get estimated time remaining in minutes. 

99 

100 Args: 

101 current: Current position. 

102 

103 Returns: 

104 Estimated minutes remaining. 

105 """ 

106 if current == 0: 

107 return 0 

108 

109 elapsed = time.time() - self._start_time 

110 rate = current / elapsed 

111 remaining = self._total - current 

112 

113 if rate > 0: 

114 return (remaining / rate) / 60 

115 

116 return 0 

117 

118 

119def estimate_generation_time( 

120 data_size: int, 

121 iterations: int, 

122 mc_methods: int, 

123) -> str: 

124 """Estimate generation time. 

125 

126 Args: 

127 data_size: Number of data points. 

128 iterations: Generation iterations. 

129 mc_methods: Number of MC methods. 

130 

131 Returns: 

132 Human-readable time estimate. 

133 """ 

134 # Rough estimation based on typical performance 

135 base_time = 0.001 # seconds per iteration 

136 mc_factor = 1 + (mc_methods * 0.5) 

137 data_factor = data_size / 1000 

138 

139 total_seconds = base_time * iterations * mc_factor * data_factor 

140 

141 if total_seconds < 60: 

142 return f"~{int(total_seconds)} seconds" 

143 elif total_seconds < 3600: 

144 return f"~{int(total_seconds / 60)} minutes" 

145 else: 

146 return f"~{total_seconds / 3600:.1f} hours"