Spaces:
Running on Zero
Running on Zero
File size: 2,730 Bytes
031e3f9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | from __future__ import annotations
import io
import logging
import os
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import Any, TypeVar
import numpy as np
from PIL import Image
LOGGER = logging.getLogger(__name__)
T = TypeVar("T")
def image_to_bytes(image: Image.Image | np.ndarray | str | Path) -> bytes:
if isinstance(image, (str, Path)):
image = Image.open(image)
elif isinstance(image, np.ndarray):
image = Image.fromarray(image)
if not isinstance(image, Image.Image):
raise TypeError("Expected a PIL image, NumPy array, or image path.")
buffer = io.BytesIO()
image.convert("RGB").save(buffer, format="JPEG", quality=92)
return buffer.getvalue()
WAV_PREFIX = "thirdeye_"
def bytes_to_wav(audio_bytes: bytes) -> str:
if not audio_bytes:
raise ValueError("Audio data is empty.")
output = tempfile.NamedTemporaryFile(delete=False, prefix=WAV_PREFIX, suffix=".wav")
output.write(audio_bytes)
output.close()
return output.name
def cleanup_wav(path: str | None) -> None:
if path and os.path.exists(path):
try:
os.unlink(path)
except OSError:
pass
def prune_old_wavs(max_age_seconds: float = 600.0) -> int:
"""Best-effort cleanup of stale Third Eye temp WAVs.
Only removes files older than ``max_age_seconds`` so a clip currently being
served to the browser is never deleted. Returns the number of files removed.
"""
import time
removed = 0
temp_dir = Path(tempfile.gettempdir())
cutoff = time.time() - max_age_seconds
try:
candidates = temp_dir.glob(f"{WAV_PREFIX}*.wav")
except OSError:
return 0
for candidate in candidates:
try:
if candidate.stat().st_mtime < cutoff:
candidate.unlink()
removed += 1
except OSError:
continue
return removed
def read_audio_bytes(audio_path: str | Path) -> bytes:
return Path(audio_path).read_bytes()
def safe_call(
fn: Callable[..., T],
*args: Any,
fallback: T | None = None,
warn: str = "Something went wrong.",
**kwargs: Any,
) -> T | None:
try:
return fn(*args, **kwargs)
except Exception as exc:
LOGGER.exception("Third Eye stage failed: %s", getattr(fn, "__name__", fn))
detail = f"{warn} ({type(exc).__name__}: {exc})"
try:
import gradio as gr
gr.Warning(detail if len(detail) < 200 else warn)
except Exception:
LOGGER.warning(detail)
return fallback
|