Support Gradio file paths for uploads
Browse files
frame_extraction/src/frame_extraction/app.py
CHANGED
|
@@ -6,7 +6,7 @@ import os
|
|
| 6 |
import shutil
|
| 7 |
import uuid
|
| 8 |
from pathlib import Path
|
| 9 |
-
from typing import Any, Tuple
|
| 10 |
|
| 11 |
import gradio as gr
|
| 12 |
import numpy as np
|
|
@@ -28,6 +28,16 @@ def ensure_output_dirs() -> None:
|
|
| 28 |
(OUTPUT_DIR / "frames").mkdir(parents=True, exist_ok=True)
|
| 29 |
|
| 30 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 31 |
def summarize_catalog(catalog_path: Path) -> Tuple[str, list[dict[str, Any]], list[list[str]]]:
|
| 32 |
if not catalog_path.exists():
|
| 33 |
return ("Catalog not found.", [], [])
|
|
@@ -51,31 +61,32 @@ def summarize_catalog(catalog_path: Path) -> Tuple[str, list[dict[str, Any]], li
|
|
| 51 |
return message, index, gallery
|
| 52 |
|
| 53 |
|
| 54 |
-
def build_catalog_from_video(files: list[gr.FileData]) -> tuple[str | None, str, list[dict[str, Any]], list[list[str]], list[dict[str, Any]], list[list[str]]]:
|
| 55 |
if not files:
|
| 56 |
raise gr.Error("Please upload a source video first.")
|
| 57 |
|
| 58 |
ensure_output_dirs()
|
| 59 |
-
|
|
|
|
|
|
|
|
|
|
| 60 |
run_id = uuid.uuid4().hex[:8]
|
| 61 |
video_dir = OUTPUT_DIR / "videos"
|
| 62 |
-
video_path = video_dir / f"{run_id}_{
|
| 63 |
-
shutil.copy(
|
| 64 |
|
| 65 |
catalog_dir = OUTPUT_DIR / "catalogs" / f"catalog_{run_id}"
|
| 66 |
cfg = CatalogConfig(video_path=video_path, output_dir=catalog_dir)
|
| 67 |
catalog_path = build_catalog(cfg)
|
| 68 |
message, index, gallery = summarize_catalog(catalog_path)
|
| 69 |
-
# When a new catalog is generated, clear existing matches display
|
| 70 |
empty_matches: list[dict[str, Any]] = []
|
| 71 |
empty_gallery: list[list[str]] = []
|
| 72 |
return str(catalog_path), message, index, gallery, empty_matches, empty_gallery
|
| 73 |
|
| 74 |
|
| 75 |
-
def
|
| 76 |
if not catalog_path:
|
| 77 |
raise gr.Error("Catalog not ready yet. Upload a video first.")
|
| 78 |
-
|
| 79 |
if not files:
|
| 80 |
raise gr.Error("Please upload at least one frame.")
|
| 81 |
|
|
@@ -85,8 +96,9 @@ def predict_from_arrays(files: list[gr.FileData], catalog_path: str | None) -> t
|
|
| 85 |
frames_dir.mkdir(parents=True, exist_ok=True)
|
| 86 |
|
| 87 |
arrays: list[np.ndarray] = []
|
| 88 |
-
for idx,
|
| 89 |
-
|
|
|
|
| 90 |
arrays.append(array)
|
| 91 |
Image.fromarray(array).save(frames_dir / f"upload_{idx:03d}.png")
|
| 92 |
|
|
@@ -147,7 +159,7 @@ def build_interface() -> gr.Blocks:
|
|
| 147 |
)
|
| 148 |
|
| 149 |
frame_upload.upload(
|
| 150 |
-
|
| 151 |
inputs=[frame_upload, catalog_state],
|
| 152 |
outputs=[matches_json, match_gallery],
|
| 153 |
)
|
|
|
|
| 6 |
import shutil
|
| 7 |
import uuid
|
| 8 |
from pathlib import Path
|
| 9 |
+
from typing import Any, Iterable, Tuple
|
| 10 |
|
| 11 |
import gradio as gr
|
| 12 |
import numpy as np
|
|
|
|
| 28 |
(OUTPUT_DIR / "frames").mkdir(parents=True, exist_ok=True)
|
| 29 |
|
| 30 |
|
| 31 |
+
def _resolve_file_path(entry: gr.FileData | str | Path) -> Path:
|
| 32 |
+
if isinstance(entry, Path):
|
| 33 |
+
return entry
|
| 34 |
+
if isinstance(entry, gr.FileData):
|
| 35 |
+
return Path(entry.name)
|
| 36 |
+
if isinstance(entry, str):
|
| 37 |
+
return Path(entry)
|
| 38 |
+
raise ValueError(f"Unsupported file type: {type(entry)!r}")
|
| 39 |
+
|
| 40 |
+
|
| 41 |
def summarize_catalog(catalog_path: Path) -> Tuple[str, list[dict[str, Any]], list[list[str]]]:
|
| 42 |
if not catalog_path.exists():
|
| 43 |
return ("Catalog not found.", [], [])
|
|
|
|
| 61 |
return message, index, gallery
|
| 62 |
|
| 63 |
|
| 64 |
+
def build_catalog_from_video(files: list[gr.FileData | str | Path]) -> tuple[str | None, str, list[dict[str, Any]], list[list[str]], list[dict[str, Any]], list[list[str]]]:
|
| 65 |
if not files:
|
| 66 |
raise gr.Error("Please upload a source video first.")
|
| 67 |
|
| 68 |
ensure_output_dirs()
|
| 69 |
+
source_path = _resolve_file_path(files[0])
|
| 70 |
+
if not source_path.exists():
|
| 71 |
+
raise gr.Error(f"Uploaded video not found: {source_path}")
|
| 72 |
+
|
| 73 |
run_id = uuid.uuid4().hex[:8]
|
| 74 |
video_dir = OUTPUT_DIR / "videos"
|
| 75 |
+
video_path = video_dir / f"{run_id}_{source_path.name}"
|
| 76 |
+
shutil.copy(source_path, video_path)
|
| 77 |
|
| 78 |
catalog_dir = OUTPUT_DIR / "catalogs" / f"catalog_{run_id}"
|
| 79 |
cfg = CatalogConfig(video_path=video_path, output_dir=catalog_dir)
|
| 80 |
catalog_path = build_catalog(cfg)
|
| 81 |
message, index, gallery = summarize_catalog(catalog_path)
|
|
|
|
| 82 |
empty_matches: list[dict[str, Any]] = []
|
| 83 |
empty_gallery: list[list[str]] = []
|
| 84 |
return str(catalog_path), message, index, gallery, empty_matches, empty_gallery
|
| 85 |
|
| 86 |
|
| 87 |
+
def predict_from_files(files: list[gr.FileData | str | Path], catalog_path: str | None) -> tuple[list[dict[str, Any]], list[list[str]]]:
|
| 88 |
if not catalog_path:
|
| 89 |
raise gr.Error("Catalog not ready yet. Upload a video first.")
|
|
|
|
| 90 |
if not files:
|
| 91 |
raise gr.Error("Please upload at least one frame.")
|
| 92 |
|
|
|
|
| 96 |
frames_dir.mkdir(parents=True, exist_ok=True)
|
| 97 |
|
| 98 |
arrays: list[np.ndarray] = []
|
| 99 |
+
for idx, entry in enumerate(files):
|
| 100 |
+
path = _resolve_file_path(entry)
|
| 101 |
+
array = np.array(Image.open(path).convert("RGB"))
|
| 102 |
arrays.append(array)
|
| 103 |
Image.fromarray(array).save(frames_dir / f"upload_{idx:03d}.png")
|
| 104 |
|
|
|
|
| 159 |
)
|
| 160 |
|
| 161 |
frame_upload.upload(
|
| 162 |
+
predict_from_files,
|
| 163 |
inputs=[frame_upload, catalog_state],
|
| 164 |
outputs=[matches_json, match_gallery],
|
| 165 |
)
|