Load workflow status from file.
(self)
| 75 | self._load_status() |
| 76 | |
| 77 | def _load_status(self): |
| 78 | """Load workflow status from file.""" |
| 79 | if self.status_file.exists(): |
| 80 | try: |
| 81 | with open(self.status_file, "r") as f: |
| 82 | data = json.load(f) |
| 83 | |
| 84 | for phase_name, result_data in data.get("results", {}).items(): |
| 85 | phase = WorkflowPhase(phase_name) |
| 86 | result = WorkflowResult( |
| 87 | phase=phase, |
| 88 | status=WorkflowStatus(result_data["status"]), |
| 89 | start_time=datetime.fromisoformat(result_data["start_time"]), |
| 90 | end_time=datetime.fromisoformat(result_data["end_time"]) if result_data.get("end_time") else None, |
| 91 | duration=result_data.get("duration"), |
| 92 | artifacts=result_data.get("artifacts", []), |
| 93 | error_message=result_data.get("error_message"), |
| 94 | ) |
| 95 | self.results[phase] = result |
| 96 | |
| 97 | except Exception as e: |
| 98 | print(f"Warning: Could not load workflow status: {e}") |
| 99 | |
| 100 | def _save_status(self): |
| 101 | """Save workflow status to file.""" |
no test coverage detected