1503 lines
46 KiB
Python
1503 lines
46 KiB
Python
|
G
|
"""Tests for generate/synth command.
|
||
|
|
|
||
|
|
This test suite covers the `ria generate` and `ria synth` (alias) commands for
|
||
|
|
generating synthetic RF signals. Tests are designed to work with the current
|
||
|
|
implementation status of the generate command.
|
||
|
|
|
||
|
|
Note: Some impairment parameters that appear in common_options are not yet
|
||
|
|
fully implemented in individual command function signatures. Tests have been
|
||
|
|
designed to work with parameters that are currently supported.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import tempfile
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from click.testing import CliRunner
|
||
|
|
|
||
|
|
from ria_toolkit_oss.ria_toolkit_oss_cli.cli import cli
|
||
|
|
|
||
|
|
|
||
|
|
class TestGenerateCommandBasics:
|
||
|
|
"""Test basic generate command functionality."""
|
||
|
|
|
||
|
|
def test_generate_help(self):
|
||
|
|
"""Test generate command help."""
|
||
|
|
runner = CliRunner()
|
||
|
|
result = runner.invoke(cli, ["generate", "--help"])
|
||
|
|
assert result.exit_code == 0
|
||
|
|
assert "generate" in result.output.lower() or "Generate signal" in result.output
|
||
|
|
# Check for some key subcommands
|
||
|
|
subcommands = ["chirp", "fsk", "gmsk", "noise", "psk", "qam", "tone"]
|
||
|
|
for cmd in subcommands:
|
||
|
|
assert cmd in result.output
|
||
|
|
|
||
|
|
def test_synth_alias_help(self):
|
||
|
|
"""Test synth alias for generate command."""
|
||
|
|
runner = CliRunner()
|
||
|
|
result = runner.invoke(cli, ["synth", "--help"])
|
||
|
|
assert result.exit_code == 0
|
||
|
|
assert "synth" in result.output.lower() or "Generate signal" in result.output
|
||
|
|
|
||
|
|
def test_missing_sample_rate(self):
|
||
|
|
"""Test that sample rate is required."""
|
||
|
|
runner = CliRunner()
|
||
|
|
result = runner.invoke(cli, ["generate", "tone", "-n", "1000", "-o", "/tmp/test.sigmf"])
|
||
|
|
assert result.exit_code != 0
|
||
|
|
# Should fail due to missing sample-rate
|
||
|
|
|
||
|
|
|
||
|
|
class TestToneCommand:
|
||
|
|
"""Test tone (CW) signal generation."""
|
||
|
|
|
||
|
|
def test_tone_basic(self):
|
||
|
|
"""Test basic tone generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "1e6", "--num-samples", "10000", "--output", output, "-q"]
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
# Check that output files were created
|
||
|
|
assert (
|
||
|
|
Path(output.replace(".sigmf", ".sigmf-data")).exists()
|
||
|
|
or Path(output.replace(".sigmf", "") + ".sigmf-data").exists()
|
||
|
|
or Path(output).exists()
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_tone_with_frequency(self):
|
||
|
|
"""Test tone with custom frequency."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_freq.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--frequency",
|
||
|
|
"100000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_tone_with_amplitude(self):
|
||
|
|
"""Test tone with custom amplitude."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_amp.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--amplitude",
|
||
|
|
"0.5",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_tone_duration_instead_of_samples(self):
|
||
|
|
"""Test tone using duration instead of num-samples."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_duration.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "1e6", "--duration", "0.01", "--output", output, "-q"]
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_tone_with_phase(self):
|
||
|
|
"""Test tone with phase offset."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_phase.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--phase",
|
||
|
|
"1.57", # pi/2
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_tone_with_center_frequency(self):
|
||
|
|
"""Test tone with center frequency metadata."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_cf.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--center-frequency",
|
||
|
|
"915e6",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestNoiseCommand:
|
||
|
|
"""Test noise signal generation."""
|
||
|
|
|
||
|
|
def test_noise_gaussian(self):
|
||
|
|
"""Test Gaussian noise generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "noise_gauss.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"noise",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--noise-type",
|
||
|
|
"gaussian",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_noise_uniform(self):
|
||
|
|
"""Test uniform noise generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "noise_uniform.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"noise",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--noise-type",
|
||
|
|
"uniform",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_noise_with_power(self):
|
||
|
|
"""Test noise with custom power."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "noise_power.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"noise",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--power",
|
||
|
|
"0.5",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestChirpCommand:
|
||
|
|
"""Test chirp/LFM signal generation."""
|
||
|
|
|
||
|
|
def test_chirp_up(self):
|
||
|
|
"""Test upward chirp generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "chirp_up.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"chirp",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--bandwidth",
|
||
|
|
"100000",
|
||
|
|
"--period",
|
||
|
|
"0.01",
|
||
|
|
"--type",
|
||
|
|
"up",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_chirp_down(self):
|
||
|
|
"""Test downward chirp generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "chirp_down.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"chirp",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--bandwidth",
|
||
|
|
"100000",
|
||
|
|
"--period",
|
||
|
|
"0.01",
|
||
|
|
"--type",
|
||
|
|
"down",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_chirp_up_down(self):
|
||
|
|
"""Test up-down chirp generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "chirp_updown.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"chirp",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--bandwidth",
|
||
|
|
"100000",
|
||
|
|
"--period",
|
||
|
|
"0.01",
|
||
|
|
"--type",
|
||
|
|
"up_down",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestWaveformCommands:
|
||
|
|
"""Test square and sawtooth waveforms."""
|
||
|
|
|
||
|
|
def test_square_basic(self):
|
||
|
|
"""Test square wave generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "square.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"square",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--frequency",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_square_duty_cycle(self):
|
||
|
|
"""Test square wave with custom duty cycle."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "square_duty.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"square",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--frequency",
|
||
|
|
"10000",
|
||
|
|
"--duty-cycle",
|
||
|
|
"0.25",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_sawtooth_basic(self):
|
||
|
|
"""Test sawtooth wave generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "sawtooth.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"sawtooth",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--frequency",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestQAMCommand:
|
||
|
|
"""Test QAM (Quadrature Amplitude Modulation) generation."""
|
||
|
|
|
||
|
|
def test_qam16(self):
|
||
|
|
"""Test 16-QAM generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam16.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam64(self):
|
||
|
|
"""Test 64-QAM generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam64.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"64",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam256(self):
|
||
|
|
"""Test 256-QAM generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam256.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"256",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam_with_filter(self):
|
||
|
|
"""Test QAM with pulse shaping filter."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam_rrc.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--filter",
|
||
|
|
"rrc",
|
||
|
|
"--filter-beta",
|
||
|
|
"0.35",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam_symbols_instead_of_samples(self):
|
||
|
|
"""Test QAM using symbol count instead of sample count."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam_symbols.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--symbols",
|
||
|
|
"100",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam_with_gaussian_filter(self):
|
||
|
|
"""Test QAM with Gaussian filter."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam_gauss.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--filter",
|
||
|
|
"gaussian",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 1
|
||
|
|
assert isinstance(result.exception, SystemExit)
|
||
|
|
|
||
|
|
|
||
|
|
class TestAPSKCommand:
|
||
|
|
"""Test APSK (Amplitude Phase Shift Keying) generation."""
|
||
|
|
|
||
|
|
def test_apsk16(self):
|
||
|
|
"""Test 16-APSK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "apsk16.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"apsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_apsk32(self):
|
||
|
|
"""Test 32-APSK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "apsk32.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"apsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"32",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_apsk_with_rrc_filter(self):
|
||
|
|
"""Test APSK with RRC filter."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "apsk_rrc.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"apsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"32",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--filter",
|
||
|
|
"rrc",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestPAMCommand:
|
||
|
|
"""Test PAM (Pulse Amplitude Modulation) generation."""
|
||
|
|
|
||
|
|
def test_pam4(self):
|
||
|
|
"""Test 4-PAM generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "pam4.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"pam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"4",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_pam16(self):
|
||
|
|
"""Test 16-PAM generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "pam16.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"pam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestPSKCommand:
|
||
|
|
"""Test PSK (Phase Shift Keying) generation."""
|
||
|
|
|
||
|
|
def test_bpsk(self):
|
||
|
|
"""Test BPSK (2-PSK) generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "bpsk.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"psk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"2",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qpsk(self):
|
||
|
|
"""Test QPSK (4-PSK) generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qpsk.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"psk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"4",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_8psk(self):
|
||
|
|
"""Test 8-PSK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "8psk.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"psk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"8",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestFSKCommand:
|
||
|
|
"""Test FSK (Frequency Shift Keying) generation."""
|
||
|
|
|
||
|
|
def test_fsk2(self):
|
||
|
|
"""Test 2-FSK (binary FSK) generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "fsk2.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"fsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"2",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--freq-spacing",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_fsk4(self):
|
||
|
|
"""Test 4-FSK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "fsk4.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"fsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"4",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--freq-spacing",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_fsk_with_modulation_index(self):
|
||
|
|
"""Test FSK with modulation index instead of frequency spacing."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "fsk_mi.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"fsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"2",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--modulation-index",
|
||
|
|
"5.0",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestGMSKCommand:
|
||
|
|
"""Test GMSK (Gaussian Minimum Shift Keying) generation."""
|
||
|
|
|
||
|
|
def test_gmsk_basic(self):
|
||
|
|
"""Test basic GMSK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "gmsk.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"gmsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--symbol-rate",
|
||
|
|
"270833",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_gmsk_custom_bt(self):
|
||
|
|
"""Test GMSK with custom BT product."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "gmsk_bt.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"gmsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--symbol-rate",
|
||
|
|
"270833",
|
||
|
|
"--bt",
|
||
|
|
"0.5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestOOKCommand:
|
||
|
|
"""Test OOK (On-Off Keying) generation."""
|
||
|
|
|
||
|
|
def test_ook_basic(self):
|
||
|
|
"""Test OOK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "ook.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"ook",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestOQPSKCommand:
|
||
|
|
"""Test OQPSK (Offset QPSK) generation."""
|
||
|
|
|
||
|
|
def test_oqpsk_basic(self):
|
||
|
|
"""Test OQPSK generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "oqpsk.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"oqpsk",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestNR5GCommand:
|
||
|
|
"""Test NR 5G frame generation."""
|
||
|
|
|
||
|
|
@pytest.mark.skip(reason="NR5G generation may not be available in all configurations")
|
||
|
|
def test_nr5g_basic(self):
|
||
|
|
"""Test basic NR 5G frame generation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "nr5g.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"nr5g",
|
||
|
|
"--sample-rate",
|
||
|
|
"30.72e6",
|
||
|
|
"--bandwidth",
|
||
|
|
"20",
|
||
|
|
"--mu",
|
||
|
|
"1",
|
||
|
|
"--num-samples",
|
||
|
|
"30720",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
# NR5G may not be available, check accordingly
|
||
|
|
if result.exit_code != 0 and "not available" in result.output.lower():
|
||
|
|
pytest.skip("NR5G not available")
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestOutputFormats:
|
||
|
|
"""Test different output formats."""
|
||
|
|
|
||
|
|
def test_output_npy(self):
|
||
|
|
"""Test saving as NPY format."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "signal.npy")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--format",
|
||
|
|
"npy",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_output_wav(self):
|
||
|
|
"""Test saving as WAV format."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "signal.wav")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--format",
|
||
|
|
"wav",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_output_blue(self):
|
||
|
|
"""Test saving as BLUE (Midas) format."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "signal.blue")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--format",
|
||
|
|
"blue",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_format_detection_from_extension(self):
|
||
|
|
"""Test that format is detected from file extension."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
# .npy extension should use NPY format
|
||
|
|
output = os.path.join(tmpdir, "signal.npy")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "1e6", "--num-samples", "1000", "--output", output, "-q"]
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestChannelModels:
|
||
|
|
"""Test channel models that are currently implemented."""
|
||
|
|
|
||
|
|
def test_frequency_shift(self):
|
||
|
|
"""Test frequency shift application."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_shifted.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--frequency-shift",
|
||
|
|
"10000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_awgn_channel(self):
|
||
|
|
"""Test AWGN channel."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_awgn.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--channel-type",
|
||
|
|
"awgn",
|
||
|
|
"--noise-power",
|
||
|
|
"0.1",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
# May not be fully implemented yet
|
||
|
|
if result.exit_code == 0:
|
||
|
|
pass # Test passes
|
||
|
|
else:
|
||
|
|
# Document if AWGN not implemented
|
||
|
|
pytest.skip("AWGN channel not yet implemented")
|
||
|
|
|
||
|
|
def test_rayleigh_channel(self):
|
||
|
|
"""Test Rayleigh fading channel."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_rayleigh.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--channel-type",
|
||
|
|
"rayleigh",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
# May not be fully implemented yet
|
||
|
|
if result.exit_code == 0:
|
||
|
|
pass # Test passes
|
||
|
|
else:
|
||
|
|
pytest.skip("Rayleigh channel not yet implemented")
|
||
|
|
|
||
|
|
def test_rician_channel(self):
|
||
|
|
"""Test Rician fading channel."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_rician.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--channel-type",
|
||
|
|
"rician",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
# May not be fully implemented yet
|
||
|
|
if result.exit_code == 0:
|
||
|
|
pass # Test passes
|
||
|
|
else:
|
||
|
|
pytest.skip("Rician channel not yet implemented")
|
||
|
|
|
||
|
|
|
||
|
|
class TestMetadataAndConfig:
|
||
|
|
"""Test metadata and configuration options."""
|
||
|
|
|
||
|
|
def test_custom_metadata(self):
|
||
|
|
"""Test adding custom metadata."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_meta.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--metadata",
|
||
|
|
"test_key=test_value",
|
||
|
|
"--metadata",
|
||
|
|
"experiment=001",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_center_frequency_metadata(self):
|
||
|
|
"""Test that center frequency is included in metadata."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_cf.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--center-frequency",
|
||
|
|
"915e6",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-v",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
# In verbose mode, should show frequency
|
||
|
|
assert "915" in result.output
|
||
|
|
|
||
|
|
def test_verbose_output(self):
|
||
|
|
"""Test verbose output."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_verbose.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "1e6", "--num-samples", "1000", "--output", output, "-v"]
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
# Verbose output should contain more info
|
||
|
|
assert len(result.output) > 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestMessageSources:
|
||
|
|
"""Test different message sources for modulation."""
|
||
|
|
|
||
|
|
def test_qam_random_bits(self):
|
||
|
|
"""Test QAM with random bits (default)."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam_random.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--message-source",
|
||
|
|
"random",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam_string_message(self):
|
||
|
|
"""Test QAM with string message source."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "qam_string.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--message-source",
|
||
|
|
"string",
|
||
|
|
"--message-content",
|
||
|
|
"Hello World",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_qam_file_message(self):
|
||
|
|
"""Test QAM with file message source."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
# Create a test file
|
||
|
|
message_file = os.path.join(tmpdir, "message.bin")
|
||
|
|
with open(message_file, "wb") as f:
|
||
|
|
f.write(b"Test message content")
|
||
|
|
|
||
|
|
output = os.path.join(tmpdir, "qam_file.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"qam",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--order",
|
||
|
|
"16",
|
||
|
|
"--symbol-rate",
|
||
|
|
"1e5",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--message-source",
|
||
|
|
"file",
|
||
|
|
"--message-content",
|
||
|
|
message_file,
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
|
||
|
|
class TestOverwriteProtection:
|
||
|
|
"""Test overwrite protection and file handling."""
|
||
|
|
|
||
|
|
def test_overwrite_protection_sigmf(self):
|
||
|
|
"""Test that overwrite protection works for sigmf files."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone.sigmf")
|
||
|
|
|
||
|
|
# First generation should succeed
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "1e6", "--num-samples", "1000", "--output", output, "-q"]
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
# Second generation without --overwrite should fail
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "1e6", "--num-samples", "1000", "--output", output]
|
||
|
|
)
|
||
|
|
assert result.exit_code != 0
|
||
|
|
assert "exist" in result.output.lower()
|
||
|
|
|
||
|
|
# Third generation with --overwrite should succeed
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"--overwrite",
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_overwrite_protection_npy(self):
|
||
|
|
"""Test that overwrite protection works for NPY files."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone.npy")
|
||
|
|
|
||
|
|
# First generation should succeed
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--format",
|
||
|
|
"npy",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
# Second generation without --overwrite should fail
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--format",
|
||
|
|
"npy",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code != 0
|
||
|
|
assert "exist" in result.output.lower()
|
||
|
|
|
||
|
|
|
||
|
|
class TestParameterValidation:
|
||
|
|
"""Test parameter validation and error handling."""
|
||
|
|
|
||
|
|
def test_invalid_sample_rate_type(self):
|
||
|
|
"""Test that invalid sample rate type is rejected."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli, ["generate", "tone", "--sample-rate", "invalid", "--num-samples", "1000", "--output", output]
|
||
|
|
)
|
||
|
|
assert result.exit_code != 0
|
||
|
|
|
||
|
|
def test_frequency_shift_formatting(self):
|
||
|
|
"""Test that frequency shift accepts scientific notation."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_shift.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--frequency-shift",
|
||
|
|
"1e5",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|
||
|
|
|
||
|
|
def test_both_num_samples_and_duration(self):
|
||
|
|
"""Test that num-samples takes precedence when both provided."""
|
||
|
|
runner = CliRunner()
|
||
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||
|
|
output = os.path.join(tmpdir, "tone_both.sigmf")
|
||
|
|
result = runner.invoke(
|
||
|
|
cli,
|
||
|
|
[
|
||
|
|
"generate",
|
||
|
|
"tone",
|
||
|
|
"--sample-rate",
|
||
|
|
"1e6",
|
||
|
|
"--num-samples",
|
||
|
|
"1000",
|
||
|
|
"--duration",
|
||
|
|
"0.01",
|
||
|
|
"--output",
|
||
|
|
output,
|
||
|
|
"-q",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
assert result.exit_code == 0
|