()
| 245 | scene_tasks: dict[str, int] = {} |
| 246 | |
| 247 | def refresh() -> None: |
| 248 | snapshot = dict(status_dict) |
| 249 | done = sum(1 for v in snapshot.values() if v == "finished") |
| 250 | overall.update(overall_task, completed=done) |
| 251 | in_progress = {k: v for k, v in snapshot.items() if v != "finished"} |
| 252 | for key in list(scene_tasks): |
| 253 | if key not in in_progress: |
| 254 | scenes.remove_task(scene_tasks.pop(key)) |
| 255 | for key, phase in in_progress.items(): |
| 256 | if key in scene_tasks: |
| 257 | scenes.update(scene_tasks[key], phase=phase) |
| 258 | else: |
| 259 | scene_tasks[key] = scenes.add_task(key, total=None, phase=phase) |
| 260 | |
| 261 | with Live(Group(overall, scenes), refresh_per_second=4): |
| 262 | while not stop_event.is_set(): |
no outgoing calls
no test coverage detected