Create artifact directory for a program
(self, program_id: str)
| 2432 | return dct |
| 2433 | |
| 2434 | def _create_artifact_dir(self, program_id: str) -> str: |
| 2435 | """Create artifact directory for a program""" |
| 2436 | base_path = getattr(self.config, "artifacts_base_path", None) |
| 2437 | if not base_path: |
| 2438 | base_path = ( |
| 2439 | os.path.join(self.config.db_path or ".", "artifacts") |
| 2440 | if self.config.db_path |
| 2441 | else "./artifacts" |
| 2442 | ) |
| 2443 | |
| 2444 | artifact_dir = os.path.join(base_path, program_id) |
| 2445 | os.makedirs(artifact_dir, exist_ok=True) |
| 2446 | return artifact_dir |
| 2447 | |
| 2448 | def _cleanup_old_artifacts(self, checkpoint_path: str) -> None: |
| 2449 | """ |