Coverage for src/ezpl/handlers/wizard/dynamic.py: 82.50%

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

2# EZPL - Wizard Dynamic Progress Mixin

3# Project: ezpl

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

6"""

7Dynamic progress methods mixin for Rich Wizard.

9This module provides dynamic layered progress bar functionality with

10layers that can appear, progress, and disappear automatically.

11"""

13from __future__ import annotations

15# ///////////////////////////////////////////////////////////////

16# IMPORTS

17# ///////////////////////////////////////////////////////////////

18# Standard library imports

19import threading

20import time

21from collections.abc import Generator

22from contextlib import contextmanager

23from typing import Any, TypedDict

25# Third-party imports

26from rich.console import Console

27from rich.progress import (

28 BarColumn,

29 DownloadColumn,

30 Progress,

31 ProgressColumn,

32 SpinnerColumn,

33 TaskID,

34 TaskProgressColumn,

35 TextColumn,

36 TimeElapsedColumn,

37 TimeRemainingColumn,

38 TransferSpeedColumn,

39)

40from rich.text import Text

42# ///////////////////////////////////////////////////////////////

43# TYPES

44# ///////////////////////////////////////////////////////////////

47class _StageConfigRequired(TypedDict):

48 """Required fields for stage configuration."""

50 name: str

51 type: str # "progress", "steps", "spinner", "download", "main"

54class StageConfig(_StageConfigRequired, total=False):

55 """Configuration for a DynamicLayeredProgress stage.

57 Required fields:

58 name: Layer name/identifier.

59 type: Layer type ("progress", "steps", "spinner", "download", "main").

61 Optional fields:

62 description: Display description for the layer.

63 style: Rich style string.

64 total: Total items (for "progress" or "download" type).

65 steps: List of step names (for "steps" or "main" type).

66 total_size: Total size in bytes (for "download" type).

67 filename: Filename (for "download" type).

68 """

70 description: str

71 style: str

72 total: int

73 steps: list[str]

74 total_size: int

75 filename: str

78# ///////////////////////////////////////////////////////////////

79# CLASSES

80# ///////////////////////////////////////////////////////////////

83class ConditionalStatusColumn(TextColumn):

84 """A text column that only shows status if the field exists."""

86 def __init__(self):

87 super().__init__("") # Empty text format, we override render

89 def render(self, task):

90 """

91 Render the status field if it exists in the task.

93 Args:

94 task: The task object to render

96 Returns:

97 Text object containing the status or empty text

98 """

99 if hasattr(task, "fields") and "status" in task.fields: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 return Text(str(task.fields["status"]), style="dim")

101 return Text("")

102

103

104class ConditionalDetailsColumn(TextColumn):

105 """A text column that only shows details if the field exists."""

106

107 def __init__(self):

108 super().__init__("") # Empty text format, we override render

109

110 def render(self, task):

111 """

112 Render the details field if it exists in the task.

113

114 Args:

115 task: The task object to render

116

117 Returns:

118 Text object containing the details or empty text

119 """

120 if hasattr(task, "fields") and "details" in task.fields: 120 ↛ 122line 120 didn't jump to line 122 because the condition on line 120 was always true

121 return Text(str(task.fields["details"]), style="dim")

122 return Text("")

123

124

125class DynamicLayeredProgress:

126 """

127 Manages a dynamic layered progress bar with disappearing layers.

128

129 This class provides a progress bar system where layers can appear,

130 progress, and disappear based on the current state of operations.

131 """

132

133 def __init__(

134 self,

135 console: Console,

136 progress_prefix: str,

137 stages: list[StageConfig],

138 show_time: bool = True,

139 ) -> None:

140 """

141 Initialize the dynamic layered progress bar.

142

143 Args:

144 console: Rich Console instance

145 progress_prefix: Prefix string for progress bars

146 stages: List of stage configurations

147 show_time: Whether to show elapsed and remaining time

148 """

149 self._console = console

150 self._progress_prefix = progress_prefix

151 self.stages = stages

152 self.show_time = show_time

153 self.progress: Progress | None = None

154 self.task_ids: dict[str, TaskID] = {}

155 self.active_layers: list[TaskID] = []

156 self.completed_layers: list[TaskID] = []

157 self.layer_metadata: dict[

158 TaskID, dict[str, Any]

159 ] = {} # Store additional layer info

160 self._emergency_stopped = False

161 self._emergency_message: str | None = None

162 self._lock = threading.Lock()

163

164 # Hierarchy attributes (initialized in _setup_hierarchy)

165 self.has_main_layer: bool = False

166 self.main_layer_name: str | None = None

167 self.sub_layers: list[StageConfig] = []

168

169 # Detect main layer and setup hierarchy

170 self._setup_hierarchy()

171

172 def _setup_hierarchy(self) -> None:

173 """Setup layer hierarchy and detect main layer."""

174 self.has_main_layer = False

175 self.main_layer_name = None

176 self.sub_layers = []

177

178 # Detect main layer

179 for stage in self.stages:

180 if stage.get("type") == "main":

181 self.has_main_layer = True

182 self.main_layer_name = stage.get("name", "main")

183 break

184

185 # If main layer found, setup sub-layers

186 if self.has_main_layer:

187 self.sub_layers = [

188 stage for stage in self.stages if stage.get("type") != "main"

189 ]

190 # Auto-configure main layer steps if not provided

191 for stage in self.stages:

192 if stage.get("type") == "main" and "steps" not in stage:

193 stage["steps"] = [

194 s.get("name", f"Step {i + 1}")

195 for i, s in enumerate(self.sub_layers)

196 ]

197

198 def _create_progress_bar(self) -> Progress:

199 """Create the Rich Progress instance with proper columns."""

200 # Build columns for the main progress bar

201 columns: list[ProgressColumn] = [

202 TextColumn(self._progress_prefix), # Ezpl prefix

203 SpinnerColumn(),

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

205 BarColumn(),

206 TaskProgressColumn(),

207 ConditionalDetailsColumn(), # Conditional details column

208 ConditionalStatusColumn(), # Conditional status column

209 ]

210

211 # Check if we have any download layers to add download columns

212 has_download = any(stage.get("type") == "download" for stage in self.stages)

213 if has_download:

214 columns.extend(

215 [

216 DownloadColumn(), # Download info column

217 TransferSpeedColumn(), # Transfer speed column

218 ]

219 )

220

221 if self.show_time:

222 columns.extend(

223 [

224 TimeElapsedColumn(),

225 TimeRemainingColumn(),

226 ]

227 )

228

229 return Progress(*columns, console=self._console)

230

231 def _create_layer(self, layer_config: StageConfig) -> int:

232 """Create a new layer in the progress bar.

233

234 Args:

235 layer_config: Layer configuration dictionary

236

237 Returns:

238 Task ID of the created layer

239 """

240 layer_name = layer_config.get("name", f"Layer_{len(self.task_ids)}")

241 layer_type = layer_config.get("type", "progress")

242 layer_desc = layer_config.get("description", layer_name)

243 layer_style = layer_config.get("style", "default")

244

245 # Determine layer prefix and styling based on hierarchy

246 if self.has_main_layer and layer_type == "main":

247 # Main layer: bold and prominent

248 layer_style = "bold " + layer_style if layer_style != "default" else "bold"

249 elif self.has_main_layer and layer_type != "main":

250 # Sub-layer: add indentation and use softer colors

251 layer_desc = f" ├─ {layer_desc}"

252

253 if not self.progress: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true

254 return -1

255

256 if layer_type == "steps":

257 # Handle step-based layer

258 steps = layer_config.get("steps", [])

259 steps_total = len(steps)

260 task_id = self.progress.add_task(

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

262 total=steps_total,

263 steps=steps, # Store steps for later use

264 details="", # Initialize details field

265 )

266 elif layer_type == "spinner":

267 # Handle spinner layer (indeterminate progress)

268 task_id = self.progress.add_task(

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

270 total=None, # Indeterminate

271 details="", # Initialize details field

272 )

273 elif layer_type == "download":

274 # Handle download layer with speed and size info

275 total_size = layer_config.get("total_size", 100)

276 filename = layer_config.get("filename", "")

277 task_id = self.progress.add_task(

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

279 total=total_size,

280 details="", # Initialize details field

281 filename=filename, # Store filename for download info

282 )

283 elif layer_type == "main":

284 # Handle main layer (special case)

285 steps = layer_config.get("steps", [])

286 main_total = len(steps)

287 task_id = self.progress.add_task(

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

289 total=main_total,

290 steps=steps, # Store steps for later use

291 details="", # Initialize details field

292 )

293 else:

294 # Handle regular progress layer

295 progress_total: int | None = layer_config.get("total")

296 task_id = self.progress.add_task(

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

298 total=progress_total,

299 details="", # Initialize details field

300 )

301

302 # Store layer metadata

303 self.layer_metadata[task_id] = {

304 "name": layer_name,

305 "type": layer_type,

306 "config": layer_config,

307 "is_main": layer_type == "main",

308 "is_sub": self.has_main_layer and layer_type != "main",

309 }

310

311 self.active_layers.append(task_id)

312 self.task_ids[layer_name] = task_id

313 return task_id

314

315 def update_layer(self, layer_name: str, progress: int, details: str = "") -> None:

316 """Update a specific layer's progress.

317

318 Args:

319 layer_name: Name of the layer to update

320 progress: Progress value (0-100 or step index)

321 details: Additional details to display

322 """

323 with self._lock:

324 if not self.progress: 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true

325 return

326

327 task_id = self.task_ids.get(layer_name)

328 if task_id is None: 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true

329 return

330

331 metadata = self.layer_metadata.get(task_id)

332 if metadata is None: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true

333 return

334

335 self._update_layer_unsafe(task_id, metadata, progress, details)

336

337 def _update_layer_unsafe(

338 self,

339 task_id: TaskID,

340 metadata: dict[str, Any],

341 progress: int,

342 details: str,

343 ) -> None:

344 """Update layer without acquiring lock (caller must hold lock).

345

346 Args:

347 task_id: Task ID to update

348 metadata: Layer metadata

349 progress: Progress value

350 details: Additional details

351 """

352 if not self.progress: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true

353 return

354

355 # Update the layer based on its type

356 if metadata["type"] == "steps":

357 # Handle step-based layer

358 task = self._get_task(task_id)

359 if task is None: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true

360 return

361 steps = getattr(task, "fields", {}).get(

362 "steps", metadata["config"].get("steps", [])

363 )

364

365 if steps and progress < len(steps): 365 ↛ 376line 365 didn't jump to line 376 because the condition on line 365 was always true

366 current_step = steps[progress]

367 step_progress = f"Step {progress + 1}/{len(steps)}: {current_step}"

368

369 self.progress.update(

370 task_id,

371 completed=progress,

372 description=f"{task.description} - {step_progress}",

373 details=details,

374 )

375 else:

376 self.progress.update(task_id, completed=progress, details=details)

377 elif metadata["type"] == "spinner":

378 # Handle spinner layer - update details message

379 self.progress.update(

380 task_id,

381 details=details, # Use details consistently

382 )

383 elif metadata["type"] == "download":

384 # Handle download layer - update progress and details

385 self.progress.update(task_id, completed=progress, details=details)

386 else:

387 # Handle regular progress layer

388 self.progress.update(task_id, completed=progress, details=details)

389

390 def complete_layer(self, layer_name: str) -> None:

391 """Mark a layer as completed and animate its success.

392

393 Args:

394 layer_name: Name of the layer to complete

395 """

396 with self._lock:

397 if not self.progress: 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true

398 return

399

400 task_id = self.task_ids.get(layer_name)

401 if task_id is None: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true

402 return

403

404 # Mark as completed based on layer type

405 metadata = self.layer_metadata.get(task_id)

406 if metadata is None: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true

407 return

408

409 if metadata["type"] == "steps":

410 steps = metadata["config"].get("steps", [])

411 self.progress.update(task_id, completed=len(steps))

412 elif metadata["type"] == "spinner":

413 # For spinners, just mark as completed (no progress to update)

414 pass

415 else:

416 total = metadata["config"].get("total", 100)

417 self.progress.update(task_id, completed=total)

418

419 # Don't remove main layer - it stays for reference

420 if metadata.get("is_main", False): 420 ↛ 422line 420 didn't jump to line 422 because the condition on line 420 was never true

421 # Just mark as completed but keep it visible

422 self.completed_layers.append(task_id)

423 return

424

425 # Remove the layer (only for sub-layers)

426 self.completed_layers.append(task_id)

427 metadata["state"] = "completed"

428

429 # Animate success for this specific layer

430 self._animate_layer_success(task_id)

431

432 # Update main layer progress if it exists

433 if self.has_main_layer:

434 self._update_main_layer_progress()

435

436 @staticmethod

437 def _clean_description(description: str) -> str:

438 """Remove status icons from a task description.

439

440 Args:

441 description: The task description string

442

443 Returns:

444 Cleaned description without status icons

445 """

446 for icon in ("❌ ", "⚠️ ", "✅ "):

447 description = description.replace(icon, "")

448 return description

449

450 def _get_task(self, task_id: TaskID) -> Any | None:

451 """Safely get a task from the progress bar.

452

453 Args:

454 task_id: Task ID to retrieve

455

456 Returns:

457 The task object, or None if not found

458 """

459 if not self.progress: 459 ↛ 460line 459 didn't jump to line 460 because the condition on line 459 was never true

460 return None

461 tasks = getattr(self.progress, "_tasks", {})

462 return tasks.get(task_id)

463

464 def _has_task(self, task_id: TaskID) -> bool:

465 """Check if a task exists in the progress bar.

466

467 Args:

468 task_id: Task ID to check

469

470 Returns:

471 True if the task exists

472 """

473 return self._get_task(task_id) is not None

474

475 def _animate_layer_success(self, task_id: TaskID) -> None:

476 """Animate success for a specific layer and then remove it.

477

478 Args:

479 task_id: Task ID to animate

480 """

481 if not self.progress: 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true

482 return

483

484 # Flash green 2 times

485 for flash in range(2):

486 task = self._get_task(task_id)

487 if task is not None: 487 ↛ 502line 487 didn't jump to line 502 because the condition on line 487 was always true

488 clean_description = self._clean_description(str(task.description))

489

490 if flash % 2 == 0: # Green flash

491 success_description = Text(

492 clean_description, style="bold green on green"

493 )

494 else: # Normal green

495 success_description = Text(clean_description, style="bold green")

496

497 self.progress.update(

498 task_id,

499 description=success_description, # type: ignore[arg-type]

500 )

501

502 time.sleep(0.1) # Quick flash

503

504 # Fade out by updating with dim style

505 task = self._get_task(task_id)

506 if task is not None: 506 ↛ 513line 506 didn't jump to line 513 because the condition on line 506 was always true

507 clean_description = self._clean_description(str(task.description))

508 faded_description = Text(clean_description, style="dim")

509 self.progress.update(task_id, description=faded_description) # type: ignore[arg-type]

510 time.sleep(0.3) # Brief fade out

511

512 # Remove the layer after animation

513 if self._has_task(task_id): 513 ↛ exitline 513 didn't return from function '_animate_layer_success' because the condition on line 513 was always true

514 self.progress.remove_task(task_id)

515 if task_id in self.active_layers: 515 ↛ 518line 515 didn't jump to line 518 because the condition on line 515 was always true

516 self.active_layers.remove(task_id)

517 # Remove from task_ids dict

518 for name, tid in list(self.task_ids.items()): 518 ↛ exitline 518 didn't return from function '_animate_layer_success' because the loop on line 518 didn't complete

519 if tid == task_id:

520 del self.task_ids[name]

521 break

522

523 def _update_main_layer_progress(self) -> None:

524 """Update main layer progress based on completed sub-layers."""

525 if not self.has_main_layer or not self.main_layer_name: 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true

526 return

527

528 if not self.progress: 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true

529 return

530

531 # Find main layer task

532 main_task_id: TaskID | None = None

533 for tid, metadata in self.layer_metadata.items(): 533 ↛ 538line 533 didn't jump to line 538 because the loop on line 533 didn't complete

534 if metadata.get("is_main", False): 534 ↛ 533line 534 didn't jump to line 533 because the condition on line 534 was always true

535 main_task_id = tid

536 break

537

538 if main_task_id is None: 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true

539 return

540

541 # Calculate progress based on completed sub-layers only (exclude main layer)

542 completed_sub_layers = sum(

543 1

544 for tid in self.completed_layers

545 if not self.layer_metadata.get(tid, {}).get("is_main", False)

546 )

547

548 # Update main layer

549 self.progress.update(main_task_id, completed=completed_sub_layers)

550

551 def handle_error(self, layer_name: str, error: str) -> None:

552 """Handle errors in a specific layer.

553

554 Args:

555 layer_name: Name of the layer with error

556 error: Error message to display

557 """

558 with self._lock:

559 if not self.progress: 559 ↛ 560line 559 didn't jump to line 560 because the condition on line 559 was never true

560 return

561

562 task_id = self.task_ids.get(layer_name)

563 if task_id is None: 563 ↛ 564line 563 didn't jump to line 564 because the condition on line 563 was never true

564 return

565

566 # Update with error styling using Rich Text objects

567 task = self._get_task(task_id)

568 if task is not None: 568 ↛ exitline 568 didn't jump to the function exit

569 error_description = Text(f"❌ {task.description}", style="red")

570 error_details = Text(f"Error: {error}", style="red")

571

572 self.progress.update(

573 task_id,

574 description=error_description, # type: ignore[arg-type]

575 details=error_details,

576 )

577

578 def emergency_stop(self, error_message: str = "Critical error occurred") -> None:

579 """Emergency stop all layers with animated failure effects.

580

581 Args:

582 error_message: The error message to display

583 """

584 with self._lock:

585 if not self.progress: 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true

586 return

587

588 # Create failure animation sequence: flash red 3 times

589 for flash in range(3):

590 # Apply flash effect to all active layers

591 for task_id in list(self.active_layers):

592 task = self._get_task(task_id)

593 if task is not None: 593 ↛ 591line 593 didn't jump to line 591 because the condition on line 593 was always true

594 clean_description = self._clean_description(

595 str(task.description)

596 )

597

598 if flash % 2 == 0: # Red flash

599 error_description = Text(

600 clean_description, style="bold red on red"

601 )

602 error_details = Text(

603 f"Stopped: {error_message}", style="red on red"

604 )

605 else: # Normal red

606 error_description = Text(

607 clean_description, style="bold red"

608 )

609 error_details = Text(

610 f"Stopped: {error_message}", style="red"

611 )

612

613 self.progress.update(

614 task_id,

615 description=error_description, # type: ignore[arg-type]

616 details=error_details,

617 )

618

619 # Brief pause for flash effect

620 time.sleep(0.15)

621

622 # Final state: settle on clean error display

623 for task_id in list(self.active_layers):

624 task = self._get_task(task_id)

625 if task is not None: 625 ↛ 623line 625 didn't jump to line 623 because the condition on line 625 was always true

626 clean_description = self._clean_description(str(task.description))

627 error_description = Text(clean_description, style="bold red")

628 error_details = Text(f"Stopped: {error_message}", style="red")

629

630 self.progress.update(

631 task_id,

632 description=error_description, # type: ignore[arg-type]

633 details=error_details,

634 )

635

636 # Stop the progress bar to freeze the display

637 self.progress.stop()

638

639 # Mark as emergency stopped

640 self._emergency_stopped = True

641 self._emergency_message = error_message

642

643 def is_emergency_stopped(self) -> bool:

644 """Check if the progress bar was emergency stopped.

645

646 Returns:

647 True if emergency stopped, False otherwise

648 """

649 return self._emergency_stopped

650

651 def get_emergency_message(self) -> str | None:

652 """Get the emergency stop message.

653

654 Returns:

655 The emergency message if stopped, None otherwise

656 """

657 return self._emergency_message

658

659 def start(self) -> None:

660 """Start the progress bar and create initial layers."""

661 self.progress = self._create_progress_bar()

662 self.progress.start()

663

664 # Create layers in order: main layer first, then sub-layers

665 if self.has_main_layer:

666 # Create main layer first

667 main_stage = next(

668 (stage for stage in self.stages if stage.get("type") == "main"), None

669 )

670 if main_stage: 670 ↛ 674line 670 didn't jump to line 674 because the condition on line 670 was always true

671 self._create_layer(main_stage)

672

673 # Then create sub-layers

674 for stage in self.stages:

675 if stage.get("type") != "main":

676 self._create_layer(stage)

677 else:

678 # No main layer, create all layers in order

679 for stage in self.stages:

680 self._create_layer(stage)

681

682 def stop(self, success: bool = True, show_success_animation: bool = True) -> None:

683 """Stop the progress bar with appropriate animations.

684

685 Args:

686 success: Whether this stop represents a successful completion

687 show_success_animation: Whether to show success animations

688 """

689 if not self.progress: 689 ↛ 690line 689 didn't jump to line 690 because the condition on line 689 was never true

690 return

691

692 if success and show_success_animation: 692 ↛ 695line 692 didn't jump to line 695 because the condition on line 692 was always true

693 # SUCCESS CASE: Final cleanup for any remaining layers

694 time.sleep(0.5)

695 elif not success:

696 # ERROR/WARNING CASE: Freeze current state

697 for task_id in list(self.active_layers):

698 task = self._get_task(task_id)

699 if task is not None:

700 clean_description = self._clean_description(str(task.description))

701 error_description = Text(clean_description, style="bold orange")

702

703 self.progress.update(

704 task_id,

705 description=error_description, # type: ignore[arg-type]

706 )

707

708 # Stop the underlying Rich progress

709 self.progress.stop()

710

711

712class DynamicProgressMixin:

713 """

714 Mixin providing dynamic layered progress bar methods for RichWizard.

715

716 This mixin adds dynamic progress functionality with layers that can

717 appear, progress, and disappear automatically.

718 """

719

720 # Type hints for attributes provided by RichWizard

721 _console: Console

722 _progress_prefix: str

723

724 # ///////////////////////////////////////////////////////////////

725 # DYNAMIC PROGRESS METHODS

726 # ///////////////////////////////////////////////////////////////

727

728 @contextmanager

729 def dynamic_layered_progress(

730 self,

731 stages: list[StageConfig],

732 show_time: bool = True,

733 ) -> Generator[DynamicLayeredProgress, None, None]:

734 """

735 Create a dynamic layered progress bar context manager.

736

737 Args:

738 stages: List of stage configurations, each containing:

739 - 'name': Layer name/identifier

740 - 'type': Layer type ('progress', 'steps', 'spinner', 'download', 'main')

741 - 'description': Display description for the layer

742 - 'style': Rich style string (optional)

743 - 'total': Total items (for 'progress' or 'download' type)

744 - 'steps': List of step names (for 'steps' or 'main' type)

745 - 'total_size': Total size in bytes (for 'download' type)

746 - 'filename': Filename (for 'download' type)

747 show_time: Whether to show elapsed and remaining time

748

749 Yields:

750 DynamicLayeredProgress instance with methods:

751 - update_layer(name, progress, details): Update a layer's progress

752 - complete_layer(name): Mark a layer as completed

753 - handle_error(name, error): Handle errors in a layer

754 - emergency_stop(message): Emergency stop all layers

755

756 Example:

757 >>> stages = [

758 ... {"name": "main", "type": "main", "description": "Overall Progress"},

759 ... {"name": "step1", "type": "progress", "description": "Step 1", "total": 100},

760 ... {"name": "step2", "type": "steps", "description": "Step 2", "steps": ["A", "B", "C"]},

761 ... ]

762 >>> with printer.wizard.dynamic_layered_progress(stages) as progress:

763 ... progress.update_layer("step1", 50, "Processing...")

764 ... progress.complete_layer("step1")

765 ... progress.update_layer("step2", 1, "Step B")

766 ... progress.complete_layer("step2")

767 """

768 progress_bar = DynamicLayeredProgress(

769 self._console, self._progress_prefix, stages, show_time

770 )

771

772 try:

773 progress_bar.start()

774 yield progress_bar

775 except BaseException:

776 if not progress_bar.is_emergency_stopped():

777 progress_bar.stop(success=False, show_success_animation=False)

778 raise

779 else:

780 if not progress_bar.is_emergency_stopped():

781 progress_bar.stop(success=True, show_success_animation=True)

782 finally:

783 if progress_bar.is_emergency_stopped():

784 emergency_msg = progress_bar.get_emergency_message()

785 if emergency_msg: 785 ↛ exitline 785 didn't return from function 'dynamic_layered_progress' because the condition on line 785 was always true

786 self._console.print(

787 f"\n[bold red]🚨 EMERGENCY STOP: {emergency_msg}[/bold red]"

788 )

Coverage for src / ezpl / handlers / wizard / dynamic.py: 82.50%

296 statements