Coverage for src / ezplog / handlers / wizard / dynamic.py: 82.54%
297 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-03 16:27 +0000
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("")
104class _ConditionalDetailsColumn(TextColumn):
105 """A text column that only shows details if the field exists."""
107 def __init__(self):
108 super().__init__("") # Empty text format, we override render
110 def render(self, task):
111 """
112 Render the details field if it exists in the task.
114 Args:
115 task: The task object to render
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("")
125class DynamicLayeredProgress:
126 """
127 Manages a dynamic layered progress bar with disappearing layers.
129 This class provides a progress bar system where layers can appear,
130 progress, and disappear based on the current state of operations.
131 """
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.
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()
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] = []
169 # Detect main layer and setup hierarchy
170 self._setup_hierarchy()
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 = []
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
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 ]
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 ]
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 )
221 if self._show_time:
222 columns.extend(
223 [
224 TimeElapsedColumn(),
225 TimeRemainingColumn(),
226 ]
227 )
229 return Progress(*columns, console=self._console)
231 def _create_layer(self, layer_config: StageConfig) -> int:
232 """Create a new layer in the progress bar.
234 Args:
235 layer_config: Layer configuration dictionary
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")
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}"
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
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 )
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 }
311 self._active_layers.append(task_id)
312 self._task_ids[layer_name] = task_id
313 return task_id
315 def update_layer(self, layer_name: str, progress: int, details: str = "") -> None:
316 """Update a specific layer's progress.
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
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
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
335 self._update_layer_unsafe(task_id, metadata, progress, details)
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).
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
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 )
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}"
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)
390 def complete_layer(self, layer_name: str) -> None:
391 """Mark a layer as completed and animate its success.
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
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
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
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)
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
425 # Remove the layer (only for sub-layers)
426 self._completed_layers.append(task_id)
427 metadata["state"] = "completed"
429 # Animate success for this specific layer
430 self._animate_layer_success(task_id)
432 # Update main layer progress if it exists
433 if self._has_main_layer:
434 self._update_main_layer_progress()
436 @staticmethod
437 def _clean_description(description: str) -> str:
438 """Remove status icons from a task description.
440 Args:
441 description: The task description string
443 Returns:
444 Cleaned description without status icons
445 """
446 for icon in ("❌ ", "⚠️ ", "✅ "):
447 description = description.replace(icon, "")
448 return description
450 def _get_task(self, task_id: TaskID) -> Any | None:
451 """Safely get a task from the progress bar.
453 Args:
454 task_id: Task ID to retrieve
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)
464 def _has_task(self, task_id: TaskID) -> bool:
465 """Check if a task exists in the progress bar.
467 Args:
468 task_id: Task ID to check
470 Returns:
471 True if the task exists
472 """
473 return self._get_task(task_id) is not None
475 def _animate_layer_success(self, task_id: TaskID) -> None:
476 """Animate success for a specific layer and then remove it.
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
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 ↛ 500line 487 didn't jump to line 500 because the condition on line 487 was always true
488 clean_description = self._clean_description(str(task.description))
490 if flash % 2 == 0: # Green flash
491 success_description = f"[bold green on green]{clean_description}[/]"
492 else: # Normal green
493 success_description = f"[bold green]{clean_description}[/]"
495 self._progress.update(
496 task_id,
497 description=success_description,
498 )
500 time.sleep(0.1) # Quick flash
502 # Fade out by updating with dim style
503 task = self._get_task(task_id)
504 if task is not None: 504 ↛ 511line 504 didn't jump to line 511 because the condition on line 504 was always true
505 clean_description = self._clean_description(str(task.description))
506 faded_description = f"[dim]{clean_description}[/]"
507 self._progress.update(task_id, description=faded_description)
508 time.sleep(0.3) # Brief fade out
510 # Remove the layer after animation
511 if self._has_task(task_id): 511 ↛ exitline 511 didn't return from function '_animate_layer_success' because the condition on line 511 was always true
512 self._progress.remove_task(task_id)
513 if task_id in self._active_layers: 513 ↛ 516line 513 didn't jump to line 516 because the condition on line 513 was always true
514 self._active_layers.remove(task_id)
515 # Remove from task_ids dict
516 for name, tid in list(self._task_ids.items()): 516 ↛ exitline 516 didn't return from function '_animate_layer_success' because the loop on line 516 didn't complete
517 if tid == task_id:
518 del self._task_ids[name]
519 break
521 def _update_main_layer_progress(self) -> None:
522 """Update main layer progress based on completed sub-layers."""
523 if not self._has_main_layer or not self._main_layer_name: 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true
524 return
526 if not self._progress: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 return
529 # Find main layer task
530 main_task_id: TaskID | None = None
531 for tid, metadata in self._layer_metadata.items(): 531 ↛ 536line 531 didn't jump to line 536 because the loop on line 531 didn't complete
532 if metadata.get("is_main", False): 532 ↛ 531line 532 didn't jump to line 531 because the condition on line 532 was always true
533 main_task_id = tid
534 break
536 if main_task_id is None: 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 return
539 # Calculate progress based on completed sub-layers only (exclude main layer)
540 completed_sub_layers = sum(
541 1
542 for tid in self._completed_layers
543 if not self._layer_metadata.get(tid, {}).get("is_main", False)
544 )
546 # Update main layer
547 self._progress.update(main_task_id, completed=completed_sub_layers)
549 def handle_error(self, layer_name: str, error: str) -> None:
550 """Handle errors in a specific layer.
552 Args:
553 layer_name: Name of the layer with error
554 error: Error message to display
555 """
556 with self._lock:
557 if not self._progress: 557 ↛ 558line 557 didn't jump to line 558 because the condition on line 557 was never true
558 return
560 task_id = self._task_ids.get(layer_name)
561 if task_id is None: 561 ↛ 562line 561 didn't jump to line 562 because the condition on line 561 was never true
562 return
564 # Update with error styling using Rich Text objects
565 task = self._get_task(task_id)
566 if task is not None: 566 ↛ exitline 566 didn't jump to the function exit
567 error_description = f"[red]❌ {task.description}[/]"
568 error_details = f"[red]Error: {error}[/]"
570 self._progress.update(
571 task_id,
572 description=error_description,
573 details=error_details,
574 )
576 def emergency_stop(self, error_message: str = "Critical error occurred") -> None:
577 """Emergency stop all layers with animated failure effects.
579 Args:
580 error_message: The error message to display
581 """
582 with self._lock:
583 if not self._progress: 583 ↛ 584line 583 didn't jump to line 584 because the condition on line 583 was never true
584 return
586 # Create failure animation sequence: flash red 3 times
587 for flash in range(3):
588 # Apply flash effect to all active layers
589 for task_id in list(self._active_layers):
590 task = self._get_task(task_id)
591 if task is not None: 591 ↛ 589line 591 didn't jump to line 589 because the condition on line 591 was always true
592 clean_description = self._clean_description(
593 str(task.description)
594 )
596 if flash % 2 == 0: # Red flash
597 error_description = (
598 f"[bold red on red]{clean_description}[/]"
599 )
600 error_details = f"[red on red]Stopped: {error_message}[/]"
601 else: # Normal red
602 error_description = f"[bold red]{clean_description}[/]"
603 error_details = f"[red]Stopped: {error_message}[/]"
605 self._progress.update(
606 task_id,
607 description=error_description,
608 details=error_details,
609 )
611 # Brief pause for flash effect
612 time.sleep(0.15)
614 # Final state: settle on clean error display
615 for task_id in list(self._active_layers):
616 task = self._get_task(task_id)
617 if task is not None: 617 ↛ 615line 617 didn't jump to line 615 because the condition on line 617 was always true
618 clean_description = self._clean_description(str(task.description))
619 error_description = f"[bold red]{clean_description}[/]"
620 error_details = f"[red]Stopped: {error_message}[/]"
622 self._progress.update(
623 task_id,
624 description=error_description,
625 details=error_details,
626 )
628 # Stop the progress bar to freeze the display
629 self._progress.stop()
631 # Mark as emergency stopped
632 self._emergency_stopped = True
633 self._emergency_message = error_message
635 def is_emergency_stopped(self) -> bool:
636 """Check if the progress bar was emergency stopped.
638 Returns:
639 True if emergency stopped, False otherwise
640 """
641 return self._emergency_stopped
643 def get_emergency_message(self) -> str | None:
644 """Get the emergency stop message.
646 Returns:
647 The emergency message if stopped, None otherwise
648 """
649 return self._emergency_message
651 def start(self) -> None:
652 """Start the progress bar and create initial layers."""
653 self._progress = self._create_progress_bar()
654 self._progress.start()
656 # Create layers in order: main layer first, then sub-layers
657 if self._has_main_layer:
658 # Create main layer first
659 main_stage = next(
660 (stage for stage in self._stages if stage.get("type") == "main"), None
661 )
662 if main_stage: 662 ↛ 666line 662 didn't jump to line 666 because the condition on line 662 was always true
663 self._create_layer(main_stage)
665 # Then create sub-layers
666 for stage in self._stages:
667 if stage.get("type") != "main":
668 self._create_layer(stage)
669 else:
670 # No main layer, create all layers in order
671 for stage in self._stages:
672 self._create_layer(stage)
674 def stop(self, success: bool = True, show_success_animation: bool = True) -> None:
675 """Stop the progress bar with appropriate animations.
677 Args:
678 success: Whether this stop represents a successful completion
679 show_success_animation: Whether to show success animations
680 """
681 if not self._progress: 681 ↛ 682line 681 didn't jump to line 682 because the condition on line 681 was never true
682 return
684 if success and show_success_animation: 684 ↛ 687line 684 didn't jump to line 687 because the condition on line 684 was always true
685 # SUCCESS CASE: Final cleanup for any remaining layers
686 time.sleep(0.5)
687 elif not success:
688 # ERROR/WARNING CASE: Freeze current state
689 for task_id in list(self._active_layers):
690 task = self._get_task(task_id)
691 if task is not None:
692 clean_description = self._clean_description(str(task.description))
693 error_description = f"[bold orange]{clean_description}[/]"
695 self._progress.update(
696 task_id,
697 description=error_description,
698 )
700 # Stop the underlying Rich progress
701 self._progress.stop()
704class DynamicProgressMixin:
705 """
706 Mixin providing dynamic layered progress bar methods for RichWizard.
708 This mixin adds dynamic progress functionality with layers that can
709 appear, progress, and disappear automatically.
710 """
712 # Type hints for attributes provided by RichWizard
713 _console: Console
714 _progress_prefix: str
716 # ///////////////////////////////////////////////////////////////
717 # DYNAMIC PROGRESS METHODS
718 # ///////////////////////////////////////////////////////////////
720 @contextmanager
721 def dynamic_layered_progress(
722 self,
723 stages: list[StageConfig],
724 show_time: bool = True,
725 ) -> Generator[DynamicLayeredProgress, None, None]:
726 """
727 Create a dynamic layered progress bar context manager.
729 Args:
730 stages: List of stage configurations, each containing:
731 - 'name': Layer name/identifier
732 - 'type': Layer type ('progress', 'steps', 'spinner', 'download', 'main')
733 - 'description': Display description for the layer
734 - 'style': Rich style string (optional)
735 - 'total': Total items (for 'progress' or 'download' type)
736 - 'steps': List of step names (for 'steps' or 'main' type)
737 - 'total_size': Total size in bytes (for 'download' type)
738 - 'filename': Filename (for 'download' type)
739 show_time: Whether to show elapsed and remaining time
741 Yields:
742 DynamicLayeredProgress instance with methods:
743 - update_layer(name, progress, details): Update a layer's progress
744 - complete_layer(name): Mark a layer as completed
745 - handle_error(name, error): Handle errors in a layer
746 - emergency_stop(message): Emergency stop all layers
748 Example:
749 >>> stages = [
750 ... {"name": "main", "type": "main", "description": "Overall Progress"},
751 ... {"name": "step1", "type": "progress", "description": "Step 1", "total": 100},
752 ... {"name": "step2", "type": "steps", "description": "Step 2", "steps": ["A", "B", "C"]},
753 ... ]
754 >>> with printer.wizard.dynamic_layered_progress(stages) as progress:
755 ... progress.update_layer("step1", 50, "Processing...")
756 ... progress.complete_layer("step1")
757 ... progress.update_layer("step2", 1, "Step B")
758 ... progress.complete_layer("step2")
759 """
760 progress_bar = DynamicLayeredProgress(
761 self._console, self._progress_prefix, stages, show_time
762 )
764 try:
765 progress_bar.start()
766 yield progress_bar
767 except BaseException:
768 if not progress_bar.is_emergency_stopped():
769 progress_bar.stop(success=False, show_success_animation=False)
770 raise
771 else:
772 if not progress_bar.is_emergency_stopped():
773 progress_bar.stop(success=True, show_success_animation=True)
774 finally:
775 if progress_bar.is_emergency_stopped():
776 emergency_msg = progress_bar.get_emergency_message()
777 if emergency_msg: 777 ↛ exitline 777 didn't return from function 'dynamic_layered_progress' because the condition on line 777 was always true
778 self._console.print(
779 f"\n[bold red]🚨 EMERGENCY STOP: {emergency_msg}[/bold red]"
780 )
783# ///////////////////////////////////////////////////////////////
784# PUBLIC API
785# ///////////////////////////////////////////////////////////////
787__all__ = [
788 "DynamicLayeredProgress",
789 "DynamicProgressMixin",
790 "StageConfig",
791]