Coverage for src / osiris_cli / progress_display.py: 0%

1#!/usr/bin/env python3 

2""" 

3Progress Display - Beautiful progress bars for long operations 

4""" 

5 

6import time 

7from rich.console import Console 

8from rich.progress import ( 

9 Progress, 

10 SpinnerColumn, 

11 TextColumn, 

12 BarColumn, 

13 TaskProgressColumn, 

14 TimeRemainingColumn, 

15 TimeElapsedColumn 

16) 

17from rich.live import Live 

18from rich.text import Text 

19 

20console = Console() 

21 

22 

23class ProgressTracker: 

24 """ 

25 Beautiful progress tracking for long operations 

26 """ 

27 

28 def __init__(self): 

29 self.progress = Progress( 

30 SpinnerColumn(spinner_name="dots"), 

31 TextColumn("[bold cyan]{task.description}"), 

32 BarColumn(complete_style="green", finished_style="bold green"), 

33 TaskProgressColumn(), 

34 TimeElapsedColumn(), 

35 TimeRemainingColumn(), 

36 console=console, 

37 transient=True 

38 ) 

39 self.tasks = {} 

40 

41 def add_task(self, description: str, total: int = 100) -> int: 

42 """Add a new task to track""" 

43 task_id = self.progress.add_task(description, total=total) 

44 self.tasks[description] = task_id 

45 return task_id 

46 

47 def update(self, task_id: int, advance: int = 1, description: str = None): 

48 """Update task progress""" 

49 kwargs = {"advance": advance} 

50 if description: 

51 kwargs["description"] = description 

52 self.progress.update(task_id, **kwargs) 

53 

54 def complete(self, task_id: int): 

55 """Mark task as complete""" 

56 self.progress.update(task_id, completed=True) 

57 

58 def __enter__(self): 

59 """Context manager entry""" 

60 self.progress.start() 

61 return self 

62 

63 def __exit__(self, exc_type, exc_val, exc_tb): 

64 """Context manager exit""" 

65 self.progress.stop() 

66 

67 

68class SimpleSpinner: 

69 """ 

70 Simple spinner for indeterminate operations 

71 """ 

72 

73 def __init__(self, message: str = "Processing..."): 

74 self.message = message 

75 self.progress = Progress( 

76 SpinnerColumn(spinner_name="dots"), 

77 TextColumn("[bold cyan]{task.description}"), 

78 console=console, 

79 transient=True 

80 ) 

81 self.task_id = None 

82 

83 def start(self): 

84 """Start the spinner""" 

85 self.progress.start() 

86 self.task_id = self.progress.add_task(self.message) 

87 

88 def update(self, message: str): 

89 """Update spinner message""" 

90 if self.task_id is not None: 

91 self.progress.update(self.task_id, description=message) 

92 

93 def stop(self, final_message: str = None): 

94 """Stop the spinner""" 

95 if final_message: 

96 console.print(f"[green]✓[/green] {final_message}") 

97 self.progress.stop() 

98 

99 def __enter__(self): 

100 """Context manager entry""" 

101 self.start() 

102 return self 

103 

104 def __exit__(self, exc_type, exc_val, exc_tb): 

105 """Context manager exit""" 

106 self.stop() 

107 

108 

109def show_progress(description: str, total: int, items: list): 

110 """ 

111 Show progress for a list of items 

112  

113 Example: 

114 show_progress("Processing files", 10, files) 

115 """ 

116 with Progress( 

117 SpinnerColumn(), 

118 TextColumn("[bold cyan]{task.description}"), 

119 BarColumn(), 

120 TaskProgressColumn(), 

121 TimeElapsedColumn(), 

122 console=console 

123 ) as progress: 

124 task = progress.add_task(description, total=total) 

125 

126 for item in items: 

127 # Process item 

128 yield item 

129 progress.advance(task) 

130 

131 

132def with_spinner(message: str): 

133 """ 

134 Decorator to show a spinner during function execution 

135  

136 Example: 

137 @with_spinner("Loading data...") 

138 def load_data(): 

139 # ... long operation 

140 pass 

141 """ 

142 def decorator(func): 

143 def wrapper(*args, **kwargs): 

144 with SimpleSpinner(message): 

145 return func(*args, **kwargs) 

146 return wrapper 

147 return decorator