Coverage for src / ezpl / handlers / wizard / progress.py: 95.33%

1# /////////////////////////////////////////////////////////////// 

2# EZPL - Wizard Progress Mixin 

3# Project: ezpl 

4# /////////////////////////////////////////////////////////////// 

5 

6""" 

7Progress methods mixin for Rich Wizard. 

8 

9This module provides all progress bar-related methods for the RichWizard class. 

10""" 

11 

12from __future__ import annotations 

13 

14# /////////////////////////////////////////////////////////////// 

15# IMPORTS 

16# /////////////////////////////////////////////////////////////// 

17# Standard library imports 

18import time 

19from collections.abc import Generator 

20from contextlib import contextmanager 

21from typing import Any 

22 

23# Third-party imports 

24from rich.console import Console 

25from rich.progress import ( 

26 BarColumn, 

27 DownloadColumn, 

28 Progress, 

29 ProgressColumn, 

30 SpinnerColumn, 

31 TaskID, 

32 TaskProgressColumn, 

33 TextColumn, 

34 TimeElapsedColumn, 

35 TimeRemainingColumn, 

36 TransferSpeedColumn, 

37) 

38 

39# /////////////////////////////////////////////////////////////// 

40# CLASSES 

41# /////////////////////////////////////////////////////////////// 

42 

43 

44class ProgressMixin: 

45 """ 

46 Mixin providing progress bar methods for RichWizard. 

47 

48 This mixin adds all progress-related functionality including 

49 generic progress bars, spinners, download progress, installation progress, 

50 and layered progress bars. 

51 """ 

52 

53 # Type hints for attributes provided by RichWizard 

54 _console: Console 

55 _progress_prefix: str 

56 

57 # /////////////////////////////////////////////////////////////// 

58 # PROGRESS METHODS 

59 # /////////////////////////////////////////////////////////////// 

60 

61 @contextmanager 

62 def progress( 

63 self, 

64 description: str = "Working...", 

65 total: int | None = None, 

66 transient: bool = False, 

67 ) -> Generator[tuple[Progress, TaskID], None, None]: 

68 """ 

69 Create a progress bar context manager. 

70 

71 Args: 

72 description: Progress description 

73 total: Total number of items (None for indeterminate) 

74 transient: Whether to clear progress on exit 

75 

76 Yields: 

77 tuple of (Progress, task_id) 

78 

79 Example: 

80 >>> with printer.wizard.progress("Processing...", total=100) as (progress, task): 

81 ... for i in range(100): 

82 ... progress.update(task, advance=1) 

83 """ 

84 progress = Progress( 

85 TextColumn(self._progress_prefix), 

86 SpinnerColumn(), 

87 TextColumn("[progress.description]{task.description}"), 

88 BarColumn(), 

89 TaskProgressColumn(), 

90 console=self._console, 

91 transient=transient, 

92 ) 

93 

94 with progress: 

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

96 yield progress, task 

97 

98 @contextmanager 

99 def spinner( 

100 self, description: str = "Working..." 

101 ) -> Generator[tuple[Progress, TaskID], None, None]: 

102 """ 

103 Create a simple spinner with description. 

104 

105 Args: 

106 description: Spinner description 

107 

108 Yields: 

109 tuple of (Progress, task_id) 

110 

111 Example: 

112 >>> with printer.wizard.spinner("Loading...") as (progress, task): 

113 ... # Do work 

114 ... pass 

115 """ 

116 progress = Progress( 

117 TextColumn(self._progress_prefix), 

118 SpinnerColumn(), 

119 TextColumn("[progress.description]{task.description}"), 

120 console=self._console, 

121 ) 

122 

123 with progress: 

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

125 yield progress, task 

126 

127 @contextmanager 

128 def spinner_with_status( 

129 self, description: str = "Working..." 

130 ) -> Generator[tuple[Progress, TaskID], None, None]: 

131 """ 

132 Create a spinner that can update status messages. 

133 

134 Args: 

135 description: Spinner description 

136 

137 Yields: 

138 tuple of (Progress, task_id) 

139 

140 Example: 

141 >>> with printer.wizard.spinner_with_status("Processing...") as (progress, task): 

142 ... progress.update(task, status="Step 1/3") 

143 """ 

144 progress = Progress( 

145 TextColumn(self._progress_prefix), 

146 SpinnerColumn(), 

147 TextColumn("[progress.description]{task.description}"), 

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

149 console=self._console, 

150 ) 

151 

152 with progress: 

153 task = progress.add_task(description, total=None, status="") 

154 yield progress, task 

155 

156 @contextmanager 

157 def download_progress( 

158 self, description: str = "Downloading..." 

159 ) -> Generator[tuple[Progress, TaskID], None, None]: 

160 """ 

161 Create a download progress bar with speed and size information. 

162 

163 Args: 

164 description: Download description 

165 

166 Yields: 

167 tuple of (Progress, task_id) 

168 

169 Example: 

170 >>> with printer.wizard.download_progress() as (progress, task): 

171 ... progress.update(task, completed=50, total=100) 

172 """ 

173 progress = Progress( 

174 TextColumn(self._progress_prefix), 

175 TextColumn("[progress.description]{task.description}"), 

176 BarColumn(), 

177 TaskProgressColumn(), 

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

179 DownloadColumn(), 

180 TransferSpeedColumn(), 

181 console=self._console, 

182 ) 

183 

184 with progress: 

185 task = progress.add_task(description, total=100) 

186 yield progress, task 

187 

188 @contextmanager 

189 def file_download_progress( 

190 self, filename: str, total_size: int, description: str = "Downloading file..." 

191 ) -> Generator[tuple[Progress, TaskID], None, None]: 

192 """ 

193 Create a progress bar for downloading a specific file. 

194 

195 Args: 

196 filename: Name of the file being downloaded 

197 total_size: Total size in bytes 

198 description: Main description 

199 

200 Yields: 

201 tuple of (Progress, task_id) 

202 

203 Example: 

204 >>> with printer.wizard.file_download_progress("file.zip", 1024000) as (progress, task): 

205 ... progress.update(task, advance=512000) 

206 """ 

207 progress = Progress( 

208 TextColumn(self._progress_prefix), 

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

210 BarColumn(), 

211 TaskProgressColumn(), 

212 DownloadColumn(), 

213 TransferSpeedColumn(), 

214 TextColumn("[dim]{task.fields[filename]}"), 

215 console=self._console, 

216 ) 

217 

218 with progress: 

219 task = progress.add_task(description, total=total_size, filename=filename) 

220 yield progress, task 

221 

222 @contextmanager 

223 def dependency_progress( 

224 self, dependencies: list[str], description: str = "Installing dependencies..." 

225 ) -> Generator[tuple[Progress, TaskID, str], None, None]: 

226 """ 

227 Create a progress bar for dependency installation. 

228 

229 Args: 

230 dependencies: list of dependency names 

231 description: Main description 

232 

233 Yields: 

234 tuple of (Progress, task_id, dependency_name) for each dependency 

235 

236 Example: 

237 >>> deps = ["package1", "package2", "package3"] 

238 >>> with printer.wizard.dependency_progress(deps) as (progress, task, dep): 

239 ... # Install dependency 

240 ... progress.advance(task) 

241 """ 

242 progress = Progress( 

243 TextColumn(self._progress_prefix), 

244 SpinnerColumn(), 

245 TextColumn("[bold green]{task.description}"), 

246 TextColumn( 

247 "[dim]Dependency {task.fields[current]}/{task.fields[total_deps]}" 

248 ), 

249 TextColumn("[dim]{task.fields[dependency]}"), 

250 BarColumn(), 

251 TaskProgressColumn(), 

252 TimeElapsedColumn(), 

253 console=self._console, 

254 ) 

255 

256 with progress: 

257 task = progress.add_task( 

258 description, 

259 total=len(dependencies), 

260 current=1, 

261 total_deps=len(dependencies), 

262 dependency="", 

263 ) 

264 

265 for i, dependency in enumerate(dependencies): 

266 progress.update( 

267 task, 

268 description=f"[bold green]Installing {dependency}", 

269 current=i + 1, 

270 dependency=dependency, 

271 ) 

272 yield progress, task, dependency 

273 progress.advance(task) 

274 time.sleep(0.1) 

275 

276 @contextmanager 

277 def package_install_progress( 

278 self, 

279 packages: list[tuple[str, str]], 

280 description: str = "Installing packages...", 

281 ) -> Generator[tuple[Progress, TaskID, str, str], None, None]: 

282 """ 

283 Create a progress bar for package installation with version info. 

284 

285 Args: 

286 packages: list of tuples (package_name, version) 

287 description: Main description 

288 

289 Yields: 

290 tuple of (Progress, task_id, package_name, version) for each package 

291 

292 Example: 

293 >>> packages = [("requests", "2.31.0"), ("click", "8.1.0")] 

294 >>> with printer.wizard.package_install_progress(packages) as (progress, task, pkg, ver): 

295 ... # Install package 

296 ... progress.advance(task) 

297 """ 

298 progress = Progress( 

299 TextColumn(self._progress_prefix), 

300 SpinnerColumn(), 

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

302 TextColumn( 

303 "[dim]Package {task.fields[current]}/{task.fields[total_packages]}" 

304 ), 

305 TextColumn("[dim]{task.fields[package]}"), 

306 TextColumn("[dim]{task.fields[version]}"), 

307 BarColumn(), 

308 TaskProgressColumn(), 

309 TimeElapsedColumn(), 

310 console=self._console, 

311 ) 

312 

313 with progress: 

314 task = progress.add_task( 

315 description, 

316 total=len(packages), 

317 current=1, 

318 total_packages=len(packages), 

319 package="", 

320 version="", 

321 ) 

322 

323 for i, (package_name, version) in enumerate(packages): 

324 progress.update( 

325 task, 

326 description=f"[bold cyan]Installing {package_name}", 

327 current=i + 1, 

328 package=package_name, 

329 version=version, 

330 ) 

331 yield progress, task, package_name, version 

332 progress.advance(task) 

333 time.sleep(0.1) 

334 

335 @contextmanager 

336 def step_progress( 

337 self, 

338 steps: list[str], 

339 description: str = "Processing...", 

340 show_step_numbers: bool = True, 

341 show_time: bool = True, 

342 ) -> Generator[tuple[Progress, TaskID, list[str]], None, None]: 

343 """ 

344 Create a step-based progress bar with detailed step information. 

345 

346 Args: 

347 steps: list of step names 

348 description: Main description 

349 show_step_numbers: Show step numbers (e.g., "Step 1/5") 

350 show_time: Show elapsed and remaining time 

351 

352 Yields: 

353 tuple of (Progress, task_id, steps_list) 

354 

355 Example: 

356 >>> steps = ["Step 1", "Step 2", "Step 3"] 

357 >>> with printer.wizard.step_progress(steps) as (progress, task, steps_list): 

358 ... for i, step in enumerate(steps_list): 

359 ... progress.update(task, completed=i, current_step=step) 

360 ... progress.advance(task) 

361 """ 

362 # Build columns based on options 

363 columns: list[ProgressColumn] = [ 

364 TextColumn(self._progress_prefix), 

365 SpinnerColumn(), 

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

367 ] 

368 

369 if show_step_numbers: 369 ↛ 374line 369 didn't jump to line 374 because the condition on line 369 was always true

370 columns.append( 

371 TextColumn("[dim]Step {task.fields[step]}/{task.fields[total_steps]}") 

372 ) 

373 

374 columns.extend( 

375 [ 

376 TextColumn("[dim]{task.fields[current_step]}"), 

377 BarColumn(), 

378 TaskProgressColumn(), 

379 ] 

380 ) 

381 

382 if show_time: 382 ↛ 390line 382 didn't jump to line 390 because the condition on line 382 was always true

383 columns.extend( 

384 [ 

385 TimeElapsedColumn(), 

386 TimeRemainingColumn(), 

387 ] 

388 ) 

389 

390 progress = Progress(*columns, console=self._console) 

391 

392 with progress: 

393 task = progress.add_task( 

394 description, 

395 total=len(steps), 

396 step=1, 

397 total_steps=len(steps), 

398 current_step="", 

399 ) 

400 

401 yield progress, task, steps 

402 

403 @contextmanager 

404 def file_copy_progress( 

405 self, files: list[str], description: str = "Copying files..." 

406 ) -> Generator[tuple[Progress, TaskID, list[str]], None, None]: 

407 """ 

408 Create a progress bar specifically for file copying operations. 

409 

410 Args: 

411 files: list of file paths to copy 

412 description: Main description 

413 

414 Yields: 

415 tuple of (Progress, task_id, files_list) 

416 

417 Example: 

418 >>> files = ["file1.txt", "file2.txt"] 

419 >>> with printer.wizard.file_copy_progress(files) as (progress, task, files_list): 

420 ... for i, file in enumerate(files_list): 

421 ... progress.update(task, current_file=file) 

422 ... progress.advance(task) 

423 """ 

424 progress = Progress( 

425 TextColumn(self._progress_prefix), 

426 SpinnerColumn(), 

427 TextColumn("[bold magenta]{task.description}"), 

428 BarColumn(), 

429 TaskProgressColumn(), 

430 TextColumn("[dim]{task.fields[current_file]}"), 

431 console=self._console, 

432 ) 

433 

434 with progress: 

435 task = progress.add_task(description, total=len(files), current_file="") 

436 yield progress, task, files 

437 

438 @contextmanager 

439 def installation_progress( 

440 self, 

441 steps: list[tuple[str, str]], 

442 description: str = "Installation in progress...", 

443 ) -> Generator[tuple[Progress, TaskID, str, str], None, None]: 

444 """ 

445 Create a progress bar for installation processes with step details. 

446 

447 Args: 

448 steps: list of tuples (step_name, step_description) 

449 description: Main description 

450 

451 Yields: 

452 tuple of (Progress, task_id, step_name, step_description) for each step 

453 

454 Example: 

455 >>> steps = [("Init", "Initializing..."), ("Install", "Installing...")] 

456 >>> with printer.wizard.installation_progress(steps) as (progress, task, name, desc): 

457 ... # Process step 

458 ... progress.advance(task) 

459 """ 

460 progress = Progress( 

461 TextColumn(self._progress_prefix), 

462 SpinnerColumn(), 

463 TextColumn("[bold green]{task.description}"), 

464 TextColumn("[dim]Step {task.fields[step]}/{task.fields[total_steps]}"), 

465 TextColumn("[dim]{task.fields[step_detail]}"), 

466 BarColumn(), 

467 TaskProgressColumn(), 

468 TimeElapsedColumn(), 

469 console=self._console, 

470 ) 

471 

472 with progress: 

473 task = progress.add_task( 

474 description, 

475 total=len(steps), 

476 step=1, 

477 total_steps=len(steps), 

478 step_detail="", 

479 ) 

480 

481 for i, (step_name, step_detail) in enumerate(steps): 

482 progress.update( 

483 task, 

484 description=f"[bold green]{step_name}", 

485 step=i + 1, 

486 step_detail=step_detail, 

487 ) 

488 yield progress, task, step_name, step_detail 

489 progress.advance(task) 

490 time.sleep(0.1) 

491 

492 @contextmanager 

493 def build_progress( 

494 self, phases: list[tuple[str, int]], description: str = "Building project..." 

495 ) -> Generator[tuple[Progress, TaskID, str, int], None, None]: 

496 """ 

497 Create a progress bar for build processes with weighted phases. 

498 

499 Args: 

500 phases: list of tuples (phase_name, weight_percentage) 

501 description: Main description 

502 

503 Yields: 

504 tuple of (Progress, task_id, phase_name, weight) for each phase 

505 

506 Example: 

507 >>> phases = [("Compile", 40), ("Test", 30), ("Package", 30)] 

508 >>> with printer.wizard.build_progress(phases) as (progress, task, phase, weight): 

509 ... # Process phase 

510 ... progress.update(task, advance=weight) 

511 """ 

512 progress = Progress( 

513 TextColumn(self._progress_prefix), 

514 SpinnerColumn(), 

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

516 TextColumn("[dim]{task.fields[current_phase]}"), 

517 BarColumn(), 

518 TaskProgressColumn(), 

519 TextColumn("[bold yellow]{task.percentage:>3.0f}%"), 

520 TimeElapsedColumn(), 

521 console=self._console, 

522 ) 

523 

524 with progress: 

525 task = progress.add_task(description, total=100, current_phase="") 

526 

527 current_progress = 0 

528 for phase_name, weight in phases: 

529 progress.update(task, current_phase=phase_name) 

530 yield progress, task, phase_name, weight 

531 current_progress += weight 

532 progress.update(task, completed=current_progress) 

533 

534 @contextmanager 

535 def deployment_progress( 

536 self, stages: list[str], description: str = "Deploying..." 

537 ) -> Generator[tuple[Progress, TaskID, str], None, None]: 

538 """ 

539 Create a progress bar for deployment processes. 

540 

541 Args: 

542 stages: list of deployment stage names 

543 description: Main description 

544 

545 Yields: 

546 tuple of (Progress, task_id, stage_name) for each stage 

547 

548 Example: 

549 >>> stages = ["Build", "Test", "Deploy"] 

550 >>> with printer.wizard.deployment_progress(stages) as (progress, task, stage): 

551 ... # Process stage 

552 ... progress.advance(task) 

553 """ 

554 progress = Progress( 

555 TextColumn(self._progress_prefix), 

556 SpinnerColumn(), 

557 TextColumn("[bold magenta]{task.description}"), 

558 TextColumn("[dim]Stage {task.fields[stage]}/{task.fields[total_stages]}"), 

559 TextColumn("[dim]{task.fields[current_stage]}"), 

560 BarColumn(), 

561 TaskProgressColumn(), 

562 TimeElapsedColumn(), 

563 TimeRemainingColumn(), 

564 console=self._console, 

565 ) 

566 

567 with progress: 

568 task = progress.add_task( 

569 description, 

570 total=len(stages), 

571 stage=1, 

572 total_stages=len(stages), 

573 current_stage="", 

574 ) 

575 

576 for i, stage in enumerate(stages): 

577 progress.update( 

578 task, 

579 description=f"[bold magenta]{stage}", 

580 stage=i + 1, 

581 current_stage=stage, 

582 ) 

583 yield progress, task, stage 

584 progress.advance(task) 

585 time.sleep(0.1) 

586 

587 @contextmanager 

588 def layered_progress( 

589 self, 

590 layers: list[dict[str, Any]], 

591 show_time: bool = True, 

592 ) -> Generator[tuple[Progress, dict[str, TaskID]], None, None]: 

593 """ 

594 Create a multi-level progress bar with dynamic layers. 

595 

596 Args: 

597 layers: list of layer configurations, each containing: 

598 - 'name': Layer name/description 

599 - 'total': Total items for this layer (optional, None for indeterminate) 

600 - 'description': Display description for the layer 

601 - 'style': Rich style for this layer (optional) 

602 - 'type': Layer type - 'progress' (default) or 'steps' 

603 - 'steps': list of step names (required if type='steps') 

604 show_time: Show elapsed and remaining time 

605 

606 Yields: 

607 tuple of (Progress, task_ids_dict) where task_ids_dict maps layer names to task IDs 

608 

609 Example: 

610 >>> layers = [ 

611 ... {"name": "layer1", "description": "Layer 1", "total": 10}, 

612 ... {"name": "layer2", "description": "Layer 2", "total": 5} 

613 ... ] 

614 >>> with printer.wizard.layered_progress(layers) as (progress, task_ids): 

615 ... progress.update(task_ids["layer1"], advance=1) 

616 """ 

617 # Build columns for the main progress bar 

618 columns: list[ProgressColumn] = [ 

619 TextColumn(self._progress_prefix), 

620 SpinnerColumn(), 

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

622 BarColumn(), 

623 TaskProgressColumn(), 

624 TextColumn("[dim]{task.fields[details]}"), 

625 ] 

626 

627 if show_time: 627 ↛ 635line 627 didn't jump to line 635 because the condition on line 627 was always true

628 columns.extend( 

629 [ 

630 TimeElapsedColumn(), 

631 TimeRemainingColumn(), 

632 ] 

633 ) 

634 

635 progress = Progress(*columns, console=self._console) 

636 

637 with progress: 

638 # Create tasks for each layer 

639 task_ids: dict[str, TaskID] = {} 

640 

641 for i, layer in enumerate(layers): 

642 layer_name = layer.get("name", f"Layer_{i}") 

643 layer_type = layer.get("type", "progress") 

644 layer_desc = layer.get("description", layer_name) 

645 layer_style = layer.get("style", "default") 

646 

647 if layer_type == "steps": 647 ↛ 649line 647 didn't jump to line 649 because the condition on line 647 was never true

648 # Handle step-based layer 

649 steps = layer.get("steps", []) 

650 layer_total = len(steps) 

651 task_id: TaskID = progress.add_task( 

652 f"[{layer_style}]{layer_desc}", 

653 total=layer_total, 

654 details="", 

655 steps=steps, 

656 ) 

657 else: 

658 # Handle regular progress layer 

659 layer_total = layer.get("total", None) 

660 task_id = progress.add_task( 

661 f"[{layer_style}]{layer_desc}", 

662 total=layer_total, 

663 details="", 

664 ) 

665 

666 task_ids[layer_name] = task_id 

667 

668 yield progress, task_ids