964 lines
34 KiB
Python
964 lines
34 KiB
Python
"""Tests for the combine command."""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
from ria_toolkit_oss_cli.cli import cli
|
|
|
|
from ria_toolkit_oss.datatypes import Annotation, Recording
|
|
from ria_toolkit_oss.io import load_recording, to_npy, to_sigmf
|
|
|
|
|
|
class TestCombineHelp:
|
|
"""Test help and basic command structure."""
|
|
|
|
def test_help(self):
|
|
"""Test combine help."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["combine", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "Combine multiple recordings" in result.output
|
|
assert "--mode" in result.output
|
|
assert "--align-mode" in result.output
|
|
|
|
def test_no_inputs(self):
|
|
"""Test error with no inputs."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["combine"])
|
|
assert result.exit_code != 0
|
|
|
|
def test_single_input(self):
|
|
"""Test error with only one input."""
|
|
runner = CliRunner()
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create test file
|
|
signal = np.arange(1000, dtype=np.complex64)
|
|
recording = Recording(data=signal, metadata={"sample_rate": 2e6})
|
|
to_npy(recording, filename=str(Path(tmpdir) / "test.npy"), overwrite=True)
|
|
|
|
result = runner.invoke(cli, ["combine", str(Path(tmpdir) / "test.npy"), str(Path(tmpdir) / "output.npy")])
|
|
assert result.exit_code != 0
|
|
|
|
|
|
class TestCombineConcat:
|
|
"""Test concatenate mode."""
|
|
|
|
@pytest.fixture
|
|
def test_recordings(self):
|
|
"""Create multiple test recording files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create 3 recordings with different data
|
|
for i in range(3):
|
|
signal = np.arange(i * 1000, (i + 1) * 1000, dtype=np.complex64)
|
|
recording = Recording(data=signal, metadata={"sample_rate": 2e6})
|
|
to_npy(recording, filename=str(Path(tmpdir) / f"chunk{i}.npy"), overwrite=True)
|
|
yield tmpdir
|
|
|
|
def test_concat_basic(self, test_recordings):
|
|
"""Test basic concatenation."""
|
|
runner = CliRunner()
|
|
tmpdir = test_recordings
|
|
output_path = str(Path(tmpdir) / "combined.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "chunk0.npy"),
|
|
str(Path(tmpdir) / "chunk1.npy"),
|
|
str(Path(tmpdir) / "chunk2.npy"),
|
|
output_path,
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert Path(output_path).exists()
|
|
|
|
# Verify result
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 3000
|
|
assert combined._metadata["combine_mode"] == "concat"
|
|
assert combined._metadata["num_inputs"] == 3
|
|
|
|
# Check data is correctly concatenated
|
|
assert np.allclose(combined.data[0, :1000], np.arange(0, 1000))
|
|
assert np.allclose(combined.data[0, 1000:2000], np.arange(1000, 2000))
|
|
assert np.allclose(combined.data[0, 2000:3000], np.arange(2000, 3000))
|
|
|
|
def test_concat_verbose(self, test_recordings):
|
|
"""Test concatenation with verbose output."""
|
|
runner = CliRunner()
|
|
tmpdir = test_recordings
|
|
output_path = str(Path(tmpdir) / "combined.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "chunk0.npy"),
|
|
str(Path(tmpdir) / "chunk1.npy"),
|
|
str(Path(tmpdir) / "chunk2.npy"),
|
|
output_path,
|
|
"--verbose",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "Combining 3 recordings" in result.output
|
|
assert "concat mode" in result.output
|
|
assert "Concatenating..." in result.output
|
|
|
|
def test_concat_with_annotations(self):
|
|
"""Test that annotations are preserved and shifted in concat mode."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create recordings with annotations
|
|
rec1 = Recording(data=np.ones(1000, dtype=np.complex64), metadata={"sample_rate": 2e6})
|
|
rec1._annotations.append(
|
|
Annotation(
|
|
sample_start=100, sample_count=200, freq_lower_edge=900e6, freq_upper_edge=920e6, label="test1"
|
|
)
|
|
)
|
|
|
|
rec2 = Recording(data=np.ones(1000, dtype=np.complex64) * 2, metadata={"sample_rate": 2e6})
|
|
rec2._annotations.append(
|
|
Annotation(
|
|
sample_start=100, sample_count=200, freq_lower_edge=900e6, freq_upper_edge=920e6, label="test2"
|
|
)
|
|
)
|
|
|
|
to_sigmf(rec1, filename="rec1", path=tmpdir, overwrite=True)
|
|
to_sigmf(rec2, filename="rec2", path=tmpdir, overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "combined.sigmf-data")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
["combine", str(Path(tmpdir) / "rec1.sigmf-data"), str(Path(tmpdir) / "rec2.sigmf-data"), output_path],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert len(combined._annotations) == 2
|
|
# First annotation unchanged
|
|
assert combined._annotations[0].sample_start == 100
|
|
assert combined._annotations[0].label == "test1"
|
|
# Second annotation shifted by 1000 samples
|
|
assert combined._annotations[1].sample_start == 1100
|
|
assert combined._annotations[1].label == "test2"
|
|
|
|
|
|
class TestCombineAddSameLength:
|
|
"""Test add mode with same-length recordings."""
|
|
|
|
def test_add_basic(self):
|
|
"""Test basic add with same-length recordings."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create two recordings with same length
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "added.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
# Verify result
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 1000
|
|
assert np.allclose(combined.data, 3 + 0j)
|
|
assert combined._metadata["combine_mode"] == "add"
|
|
assert combined._metadata["align_mode"] == "error"
|
|
|
|
def test_add_three_recordings(self):
|
|
"""Test adding three same-length recordings."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create three recordings
|
|
for i in range(1, 4):
|
|
sig = np.ones(1000, dtype=np.complex64) * i
|
|
rec = Recording(data=sig, metadata={"sample_rate": 2e6})
|
|
to_npy(rec, filename=str(Path(tmpdir) / f"sig{i}.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "added.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
str(Path(tmpdir) / "sig3.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
# 1 + 2 + 3 = 6
|
|
assert np.allclose(combined.data, 6 + 0j)
|
|
|
|
|
|
class TestCombineAddAlignError:
|
|
"""Test add mode with error alignment (default)."""
|
|
|
|
def test_different_length_error(self):
|
|
"""Test that different lengths cause error by default."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create recordings with different lengths
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "different lengths" in result.output
|
|
assert "--align-mode" in result.output
|
|
|
|
|
|
class TestCombineAddAlignTruncate:
|
|
"""Test add mode with truncate alignment."""
|
|
|
|
def test_truncate(self):
|
|
"""Test truncate to shortest recording."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "truncated.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"truncate",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 5000
|
|
assert np.allclose(combined.data, 3 + 0j)
|
|
|
|
|
|
class TestCombineAddAlignPad:
|
|
"""Test add mode with pad alignment."""
|
|
|
|
def test_pad(self):
|
|
"""Test zero-padding to longest recording."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "padded.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"pad",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 10000
|
|
# First 5000: 1 + 2 = 3
|
|
assert np.allclose(combined.data[0, :5000], 3 + 0j)
|
|
# Last 5000: 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, 5000:], 1 + 0j)
|
|
|
|
|
|
class TestCombineAddAlignPadStart:
|
|
"""Test add mode with pad-start alignment."""
|
|
|
|
def test_pad_start(self):
|
|
"""Test pad-start at specific sample."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "pad_start.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"pad-start",
|
|
"--pad-start-sample",
|
|
"3000",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 10000
|
|
# Before 3000: 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, :3000], 1 + 0j)
|
|
# 3000-8000: 1 + 2 = 3
|
|
assert np.allclose(combined.data[0, 3000:8000], 3 + 0j)
|
|
# After 8000: 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, 8000:], 1 + 0j)
|
|
|
|
def test_pad_start_invalid(self):
|
|
"""Test invalid pad-start-sample."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"pad-start",
|
|
"--pad-start-sample",
|
|
"7000", # Too large
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "exceeds max length" in result.output
|
|
|
|
|
|
class TestCombineAddAlignPadCenter:
|
|
"""Test add mode with pad-center alignment."""
|
|
|
|
def test_pad_center(self):
|
|
"""Test centering shorter recording."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "pad_center.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"pad-center",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 10000
|
|
# Before 2500: 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, :2500], 1 + 0j)
|
|
# 2500-7500: 1 + 2 = 3
|
|
assert np.allclose(combined.data[0, 2500:7500], 3 + 0j)
|
|
# After 7500: 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, 7500:], 1 + 0j)
|
|
|
|
|
|
class TestCombineAddAlignPadEnd:
|
|
"""Test add mode with pad-end alignment."""
|
|
|
|
def test_pad_end(self):
|
|
"""Test aligning end of recordings."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "pad_end.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"pad-end",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 10000
|
|
# First 5000: 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, :5000], 1 + 0j)
|
|
# Last 5000: 1 + 2 = 3
|
|
assert np.allclose(combined.data[0, 5000:], 3 + 0j)
|
|
|
|
|
|
class TestCombineAddAlignRepeat:
|
|
"""Test add mode with repeat alignment."""
|
|
|
|
def test_repeat(self):
|
|
"""Test repeating shorter recording."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "repeated.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"repeat",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 10000
|
|
# Entire recording: 1 + 2 = 3 (pattern repeated)
|
|
assert np.allclose(combined.data, 3 + 0j)
|
|
|
|
def test_repeat_partial(self):
|
|
"""Test repeat with non-exact multiple."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.arange(3000, dtype=np.complex64)
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "repeated.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"repeat",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
# Check pattern repeats correctly
|
|
# First 3000: 1 + [0,1,2,...,2999]
|
|
assert np.allclose(combined.data[0, :3000], 1 + np.arange(3000))
|
|
# Next 3000: 1 + [0,1,2,...,2999]
|
|
assert np.allclose(combined.data[0, 3000:6000], 1 + np.arange(3000))
|
|
# Next 3000: 1 + [0,1,2,...,2999]
|
|
assert np.allclose(combined.data[0, 6000:9000], 1 + np.arange(3000))
|
|
# Last 1000: 1 + [0,1,2,...,999]
|
|
assert np.allclose(combined.data[0, 9000:10000], 1 + np.arange(1000))
|
|
|
|
|
|
class TestCombineAddAlignRepeatSpaced:
|
|
"""Test add mode with repeat-spaced alignment."""
|
|
|
|
def test_repeat_spaced(self):
|
|
"""Test repeating with spacing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(2000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "repeat_spaced.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"repeat-spaced",
|
|
"--repeat-spacing",
|
|
"1000",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined.data.shape[1] == 10000
|
|
|
|
# First 2000: 1 + 2 = 3
|
|
assert np.allclose(combined.data[0, :2000], 3 + 0j)
|
|
# Next 1000 (gap): 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, 2000:3000], 1 + 0j)
|
|
# Next 2000: 1 + 2 = 3
|
|
assert np.allclose(combined.data[0, 3000:5000], 3 + 0j)
|
|
# Next 1000 (gap): 1 + 0 = 1
|
|
assert np.allclose(combined.data[0, 5000:6000], 1 + 0j)
|
|
|
|
def test_repeat_spaced_missing_spacing(self):
|
|
"""Test error when spacing not provided."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(10000, dtype=np.complex64)
|
|
sig2 = np.ones(5000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "long.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "short.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "long.npy"),
|
|
str(Path(tmpdir) / "short.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--align-mode",
|
|
"repeat-spaced",
|
|
# Missing --repeat-spacing
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "requires --repeat-spacing" in result.output
|
|
|
|
|
|
class TestCombineValidation:
|
|
"""Test validation and error handling."""
|
|
|
|
def test_sample_rate_mismatch(self):
|
|
"""Test error on sample rate mismatch in add mode."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 1e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "different sample rates" in result.output
|
|
|
|
def test_channel_count_mismatch(self):
|
|
"""Test error on channel count mismatch."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Single channel
|
|
sig1 = np.ones((1, 1000), dtype=np.complex64)
|
|
# Two channels
|
|
sig2 = np.ones((2, 1000), dtype=np.complex64)
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "different channel counts" in result.output
|
|
|
|
def test_overwrite_protection(self):
|
|
"""Test overwrite protection."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create test recordings
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
# Create existing output file
|
|
existing = Recording(data=np.zeros(100, dtype=np.complex64), metadata={})
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
to_npy(existing, filename=output_path, overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
|
|
# Should fail without --overwrite
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "already exists" in result.output
|
|
|
|
# Should succeed with --overwrite
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--overwrite",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestCombineOutputOptions:
|
|
"""Test output format and options."""
|
|
|
|
def test_output_formats(self):
|
|
"""Test different output formats."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create test recordings
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
|
|
# Test SigMF output
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
str(Path(tmpdir) / "output.sigmf-data"),
|
|
"--mode",
|
|
"add",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert Path(tmpdir, "output.sigmf-data").exists()
|
|
assert Path(tmpdir, "output.sigmf-meta").exists()
|
|
|
|
def test_normalize(self):
|
|
"""Test normalize option."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(1000, dtype=np.complex64) * 10
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 20
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "normalized.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--normalize",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
# Should be normalized to max magnitude 1
|
|
assert np.allclose(np.max(np.abs(combined.data)), 1.0)
|
|
assert combined._metadata.get("normalized") is True
|
|
|
|
def test_custom_metadata(self):
|
|
"""Test adding custom metadata."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--metadata",
|
|
"test_id=test123",
|
|
"--metadata",
|
|
"author=tester",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
|
|
combined = load_recording(output_path)
|
|
assert combined._metadata["test_id"] == "test123"
|
|
assert combined._metadata["author"] == "tester"
|
|
|
|
|
|
class TestCombineVerboseQuiet:
|
|
"""Test verbose and quiet modes."""
|
|
|
|
def test_verbose(self):
|
|
"""Test verbose output."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--verbose",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "Loading" in result.output
|
|
assert "Done" in result.output
|
|
|
|
def test_quiet(self):
|
|
"""Test quiet output."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
sig1 = np.ones(1000, dtype=np.complex64)
|
|
sig2 = np.ones(1000, dtype=np.complex64) * 2
|
|
|
|
rec1 = Recording(data=sig1, metadata={"sample_rate": 2e6})
|
|
rec2 = Recording(data=sig2, metadata={"sample_rate": 2e6})
|
|
|
|
to_npy(rec1, filename=str(Path(tmpdir) / "sig1.npy"), overwrite=True)
|
|
to_npy(rec2, filename=str(Path(tmpdir) / "sig2.npy"), overwrite=True)
|
|
|
|
runner = CliRunner()
|
|
output_path = str(Path(tmpdir) / "output.npy")
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"combine",
|
|
str(Path(tmpdir) / "sig1.npy"),
|
|
str(Path(tmpdir) / "sig2.npy"),
|
|
output_path,
|
|
"--mode",
|
|
"add",
|
|
"--quiet",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert result.output == ""
|