MogensR commited on
Commit
f56676e
·
verified ·
1 Parent(s): d4c0c41

Create progress_tracker.py

Browse files
Files changed (1) hide show
  1. utils/progress_tracker.py +75 -0
utils/progress_tracker.py ADDED
@@ -0,0 +1,75 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ #!/usr/bin/env python3
2
+ """
3
+ Progress tracking via file-based status updates
4
+ Allows background processing while UI stays responsive
5
+ """
6
+ import json
7
+ import os
8
+ from pathlib import Path
9
+ from typing import Optional
10
+
11
+ PROGRESS_FILE = "/tmp/processing_status.json"
12
+
13
+ def init_progress():
14
+ """Initialize progress tracking"""
15
+ status = {
16
+ "active": False,
17
+ "message": "",
18
+ "progress": 0,
19
+ "stage": "",
20
+ "error": None,
21
+ "complete": False
22
+ }
23
+ with open(PROGRESS_FILE, "w") as f:
24
+ json.dump(status, f)
25
+
26
+ def update_progress(message: str, progress: int = None, stage: str = None):
27
+ """Update progress status"""
28
+ try:
29
+ if os.path.exists(PROGRESS_FILE):
30
+ with open(PROGRESS_FILE, "r") as f:
31
+ status = json.load(f)
32
+ else:
33
+ status = {"active": True, "message": "", "progress": 0, "stage": "", "error": None, "complete": False}
34
+
35
+ status["message"] = message
36
+ status["active"] = True
37
+ if progress is not None:
38
+ status["progress"] = progress
39
+ if stage is not None:
40
+ status["stage"] = stage
41
+
42
+ with open(PROGRESS_FILE, "w") as f:
43
+ json.dump(status, f)
44
+ except Exception:
45
+ pass # Fail silently to not interrupt processing
46
+
47
+ def mark_complete(success: bool = True, error: str = None):
48
+ """Mark processing as complete"""
49
+ try:
50
+ if os.path.exists(PROGRESS_FILE):
51
+ with open(PROGRESS_FILE, "r") as f:
52
+ status = json.load(f)
53
+ else:
54
+ status = {}
55
+
56
+ status["active"] = False
57
+ status["complete"] = True
58
+ status["progress"] = 100 if success else status.get("progress", 0)
59
+ if error:
60
+ status["error"] = error
61
+
62
+ with open(PROGRESS_FILE, "w") as f:
63
+ json.dump(status, f)
64
+ except Exception:
65
+ pass
66
+
67
+ def get_progress() -> dict:
68
+ """Read current progress status"""
69
+ try:
70
+ if os.path.exists(PROGRESS_FILE):
71
+ with open(PROGRESS_FILE, "r") as f:
72
+ return json.load(f)
73
+ except Exception:
74
+ pass
75
+ return {"active": False, "message": "", "progress": 0, "stage": "", "error": None, "complete": False}