File size: 2,330 Bytes
599abe6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c079f49
599abe6
 
 
 
 
 
 
 
 
c079f49
599abe6
 
 
 
 
 
 
 
 
 
c079f49
599abe6
 
 
 
 
 
 
 
 
c079f49
 
 
 
 
 
 
 
 
 
599abe6
c079f49
599abe6
 
 
 
c079f49
599abe6
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import time
import json
from pathlib import Path

def load_messages() -> list:
    """Load status messages from JSON file."""
    with open('loading_messages.json', 'r') as f:
        return json.load(f)['messages']

class StatusTracker:
    """
    Track and display progress status for video generation.
    """
    def __init__(self, progress, status_box=None):
        self.progress = progress
        self.status_box = status_box
        self.steps = []
        self.current_message = ""
        self._status_markdown = ""
        
    def add_step(self, message: str, progress_value: float):
        """
        Add a permanent step to the progress display.
        
        Args:
            message: Step description
            progress_value: Progress value between 0 and 1
        """
        self.steps.append(message)
        self._update_display(progress_value)
        
    def update_message(self, message: str, progress_value: float):
        """
        Update the current working message.
        
        Args:
            message: Current status message
            progress_value: Progress value between 0 and 1
        """
        self.current_message = message
        self._update_display(progress_value)
        
    def _update_display(self, progress_value: float):
        """
        Update the status display with current progress.
        
        Args:
            progress_value: Progress value between 0 and 1
        """
        # Create status display
        status_md = "### 🎬 Generation Progress\n\n"
        
        # Show completed steps
        if self.steps:
            for step in self.steps:
                status_md += f"βœ“ {step}\n"
            status_md += "\n"
        
        # Show current message prominently
        if self.current_message:
            status_md += f"### πŸ”„ {self.current_message}\n"
            
        self._status_markdown = status_md
        self.progress(progress_value)
        
        # Update status box if it exists
        if self.status_box is not None:
            try:
                self.status_box.update(value=self._status_markdown)
            except:
                pass  # Silently handle if update fails
    
    def get_status(self) -> str:
        """Get the current status markdown."""
        return self._status_markdown