f-denkena/livekit-cameras
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: smoke worker with speechbrain force on local livekit

Browse Source
This commit is contained in:
Hermes
2026-08-28 05:19:05 +00:00
parent ad3b43eccd
commit 106790cab8
1 changed files with 18 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
smoke_test.py
+18 -8
View File
@@ -12,18 +12,23 @@ sys.path.insert(0, str(HERE))
from discovery import discover_cameras from discovery import discover_cameras
from tokens import make_token from tokens import make_token
cam = discover_cameras()[0] cams = discover_cameras()
tok = make_token("ws://localhost:7880", "devkey", "devsecret", "smoke", "smoke-test") if not cams:
raise SystemExit("no cameras discovered")
cam = cams[0]
url = "ws://127.0.0.1:7880"
tok = make_token(url, "devkey", "devsecret", "smoke", cam.identity)
p = subprocess.Popen( p = subprocess.Popen(
[str(HERE / ".venv/bin/python"), str(HERE / "run_worker.py"), [str(HERE / ".venv/bin/python"), str(HERE / "run_worker.py"),
"--camera-json", json.dumps({ "--camera-json", json.dumps({
"index": 0, "identity": "smoke-test", "index": cam.index, "identity": cam.identity,
"video_device": cam.video_device, "audio_card": cam.audio_card, "video_device": cam.video_device, "audio_card": cam.audio_card,
"usb_port": cam.usb_port}), "usb_port": cam.usb_port, "serial": cam.serial}),
"--room", "smoke", "--url", "ws://localhost:7880", "--token", tok, "--room", "smoke", "--url", url, "--token", tok,
"--width", "640", "--height", "480", "--fps", "30", "--width", "640", "--height", "480", "--fps", "15",
"--audio-rate", "16000", "--enhance-mode", "off"], "--audio-rate", "16000", "--enhance-mode", "force",
"--enhance-chunk-s", "1.0", "--enhance-hop-s", "0.2"],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
) )
t0 = time.time() t0 = time.time()
@@ -40,4 +45,9 @@ except subprocess.TimeoutExpired:
p.kill() p.kill()
rc = "killed" rc = "killed"
print(f"--- smoke test done: rc={rc}") print(f"--- smoke test done: rc={rc}")
print("published markers:", [l for l in lines if "published" in l or "status" in l or "UP" in l]) print("published markers:",
[l for l in lines if "published" in l or "status" in l or "UP" in l
or "backend" in l])
ok = any("audio backend: model" in l for l in lines) and any(
"published video+audio" in l or "status: UP" in l for l in lines)
sys.exit(0 if ok else 1)