Upload folder using huggingface_hub

fusion-app/app_local.py

@@ -3,6 +3,7 @@ import time
 import json
 import numpy as np
 from pathlib import Path
+from utils_media import video_to_frame_audio,  load_audio_16k
 
 HERE = Path(__file__).parent
 lables_PATH = HERE / "labels.json"
@@ -13,16 +14,19 @@ lables = [x["name"] for x in json.loads(lables_PATH.read_text())["labels"]]
 
 def predict_vid(video):
     t0= time.time()
+    frame, audio16k = video_to_frame_audio(video)
     probs = np.ones(len(lables))/len(lables)
     pred = lables[int(np.argmax(probs))]
-    lat = {"t_total_ms": int((time.time()-t0)*1000), "note": "dummy"}
+    lat = {"t_total_ms": int((time.time()-t0)*1000), "note": "decoded media"}
     return pred, {k: float(v) for k,v in zip(lables, probs)}, lat
 
 def predict_aud_img(audio, image):
     t0 = time.time()
+    wave = load_audio_16k(audio)
+    frame = image
     probs = np.ones(len(lables)) / len(lables)
     pred = lables[int(np.argmax(probs))]
-    lat = {"t_total_ms": int((time.time()-t0)*1000), "note": "dummy"}
+    lat = {"t_total_ms": int((time.time()-t0)*1000), "note": "loaded media"}
     return pred, {k: float(v) for k,v in zip(lables, probs)}, lat
 
 

fusion-app/labels.json

@@ -7,3 +7,5 @@
     {"name": "sad",       "prompt": "a somber, gloomy scene",             "def": "cool/dark tones, slow pace, quiet audio"}
   ]
 }
+
+

fusion-app/utils_media.py

@@ -0,0 +1,45 @@
+from pathlib import Path
+from typing import Tuple, Union
+import io
+import numpy as np
+from PIL import Image
+import ffmpeg 
+from pydub import AudioSegment
+
+#  helpers 
+def _to_path(p: Union[str, dict, Path]) -> str:
+    if isinstance(p, dict):
+        return p.get("name") or p.get("path") or p.get("data") or ""
+    return str(p)
+
+def _audiosegment_float32(seg: AudioSegment) -> np.ndarray:
+    seg = seg.set_frame_rate(16000).set_channels(1).set_sample_width(2)  # 16-bit
+    samples = np.array(seg.get_array_of_samples(), dtype=np.int16)
+    return (samples.astype(np.float32) / 32768.0)
+
+#  public API
+def video_to_frame_audio(video_in) -> Tuple[Image.Image, np.ndarray]:
+    video_path = _to_path(video_in)
+    if not video_path:
+        raise ValueError("Empty video path")
+
+    try:
+        out, _ = (
+            ffmpeg
+            .input(video_path)
+            .output('pipe:', vframes=1, format='image2', vcodec='mjpeg')
+            .run(capture_stdout=True, capture_stderr=True)
+        )
+        frame = Image.open(io.BytesIO(out)).convert("RGB")
+    except ffmpeg.Error as e:
+        raise RuntimeError(f"ffmpeg frame extract failed: {e.stderr.decode()[:2000]}")
+
+    
+    seg = AudioSegment.from_file(video_path)
+    audio16k = _audiosegment_float32(seg)
+    return frame, audio16k
+
+def load_audio_16k(audio_path_like) -> np.ndarray:
+    path = _to_path(audio_path_like)
+    seg = AudioSegment.from_file(path)
+    return _audiosegment_float32(seg)

packages.txt

@@ -0,0 +1 @@
+ffmpeg