Coverage for src/ezplog/handlers/wizard/_progress.py: 95.27%

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# GLOBALS 

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

42 

43_DEFAULT_DESCRIPTION = "Working..." 

44_DESCRIPTION_COLUMN = "[progress.description]{task.description}" 

45_BLUE_DESCRIPTION_COLUMN = "[bold blue]{task.description}" 

46 

47# /////////////////////////////////////////////////////////////// 

48# CLASSES 

49# /////////////////////////////////////////////////////////////// 

50 

51 

52class ProgressMixin: 

53 """ 

54 Mixin providing progress bar methods for RichWizard. 

55 

56 This mixin adds all progress-related functionality including 

57 generic progress bars, spinners, download progress, installation progress, 

58 and layered progress bars. 

59 """ 

60 

61 # Type hints for attributes provided by RichWizard 

62 _console: Console 

63 _progress_prefix: str 

64 

65 # /////////////////////////////////////////////////////////////// 

66 # PROGRESS METHODS 

67 # /////////////////////////////////////////////////////////////// 

68 

69 @contextmanager 

70 def progress( 

71 self, 

72 description: str = _DEFAULT_DESCRIPTION, 

73 total: int | None = None, 

74 transient: bool = False, 

75 ) -> Generator[tuple[Progress, TaskID]]: 

76 """ 

77 Create a progress bar context manager. 

78 

79 Args: 

80 description: Progress description 

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

82 transient: Whether to clear progress on exit 

83 

84 Yields: 

85 tuple of (Progress, task_id) 

86 

87 Example: 

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

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

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

91 """ 

92 progress = Progress( 

93 TextColumn(self._progress_prefix), 

94 SpinnerColumn(), 

95 TextColumn(_DESCRIPTION_COLUMN), 

96 BarColumn(), 

97 TaskProgressColumn(), 

98 console=self._console, 

99 transient=transient, 

100 ) 

101 

102 with progress: 

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

104 yield progress, task 

105 

106 @contextmanager 

107 def spinner( 

108 self, description: str = _DEFAULT_DESCRIPTION 

109 ) -> Generator[tuple[Progress, TaskID]]: 

110 """ 

111 Create a simple spinner with description. 

112 

113 Args: 

114 description: Spinner description 

115 

116 Yields: 

117 tuple of (Progress, task_id) 

118 

119 Example: 

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

121 ... # Do work 

122 ... pass 

123 """ 

124 progress = Progress( 

125 TextColumn(self._progress_prefix), 

126 SpinnerColumn(), 

127 TextColumn(_DESCRIPTION_COLUMN), 

128 console=self._console, 

129 ) 

130 

131 with progress: 

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

133 yield progress, task 

134 

135 @contextmanager 

136 def spinner_with_status( 

137 self, description: str = _DEFAULT_DESCRIPTION 

138 ) -> Generator[tuple[Progress, TaskID]]: 

139 """ 

140 Create a spinner that can update status messages. 

141 

142 Args: 

143 description: Spinner description 

144 

145 Yields: 

146 tuple of (Progress, task_id) 

147 

148 Example: 

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

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

151 """ 

152 progress = Progress( 

153 TextColumn(self._progress_prefix), 

154 SpinnerColumn(), 

155 TextColumn(_DESCRIPTION_COLUMN), 

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

157 console=self._console, 

158 ) 

159 

160 with progress: 

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

162 yield progress, task 

163 

164 @contextmanager 

165 def download_progress( 

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

167 ) -> Generator[tuple[Progress, TaskID]]: 

168 """ 

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

170 

171 Args: 

172 description: Download description 

173 

174 Yields: 

175 tuple of (Progress, task_id) 

176 

177 Example: 

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

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

180 """ 

181 progress = Progress( 

182 TextColumn(self._progress_prefix), 

183 TextColumn(_DESCRIPTION_COLUMN), 

184 BarColumn(), 

185 TaskProgressColumn(), 

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

187 DownloadColumn(), 

188 TransferSpeedColumn(), 

189 console=self._console, 

190 ) 

191 

192 with progress: 

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

194 yield progress, task 

195 

196 @contextmanager 

197 def file_download_progress( 

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

199 ) -> Generator[tuple[Progress, TaskID]]: 

200 """ 

201 Create a progress bar for downloading a specific file. 

202 

203 Args: 

204 filename: Name of the file being downloaded 

205 total_size: Total size in bytes 

206 description: Main description 

207 

208 Yields: 

209 tuple of (Progress, task_id) 

210 

211 Example: 

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

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

214 """ 

215 progress = Progress( 

216 TextColumn(self._progress_prefix), 

217 TextColumn(_BLUE_DESCRIPTION_COLUMN), 

218 BarColumn(), 

219 TaskProgressColumn(), 

220 DownloadColumn(), 

221 TransferSpeedColumn(), 

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

223 console=self._console, 

224 ) 

225 

226 with progress: 

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

228 yield progress, task 

229 

230 def dependency_progress( 

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

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

233 """ 

234 Create a progress bar for dependency installation. 

235 

236 Args: 

237 dependencies: list of dependency names 

238 description: Main description 

239 

240 Yields: 

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

242 

243 Example: 

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

245 >>> for progress, task, dep in printer.wizard.dependency_progress(deps): 

246 ... # Install dependency 

247 ... progress.advance(task) 

248 """ 

249 progress = Progress( 

250 TextColumn(self._progress_prefix), 

251 SpinnerColumn(), 

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

253 TextColumn( 

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

255 ), 

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

257 BarColumn(), 

258 TaskProgressColumn(), 

259 TimeElapsedColumn(), 

260 console=self._console, 

261 ) 

262 

263 with progress: 

264 task = progress.add_task( 

265 description, 

266 total=len(dependencies), 

267 current=1, 

268 total_deps=len(dependencies), 

269 dependency="", 

270 ) 

271 

272 for i, dependency in enumerate(dependencies): 

273 progress.update( 

274 task, 

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

276 current=i + 1, 

277 dependency=dependency, 

278 ) 

279 yield progress, task, dependency 

280 progress.advance(task) 

281 time.sleep(0.1) 

282 

283 def package_install_progress( 

284 self, 

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

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

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

288 """ 

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

290 

291 Args: 

292 packages: list of tuples (package_name, version) 

293 description: Main description 

294 

295 Yields: 

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

297 

298 Example: 

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

300 >>> for progress, task, pkg, ver in printer.wizard.package_install_progress(packages): 

301 ... # Install package 

302 ... progress.advance(task) 

303 """ 

304 progress = Progress( 

305 TextColumn(self._progress_prefix), 

306 SpinnerColumn(), 

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

308 TextColumn( 

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

310 ), 

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

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

313 BarColumn(), 

314 TaskProgressColumn(), 

315 TimeElapsedColumn(), 

316 console=self._console, 

317 ) 

318 

319 with progress: 

320 task = progress.add_task( 

321 description, 

322 total=len(packages), 

323 current=1, 

324 total_packages=len(packages), 

325 package="", 

326 version="", 

327 ) 

328 

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

330 progress.update( 

331 task, 

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

333 current=i + 1, 

334 package=package_name, 

335 version=version, 

336 ) 

337 yield progress, task, package_name, version 

338 progress.advance(task) 

339 time.sleep(0.1) 

340 

341 @contextmanager 

342 def step_progress( 

343 self, 

344 steps: list[str], 

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

346 show_step_numbers: bool = True, 

347 show_time: bool = True, 

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

349 """ 

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

351 

352 Args: 

353 steps: list of step names 

354 description: Main description 

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

356 show_time: Show elapsed and remaining time 

357 

358 Yields: 

359 tuple of (Progress, task_id, steps_list) 

360 

361 Example: 

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

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

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

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

366 ... progress.advance(task) 

367 """ 

368 # Build columns based on options 

369 columns: list[ProgressColumn] = [ 

370 TextColumn(self._progress_prefix), 

371 SpinnerColumn(), 

372 TextColumn(_BLUE_DESCRIPTION_COLUMN), 

373 ] 

374 

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

376 columns.append( 

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

378 ) 

379 

380 columns.extend( 

381 [ 

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

383 BarColumn(), 

384 TaskProgressColumn(), 

385 ] 

386 ) 

387 

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

389 columns.extend( 

390 [ 

391 TimeElapsedColumn(), 

392 TimeRemainingColumn(), 

393 ] 

394 ) 

395 

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

397 

398 with progress: 

399 task = progress.add_task( 

400 description, 

401 total=len(steps), 

402 step=1, 

403 total_steps=len(steps), 

404 current_step="", 

405 ) 

406 

407 yield progress, task, steps 

408 

409 @contextmanager 

410 def file_copy_progress( 

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

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

413 """ 

414 Create a progress bar specifically for file copying operations. 

415 

416 Args: 

417 files: list of file paths to copy 

418 description: Main description 

419 

420 Yields: 

421 tuple of (Progress, task_id, files_list) 

422 

423 Example: 

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

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

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

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

428 ... progress.advance(task) 

429 """ 

430 progress = Progress( 

431 TextColumn(self._progress_prefix), 

432 SpinnerColumn(), 

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

434 BarColumn(), 

435 TaskProgressColumn(), 

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

437 console=self._console, 

438 ) 

439 

440 with progress: 

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

442 yield progress, task, files 

443 

444 def installation_progress( 

445 self, 

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

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

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

449 """ 

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

451 

452 Args: 

453 steps: list of tuples (step_name, step_description) 

454 description: Main description 

455 

456 Yields: 

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

458 

459 Example: 

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

461 >>> for progress, task, name, desc in printer.wizard.installation_progress(steps): 

462 ... # Process step 

463 ... progress.advance(task) 

464 """ 

465 progress = Progress( 

466 TextColumn(self._progress_prefix), 

467 SpinnerColumn(), 

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

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

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

471 BarColumn(), 

472 TaskProgressColumn(), 

473 TimeElapsedColumn(), 

474 console=self._console, 

475 ) 

476 

477 with progress: 

478 task = progress.add_task( 

479 description, 

480 total=len(steps), 

481 step=1, 

482 total_steps=len(steps), 

483 step_detail="", 

484 ) 

485 

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

487 progress.update( 

488 task, 

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

490 step=i + 1, 

491 step_detail=step_detail, 

492 ) 

493 yield progress, task, step_name, step_detail 

494 progress.advance(task) 

495 time.sleep(0.1) 

496 

497 def build_progress( 

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

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

500 """ 

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

502 

503 Args: 

504 phases: list of tuples (phase_name, weight_percentage) 

505 description: Main description 

506 

507 Yields: 

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

509 

510 Example: 

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

512 >>> for progress, task, phase, weight in printer.wizard.build_progress(phases): 

513 ... # Process phase 

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

515 """ 

516 progress = Progress( 

517 TextColumn(self._progress_prefix), 

518 SpinnerColumn(), 

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

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

521 BarColumn(), 

522 TaskProgressColumn(), 

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

524 TimeElapsedColumn(), 

525 console=self._console, 

526 ) 

527 

528 with progress: 

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

530 

531 current_progress = 0 

532 for phase_name, weight in phases: 

533 progress.update(task, current_phase=phase_name) 

534 yield progress, task, phase_name, weight 

535 current_progress += weight 

536 progress.update(task, completed=current_progress) 

537 

538 def deployment_progress( 

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

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

541 """ 

542 Create a progress bar for deployment processes. 

543 

544 Args: 

545 stages: list of deployment stage names 

546 description: Main description 

547 

548 Yields: 

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

550 

551 Example: 

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

553 >>> for progress, task, stage in printer.wizard.deployment_progress(stages): 

554 ... # Process stage 

555 ... progress.advance(task) 

556 """ 

557 progress = Progress( 

558 TextColumn(self._progress_prefix), 

559 SpinnerColumn(), 

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

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

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

563 BarColumn(), 

564 TaskProgressColumn(), 

565 TimeElapsedColumn(), 

566 TimeRemainingColumn(), 

567 console=self._console, 

568 ) 

569 

570 with progress: 

571 task = progress.add_task( 

572 description, 

573 total=len(stages), 

574 stage=1, 

575 total_stages=len(stages), 

576 current_stage="", 

577 ) 

578 

579 for i, stage in enumerate(stages): 

580 progress.update( 

581 task, 

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

583 stage=i + 1, 

584 current_stage=stage, 

585 ) 

586 yield progress, task, stage 

587 progress.advance(task) 

588 time.sleep(0.1) 

589 

590 @contextmanager 

591 def layered_progress( 

592 self, 

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

594 show_time: bool = True, 

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

596 """ 

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

598 

599 Args: 

600 layers: list of layer configurations, each containing: 

601 - 'name': Layer name/description 

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

603 - 'description': Display description for the layer 

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

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

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

607 show_time: Show elapsed and remaining time 

608 

609 Yields: 

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

611 

612 Example: 

613 >>> layers = [ 

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

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

616 ... ] 

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

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

619 """ 

620 # Build columns for the main progress bar 

621 columns: list[ProgressColumn] = [ 

622 TextColumn(self._progress_prefix), 

623 SpinnerColumn(), 

624 TextColumn(_BLUE_DESCRIPTION_COLUMN), 

625 BarColumn(), 

626 TaskProgressColumn(), 

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

628 ] 

629 

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

631 columns.extend( 

632 [ 

633 TimeElapsedColumn(), 

634 TimeRemainingColumn(), 

635 ] 

636 ) 

637 

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

639 

640 with progress: 

641 # Create tasks for each layer 

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

643 

644 for i, layer in enumerate(layers): 

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

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

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

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

649 

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

651 # Handle step-based layer 

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

653 layer_total = len(steps) 

654 task_id: TaskID = progress.add_task( 

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

656 total=layer_total, 

657 details="", 

658 steps=steps, 

659 ) 

660 else: 

661 # Handle regular progress layer 

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

663 task_id = progress.add_task( 

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

665 total=layer_total, 

666 details="", 

667 ) 

668 

669 task_ids[layer_name] = task_id 

670 

671 yield progress, task_ids