Add Edge TPU subprocess probe to safely detect segfaults

Probes the Edge TPU in a subprocess before loading — catches segfaults
(libedgetpu ABI mismatch on Debian Trixie/Python 3.13) and falls back
to CPU automatically. No more service crashes on Coral incompatibility.

When the runtime is eventually fixed, Edge TPU will be used automatically
with no config changes needed.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alex
2026-04-11 17:40:03 -05:00
parent 43f40bf48c
commit 05409403e9
2 changed files with 40 additions and 10 deletions

View File

@@ -85,6 +85,29 @@ CATEGORY_GROUPS = {
class SoundClassifier:
@staticmethod
def _probe_edgetpu(model_path: str) -> bool:
"""Test Edge TPU in a subprocess to catch segfaults safely."""
import subprocess, sys
try:
result = subprocess.run(
[sys.executable, "-c",
"import ai_edge_litert.interpreter as tfl; "
f"d = tfl.load_delegate('libedgetpu.so.1'); "
f"i = tfl.Interpreter(model_path='{model_path}', experimental_delegates=[d]); "
"i.allocate_tensors(); "
"print('ok')"],
capture_output=True, text=True, timeout=10
)
if result.returncode == 0 and "ok" in result.stdout:
logger.info("Edge TPU probe: OK")
return True
logger.warning("Edge TPU probe failed: %s", result.stderr.strip() or f"exit {result.returncode}")
return False
except Exception as e:
logger.warning("Edge TPU probe error: %s", e)
return False
def __init__(self, model_path, class_map_path, use_edgetpu=False):
# Load class names
self._class_names = []
@@ -105,14 +128,21 @@ class SoundClassifier:
import ai_edge_litert.interpreter as tfl
if use_edgetpu:
delegate = tfl.load_delegate("libedgetpu.so.1")
self._interp = tfl.Interpreter(
model_path=str(model_path),
experimental_delegates=[delegate],
)
logger.info("YAMNet loaded on Edge TPU")
else:
self._interp = tfl.Interpreter(model_path=str(model_path))
if self._probe_edgetpu(model_path):
delegate = tfl.load_delegate("libedgetpu.so.1")
self._interp = tfl.Interpreter(
model_path=str(model_path),
experimental_delegates=[delegate],
)
logger.info("YAMNet loaded on Edge TPU")
else:
logger.warning("Edge TPU probe failed (segfault or error) — falling back to CPU")
use_edgetpu = False
if not use_edgetpu:
# Use CPU model (swap edgetpu model path for CPU model if needed)
cpu_path = str(model_path).replace("_edgetpu.tflite", ".tflite")
self._interp = tfl.Interpreter(model_path=cpu_path)
logger.info("YAMNet loaded on CPU")
self._interp.allocate_tensors()