Skip to content

progress

progress

AsyncProgressTracker 

AsyncProgressTracker(*args, **kwargs)

Bases: ProgressTracker

Asynchronous version of progress tracker

Source code in src/process_manager/examples/monte_carlo/progress.py 
148
149
150
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self._lock = asyncio.Lock()

complete async 

complete(success: bool = True)

Complete progress tracking (async)

Source code in src/process_manager/examples/monte_carlo/progress.py 
172
173
174
175
async def complete(self, success: bool = True):
    """Complete progress tracking (async)"""
    async with self._lock:
        super().complete(success)

complete_case async 

complete_case(case_id: str, cached: bool = False)

Mark a case as completed (async)

Source code in src/process_manager/examples/monte_carlo/progress.py 
162
163
164
165
async def complete_case(self, case_id: str, cached: bool = False):
    """Mark a case as completed (async)"""
    async with self._lock:
        super().complete_case(case_id, cached)

fail_case async 

fail_case(case_id: str)

Mark a case as failed (async)

Source code in src/process_manager/examples/monte_carlo/progress.py 
167
168
169
170
async def fail_case(self, case_id: str):
    """Mark a case as failed (async)"""
    async with self._lock:
        super().fail_case(case_id)

start async 

start(total_cases: int)

Start progress tracking (async)

Source code in src/process_manager/examples/monte_carlo/progress.py 
152
153
154
155
async def start(self, total_cases: int):
    """Start progress tracking (async)"""
    async with self._lock:
        super().start(total_cases)

update_state async 

update_state(state: ProgressState)

Update progress state (async)

Source code in src/process_manager/examples/monte_carlo/progress.py 
157
158
159
160
async def update_state(self, state: ProgressState):
    """Update progress state (async)"""
    async with self._lock:
        super().update_state(state)

ProgressState

Bases: Enum

Detailed progress states for tracking

ProgressStats dataclass 

ProgressStats(
    total_cases: int = 0,
    completed_cases: int = 0,
    failed_cases: int = 0,
    cached_cases: int = 0,
    current_case: Optional[str] = None,
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
)

Statistics for progress tracking

elapsed_time property 

elapsed_time: Optional[float]

Get elapsed time in seconds

progress_percentage property 

progress_percentage: float

Get overall progress percentage

to_dict 

to_dict() -> Dict[str, Any]

Convert stats to dictionary format

Source code in src/process_manager/examples/monte_carlo/progress.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
def to_dict(self) -> Dict[str, Any]:
    """Convert stats to dictionary format"""
    return {
        "total_cases": self.total_cases,
        "completed_cases": self.completed_cases,
        "failed_cases": self.failed_cases,
        "cached_cases": self.cached_cases,
        "current_case": self.current_case,
        "start_time": self.start_time.isoformat() if self.start_time else None,
        "end_time": self.end_time.isoformat() if self.end_time else None,
        "elapsed_time": self.elapsed_time,
        "progress_percentage": self.progress_percentage
    }

ProgressTracker 

ProgressTracker(
    output_dir: Path,
    progress_callback: Optional[Callable[[ProgressState, ProgressStats], None]] = None,
)

Tracks and reports progress of Monte Carlo simulation.

Features: - Detailed progress states - Progress statistics - Progress file output - Progress callback support - ETA calculation

Source code in src/process_manager/examples/monte_carlo/progress.py 
72
73
74
75
76
77
78
79
80
81
82
def __init__(self,
             output_dir: Path,
             progress_callback: Optional[Callable[[ProgressState, ProgressStats], None]] = None):
    self.output_dir = output_dir
    self.progress_callback = progress_callback
    self.stats = ProgressStats()
    self.state = ProgressState.NOT_STARTED
    self.progress_file = output_dir / "progress.json"

    # Ensure output directory exists
    self.output_dir.mkdir(parents=True, exist_ok=True)

complete 

complete(success: bool = True)

Complete progress tracking

Source code in src/process_manager/examples/monte_carlo/progress.py 
110
111
112
113
114
def complete(self, success: bool = True):
    """Complete progress tracking"""
    self.stats.end_time = datetime.now()
    self.state = ProgressState.COMPLETED if success else ProgressState.FAILED
    self._update()

complete_case 

complete_case(case_id: str, cached: bool = False)

Mark a case as completed

Source code in src/process_manager/examples/monte_carlo/progress.py 
 96
 97
 98
 99
100
101
102
def complete_case(self, case_id: str, cached: bool = False):
    """Mark a case as completed"""
    self.stats.completed_cases += 1
    if cached:
        self.stats.cached_cases += 1
    self.stats.current_case = case_id
    self._update()

fail_case 

fail_case(case_id: str)

Mark a case as failed

Source code in src/process_manager/examples/monte_carlo/progress.py 
104
105
106
107
108
def fail_case(self, case_id: str):
    """Mark a case as failed"""
    self.stats.failed_cases += 1
    self.stats.current_case = case_id
    self._update()

start 

start(total_cases: int)

Start progress tracking

Source code in src/process_manager/examples/monte_carlo/progress.py 
84
85
86
87
88
89
def start(self, total_cases: int):
    """Start progress tracking"""
    self.stats.total_cases = total_cases
    self.stats.start_time = datetime.now()
    self.state = ProgressState.INITIALIZING
    self._update()

update_state 

update_state(state: ProgressState)

Update progress state

Source code in src/process_manager/examples/monte_carlo/progress.py 
91
92
93
94
def update_state(self, state: ProgressState):
    """Update progress state"""
    self.state = state
    self._update()