Spaces:
Sleeping
Sleeping
File size: 8,295 Bytes
0e90a82 8f9848d 0e90a82 | 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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 | """
FST (Fusion Segment Transformer) external API client (Layer 3).
Calls the HuggingFace Space ``mippia/AI-Music-Detection-FST``
Gradio API for high-accuracy AI music detection.
FST uses MERT + beat-aware segmentation and reports 99.99%
accuracy on benchmark datasets. We treat it as a strong
external signal in the score-fusion pipeline.
Gracefully returns unavailable result on timeout or error.
"""
from __future__ import annotations
import io
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Union
from .logging_config import get_logger
logger = get_logger(__name__)
# HuggingFace Space endpoint
FST_SPACE_ID = "mippia/AI-Music-Detection-FST"
FST_API_URL = f"https://{FST_SPACE_ID.replace('/', '-')}.hf.space"
# Timeouts
FST_CONNECT_TIMEOUT = 10.0 # seconds
FST_PREDICT_TIMEOUT = 120.0 # seconds (model inference can be slow)
@dataclass
class FSTResult:
"""Result from FST external service."""
available: bool
is_ai: bool = False
confidence: float = 0.5
label: str = "unknown"
raw_scores: Optional[dict] = None
error: Optional[str] = None
class FSTClientService:
"""
Client for FST AI Music Detection HuggingFace Space.
Uses the Gradio Client API to submit audio and receive
predictions. Falls back gracefully if the space is
sleeping, overloaded, or unreachable.
"""
def __init__(self) -> None:
self._client = None
self._available: Optional[bool] = None
def _ensure_client(self) -> bool:
"""Lazy-initialize Gradio client."""
if self._available is not None:
return self._available
try:
from gradio_client import Client
self._client = Client(
FST_SPACE_ID,
hf_token=None, # Public space
)
self._available = True
logger.info(f"FST client connected: {FST_SPACE_ID}")
return True
except ImportError:
logger.warning(
"gradio_client not installed — FST layer disabled"
)
self._available = False
return False
except Exception as e:
logger.warning(f"FST client init failed: {e}")
self._available = False
return False
async def predict(
self,
source: Union[Path, bytes, io.BytesIO],
) -> FSTResult:
"""
Submit audio to FST Space for AI detection.
Args:
source: Audio file path, raw bytes, or BytesIO.
Returns:
FSTResult with detection outcome.
"""
if not self._ensure_client():
return FSTResult(
available=False,
error="fst_client_unavailable",
)
try:
audio_path = self._to_file_path(source)
return await self._call_api(audio_path)
except Exception as e:
logger.warning(f"FST prediction failed: {e}")
return FSTResult(
available=False,
error=str(e),
)
async def _call_api(self, audio_path: Path) -> FSTResult:
"""Call the FST Gradio API."""
import asyncio
try:
# Run synchronous Gradio client in executor
loop = asyncio.get_event_loop()
result = await asyncio.wait_for(
loop.run_in_executor(
None,
self._sync_predict,
audio_path,
),
timeout=FST_PREDICT_TIMEOUT,
)
return result
except asyncio.TimeoutError:
logger.warning(
f"FST prediction timed out after "
f"{FST_PREDICT_TIMEOUT}s"
)
return FSTResult(
available=False,
error="fst_timeout",
)
def _sync_predict(self, audio_path: Path) -> FSTResult:
"""Synchronous Gradio predict call."""
try:
result = self._client.predict(
str(audio_path),
api_name="/predict",
)
# Parse Gradio response
# FST typically returns label + confidence dict
return self._parse_response(result)
except Exception as e:
logger.warning(f"FST sync predict error: {e}")
return FSTResult(
available=False,
error=str(e),
)
def _parse_response(self, response: object) -> FSTResult:
"""
Parse FST Gradio API response.
FST response format varies — handle multiple formats:
1. Dict with 'label' and 'confidences'
2. String label with confidence
3. Raw dict with scores
"""
try:
if isinstance(response, dict):
return self._parse_dict_response(response)
elif isinstance(response, str):
return self._parse_string_response(response)
elif isinstance(response, (list, tuple)):
# First element is usually the classification
if len(response) > 0:
return self._parse_response(response[0])
else:
logger.warning(
f"Unexpected FST response type: "
f"{type(response)}"
)
return FSTResult(
available=True,
label="parse_error",
error=f"unexpected_type: {type(response).__name__}",
)
except Exception as e:
logger.warning(f"FST response parse error: {e}")
return FSTResult(
available=False,
error=f"parse_error: {e}",
)
def _parse_dict_response(self, data: dict) -> FSTResult:
"""Parse dict-style response."""
# Format: {"label": "AI", "confidences": [{"label": "AI", "confidence": 0.99}, ...]}
label = data.get("label", "unknown")
confidences = data.get("confidences", [])
is_ai = "ai" in label.lower() or "fake" in label.lower()
confidence = 0.5
if confidences and isinstance(confidences, list):
for item in confidences:
if isinstance(item, dict):
item_label = item.get("label", "")
if "ai" in item_label.lower() or "fake" in item_label.lower():
confidence = float(
item.get("confidence", 0.5)
)
break
# If no AI confidence found, use first confidence
if confidence == 0.5 and confidences:
first = confidences[0]
if isinstance(first, dict):
confidence = float(
first.get("confidence", 0.5)
)
if not is_ai:
confidence = 1.0 - confidence
return FSTResult(
available=True,
is_ai=is_ai,
confidence=round(
max(0.01, min(0.99, confidence)), 4
),
label=label,
raw_scores=data,
)
def _parse_string_response(self, text: str) -> FSTResult:
"""Parse string-style response."""
lower = text.lower().strip()
is_ai = any(
kw in lower
for kw in ("ai", "fake", "generated", "synthetic")
)
# Conservative confidence for string-only responses
confidence = 0.75 if is_ai else 0.25
return FSTResult(
available=True,
is_ai=is_ai,
confidence=confidence,
label=text.strip(),
)
@staticmethod
def _to_file_path(
source: Union[Path, bytes, io.BytesIO],
) -> Path:
"""Convert source to a file path for Gradio upload."""
if isinstance(source, Path):
return source
if isinstance(source, bytes):
source = io.BytesIO(source)
tmp = tempfile.NamedTemporaryFile(
suffix=".wav", delete=False,
)
tmp.write(source.read())
tmp.flush()
tmp.close()
return Path(tmp.name)
|