mirror of
https://github.com/smittix/intercept.git
synced 2026-07-07 00:58:12 -07:00
chore: commit all pending changes
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
"""Tests for pager scope waveform payload generation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import queue
|
||||
import struct
|
||||
import threading
|
||||
|
||||
from routes.pager import _encode_scope_waveform, audio_relay_thread
|
||||
|
||||
|
||||
def test_encode_scope_waveform_respects_window_and_range():
|
||||
samples = (-32768, -16384, 0, 16384, 32767)
|
||||
waveform = _encode_scope_waveform(samples, window_size=4)
|
||||
|
||||
assert len(waveform) == 4
|
||||
assert waveform[0] == -64
|
||||
assert waveform[1] == 0
|
||||
assert waveform[2] == 64
|
||||
assert waveform[3] == 127
|
||||
assert max(waveform) <= 127
|
||||
assert min(waveform) >= -127
|
||||
|
||||
|
||||
def test_audio_relay_thread_emits_scope_waveform(monkeypatch):
|
||||
base_samples = (0, 32767, -32768, 16384) * 512
|
||||
pcm = struct.pack(f"<{len(base_samples)}h", *base_samples)
|
||||
|
||||
rtl_stdout = io.BytesIO(pcm)
|
||||
multimon_stdin = io.BytesIO()
|
||||
output_queue: queue.Queue = queue.Queue()
|
||||
stop_event = threading.Event()
|
||||
|
||||
ticks = iter([0.0, 0.2, 0.2, 0.2])
|
||||
monkeypatch.setattr("routes.pager.time.monotonic", lambda: next(ticks, 0.2))
|
||||
|
||||
audio_relay_thread(rtl_stdout, multimon_stdin, output_queue, stop_event)
|
||||
|
||||
scope_event = output_queue.get_nowait()
|
||||
assert scope_event["type"] == "scope"
|
||||
assert scope_event["rms"] > 0
|
||||
assert scope_event["peak"] > 0
|
||||
assert "waveform" in scope_event
|
||||
assert len(scope_event["waveform"]) > 0
|
||||
assert max(scope_event["waveform"]) <= 127
|
||||
assert min(scope_event["waveform"]) >= -127
|
||||
@@ -0,0 +1,21 @@
|
||||
"""Tests for synthesized 433 MHz scope waveform payload."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from routes.sensor import _build_scope_waveform
|
||||
|
||||
|
||||
def test_build_scope_waveform_has_expected_shape_and_bounds():
|
||||
waveform = _build_scope_waveform(rssi=-8.5, snr=11.2, noise=-26.0, points=96)
|
||||
|
||||
assert len(waveform) == 96
|
||||
assert max(waveform) <= 127
|
||||
assert min(waveform) >= -127
|
||||
assert any(sample != 0 for sample in waveform)
|
||||
|
||||
|
||||
def test_build_scope_waveform_changes_with_signal_profile():
|
||||
low_snr = _build_scope_waveform(rssi=-14.0, snr=2.0, noise=-12.0, points=64)
|
||||
high_snr = _build_scope_waveform(rssi=-14.0, snr=20.0, noise=-12.0, points=64)
|
||||
|
||||
assert low_snr != high_snr
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Tests for SSTV scope waveform encoding."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
|
||||
from utils.sstv.sstv_decoder import _encode_scope_waveform
|
||||
|
||||
|
||||
def test_encode_scope_waveform_respects_window_and_bounds():
|
||||
samples = np.array([-32768, -16384, 0, 16384, 32767], dtype=np.int16)
|
||||
waveform = _encode_scope_waveform(samples, window_size=4)
|
||||
|
||||
assert len(waveform) == 4
|
||||
assert waveform[0] == -64
|
||||
assert waveform[1] == 0
|
||||
assert waveform[2] == 64
|
||||
assert waveform[3] == 127
|
||||
assert max(waveform) <= 127
|
||||
assert min(waveform) >= -127
|
||||
|
||||
|
||||
def test_encode_scope_waveform_empty_input():
|
||||
waveform = _encode_scope_waveform(np.array([], dtype=np.int16))
|
||||
assert waveform == []
|
||||
Reference in New Issue
Block a user