thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from app.bpm.service import BeatAnalyzer, BpmService
|
|
|
|
|
|
def test_beat_analyzer_estimates_click_track_bpm() -> None:
|
|
analyzer = BeatAnalyzer(sample_rate=44_100, window_size=1024)
|
|
click = (b"\x00\x40" * 2048) + (b"\x00\x00" * (44_100 // 2 - 2048))
|
|
|
|
result = None
|
|
for _ in range(6):
|
|
result = analyzer.feed_pcm16(click)
|
|
|
|
assert result is not None
|
|
bpm, confidence = result
|
|
assert 117 <= bpm <= 124
|
|
assert confidence > 0.5
|
|
|
|
|
|
def test_bpm_service_synthetic_audio_updates_snapshot() -> None:
|
|
async def run_test() -> None:
|
|
service = BpmService()
|
|
await service.start_audio("synthetic-click-track")
|
|
await asyncio.sleep(1.2)
|
|
snapshot = service.snapshot()
|
|
await service.stop_audio()
|
|
|
|
assert snapshot["mode"] == "audio:synthetic-click-track"
|
|
assert snapshot["audio_connected"] is True
|
|
assert 117 <= float(snapshot["bpm"]) <= 123
|
|
assert float(snapshot["confidence"]) > 0.3
|
|
assert float(snapshot["input_level"]) > 0.0
|
|
assert float(snapshot["peak_level"]) >= float(snapshot["input_level"])
|
|
assert snapshot["clipping"] is False
|
|
|
|
asyncio.run(run_test())
|
|
|
|
|
|
def test_parse_alsa_card_line_supports_named_aliases() -> None:
|
|
service = BpmService()
|
|
|
|
card = service._parse_card_line("card 1: SB [HDA ATI SB], device 0: ALC883 Analog [ALC883 Analog]")
|
|
device = service._parse_device_from_card_line(
|
|
"card 1: SB [HDA ATI SB], device 0: ALC883 Analog [ALC883 Analog]"
|
|
)
|
|
|
|
assert card == ("1", "SB", "HDA ATI SB")
|
|
assert device == ("0", "ALC883 Analog", "ALC883 Analog")
|
|
|
|
|
|
def test_busy_alsa_error_is_humanized() -> None:
|
|
service = BpmService()
|
|
|
|
message = service._humanize_alsa_error(
|
|
"alsa:plughw:CARD=SB,DEV=0",
|
|
"arecord: main:831: audio open error: Device or resource busy",
|
|
)
|
|
|
|
assert "Mikrofonen er optaget" in message
|