| import shutil |
| from pathlib import Path |
|
|
| try: |
| from trackio.utils import MEDIA_DIR |
| except ImportError: |
| from utils import MEDIA_DIR |
|
|
|
|
| def check_path(file_path: str | Path) -> None: |
| """Raise an error if the parent directory does not exist.""" |
| file_path = Path(file_path) |
| if not file_path.parent.exists(): |
| try: |
| file_path.parent.mkdir(parents=True, exist_ok=True) |
| except OSError as e: |
| raise ValueError( |
| f"Failed to create parent directory {file_path.parent}: {e}" |
| ) |
|
|
|
|
| def check_ffmpeg_installed() -> None: |
| """Raise an error if ffmpeg is not available on the system PATH.""" |
| if shutil.which("ffmpeg") is None: |
| raise RuntimeError( |
| "ffmpeg is required to write video but was not found on your system. " |
| "Please install ffmpeg and ensure it is available on your PATH." |
| ) |
|
|
|
|
| def get_project_media_path( |
| project: str, |
| run: str | None = None, |
| step: int | None = None, |
| relative_path: str | Path | None = None, |
| ) -> Path: |
| """ |
| Get the full path where uploaded files are stored for a Trackio project (and create the directory if it doesn't exist). |
| If a run is not provided, the files are stored in a project-level directory with the given relative path. |
| |
| Args: |
| project: The project name |
| run: The run name |
| step: The step number |
| relative_path: The relative path within the directory (only used if run is not provided) |
| |
| Returns: |
| The full path to the media file |
| """ |
| if step is not None and run is None: |
| raise ValueError("Uploading files at a specific step requires a run") |
|
|
| path = MEDIA_DIR / project |
| if run: |
| path /= run |
| if step is not None: |
| path /= str(step) |
| else: |
| path /= "files" |
| if relative_path: |
| path /= relative_path |
| path.mkdir(parents=True, exist_ok=True) |
| return path |
|
|