qoherent/ria-toolkit-oss
19
6
Fork 0
Code Issues Pull Requests 1 Workflows Packages Projects Releases 6 Wiki Activity
ria-toolkit-oss/tests/ria_toolkit_oss_cli/test.combine.py

964 lines
34 KiB
Python
Raw Normal View History

G
added tests for cli
2025-12-11 15:59:08 -05:00
"""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.datatypes import Annotation, Recording
from ria_toolkit_oss.io import load_recording, to_npy, to_sigmf
M
Formatting
2025-12-19 11:25:06 -05:00
from ria_toolkit_oss_cli.cli import cli
G
added tests for cli
2025-12-11 15:59:08 -05:00
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 == ""
Reference in New Issue Copy Permalink