Initial Galileo receiver and SignalSim integration

This commit is contained in:
2026-07-30 23:26:03 +03:00
commit 2a6ccfc23d
153 changed files with 8080 additions and 0 deletions
+86
View File
@@ -0,0 +1,86 @@
import numpy as np
import pytest
from galileo_rx.acquisition.acquisition_manager import AcquisitionManager
from galileo_rx.config import AcquisitionConfig, FrontendConfig, RfConfig
from galileo_rx.frontend.frontend_chain import FrontendChain
from galileo_rx.signal_profiles import galileo_e1_profile
from galileo_rx.signals.code_provider import SyntheticCodeProvider
from galileo_rx.signals.signal_generator import ChannelModelConfig, SatelliteSignalConfig, SyntheticSignalGenerator
def make_signal(fs: float, delay_chips: float, doppler_hz: float, periods: int = 2, real_if: bool = False):
profile = galileo_e1_profile()
provider = SyntheticCodeProvider()
if_hz = 2.0e6 if real_if else 0.0
generator = SyntheticSignalGenerator(
profile,
provider,
fs,
if_hz,
real_if,
[SatelliteSignalConfig(prn=5, component="E1C", code_delay_chips=delay_chips, doppler_hz=doppler_hz, cn0_db_hz=48.0)],
ChannelModelConfig(seed=22),
)
samples, _ = generator.generate(int(round(periods * 0.004 * fs)))
if real_if:
rf = RfConfig(sample_format="float64", input_is_real_if=True, sampling_frequency_hz=fs, intermediate_frequency_hz=if_hz)
samples = FrontendChain(rf, FrontendConfig()).process(samples)
return samples, profile, provider
@pytest.mark.parametrize("doppler_hz", [-1250.0, 1750.0])
def test_pcps_detects_doppler_and_code(doppler_hz: float) -> None:
fs = 4.092e6
delay_chips = 37.0
samples, profile, provider = make_signal(fs, delay_chips, doppler_hz)
config = AcquisitionConfig(
doppler_min_hz=-2500,
doppler_max_hz=2500,
doppler_step_hz=250,
coherent_integration_ms=4,
noncoherent_accumulations=2,
detection_threshold=2.0,
zero_padding_mode="none",
parabolic_interpolation=False,
)
result, _ = AcquisitionManager(profile, provider, fs, config).acquire_one(samples, "E1C", 5)
assert result.detected
assert result.raw_doppler_hz == doppler_hz
assert abs(result.code_phase_chips - delay_chips) <= 0.26
def test_pcps_noninteger_samples_per_chip_and_boundary() -> None:
fs = 4.5e6
delay_chips = 4091.2
samples, profile, provider = make_signal(fs, delay_chips, 500.0)
config = AcquisitionConfig(
doppler_min_hz=0,
doppler_max_hz=1000,
doppler_step_hz=250,
coherent_integration_ms=4,
noncoherent_accumulations=2,
detection_threshold=1.8,
zero_padding_mode="none",
parabolic_interpolation=False,
)
result, _ = AcquisitionManager(profile, provider, fs, config).acquire_one(samples, "E1C", 5)
circular_error = min(abs(result.code_phase_chips - delay_chips), 4092 - abs(result.code_phase_chips - delay_chips))
assert result.detected and circular_error < 0.6
def test_real_if_acquisition() -> None:
fs = 8.0e6
samples, profile, provider = make_signal(fs, 21.0, 750.0, real_if=True)
config = AcquisitionConfig(
doppler_min_hz=250,
doppler_max_hz=1250,
doppler_step_hz=250,
coherent_integration_ms=4,
noncoherent_accumulations=2,
detection_threshold=1.7,
zero_padding_mode="none",
parabolic_interpolation=False,
)
result, _ = AcquisitionManager(profile, provider, fs, config).acquire_one(samples, "E1C", 5)
assert result.detected and result.raw_doppler_hz == 750.0
@@ -0,0 +1,33 @@
import numpy as np
from galileo_rx.acquisition.acquisition_manager import AcquisitionManager
from galileo_rx.config import AcquisitionConfig, TrackingConfig
from galileo_rx.signal_profiles import galileo_e1_profile
from galileo_rx.signals.code_provider import SyntheticCodeProvider
from galileo_rx.tracking.tracking_channel import TrackingChannel
from galileo_rx.types import AcquisitionResult, TrackingState
def test_noise_only_does_not_pass_peak_ratio_threshold() -> None:
fs = 4.092e6
rng = np.random.default_rng(345)
samples = rng.standard_normal(int(0.008 * fs)) + 1j * rng.standard_normal(int(0.008 * fs))
config = AcquisitionConfig(
doppler_min_hz=-500, doppler_max_hz=500, doppler_step_hz=250,
noncoherent_accumulations=2, detection_threshold=2.5, zero_padding_mode="none",
)
result, _ = AcquisitionManager(galileo_e1_profile(), SyntheticCodeProvider(), fs, config).acquire_one(samples, "E1C", 1)
assert not result.detected
def test_consecutive_failures_request_reacquisition() -> None:
fs = 4.092e6
acquisition = AcquisitionResult(
True, 1, "E1B", 0.0, 0.0, 0.0, 0.0, 1.0, 0.1, 10.0, None, 4.0, 1
)
config = TrackingConfig(use_fll_assist=False, loss_of_lock_epochs=3)
channel = TrackingChannel(galileo_e1_profile(), SyntheticCodeProvider(), fs, config, acquisition)
zero_epoch = np.zeros(channel.integration_samples, dtype=np.complex128)
measurements = [channel.process_epoch(zero_epoch) for _ in range(3)]
assert measurements[-1].reacquisition_requested
assert measurements[-1].state == TrackingState.REACQUISITION
@@ -0,0 +1,33 @@
import numpy as np
from galileo_rx.fixed_point.bittrue_pipeline import compare_acquisition, sweep_acquisition_word_lengths
from galileo_rx.fixed_point.fixed_acquisition import FixedPcpsAcquisition
def test_fixed_acquisition_peak_agrees_with_float() -> None:
rng = np.random.default_rng(4)
count = 256
fs = 1.024e6
replica = np.where(rng.integers(0, 2, count), 1.0, -1.0)
delay = 73
doppler = 4000.0
time_s = np.arange(count) / fs
samples = np.roll(replica, delay) * np.exp(2j * np.pi * doppler * time_s)
bins = np.asarray([-4000.0, 0.0, 4000.0])
reference = np.asarray([
np.abs(np.fft.ifft(np.fft.fft(samples * np.exp(-2j * np.pi * frequency * time_s)) * np.conjugate(np.fft.fft(replica)))) ** 2
for frequency in bins
])
fixed = FixedPcpsAcquisition().search(samples, replica, fs, bins)
comparison = compare_acquisition(reference / count**2, fixed, threshold=2.5)
assert comparison.doppler_bin_agreement
assert comparison.code_phase_agreement
assert comparison.acquisition_decision_agreement
def test_word_length_sweep_reports_each_width() -> None:
rng = np.random.default_rng(7)
replica = np.where(rng.integers(0, 2, 64), 1.0, -1.0)
reports = sweep_acquisition_word_lengths(np.roll(replica, 9).astype(complex), replica, 1e6, np.asarray([0.0]), [6, 8])
assert [item["input_bits"] for item in reports] == [6, 8]
assert all(item["code_phase_agreement"] for item in reports)
@@ -0,0 +1,57 @@
import numpy as np
from galileo_rx.acquisition.acquisition_manager import AcquisitionManager
from galileo_rx.config import AcquisitionConfig
from galileo_rx.signal_profiles import galileo_e1_profile
from galileo_rx.signals.code_provider import SyntheticCodeProvider
from galileo_rx.signals.signal_generator import ChannelModelConfig, SatelliteSignalConfig, SyntheticSignalGenerator
def test_e1c_secondary_wipeoff_for_eight_ms() -> None:
fs = 4.092e6
profile = galileo_e1_profile()
provider = SyntheticCodeProvider()
generator = SyntheticSignalGenerator(
profile, provider, fs,
satellites=[SatelliteSignalConfig(8, "E1C", code_delay_chips=12.0, doppler_hz=500.0)],
channel=ChannelModelConfig(add_awgn=False),
)
samples, _ = generator.generate(int(0.008 * fs))
config = AcquisitionConfig(
doppler_min_hz=500,
doppler_max_hz=500,
doppler_step_hz=250,
coherent_integration_ms=8,
noncoherent_accumulations=1,
detection_threshold=1.5,
zero_padding_mode="none",
secondary_code_mode="wipeoff",
parabolic_interpolation=False,
)
result, _ = AcquisitionManager(profile, provider, fs, config).acquire_one(samples, "E1C", 8)
assert result.detected and abs(result.code_phase_chips - 12.0) < 0.3
def test_e1b_bit_transition_hypothesis() -> None:
fs = 4.092e6
profile = galileo_e1_profile()
provider = SyntheticCodeProvider()
generator = SyntheticSignalGenerator(
profile, provider, fs,
satellites=[SatelliteSignalConfig(9, "E1B", code_delay_chips=20.0, navigation_bits=[1, -1])],
channel=ChannelModelConfig(add_awgn=False),
)
samples, _ = generator.generate(int(0.008 * fs))
config = AcquisitionConfig(
doppler_min_hz=0,
doppler_max_hz=0,
doppler_step_hz=250,
coherent_integration_ms=8,
noncoherent_accumulations=1,
detection_threshold=1.5,
zero_padding_mode="none",
bit_transition_mode="hypothesis",
parabolic_interpolation=False,
)
result, _ = AcquisitionManager(profile, provider, fs, config).acquire_one(samples, "E1B", 9)
assert result.detected and abs(result.code_phase_chips - 20.0) < 0.3
@@ -0,0 +1,25 @@
import pytest
@pytest.mark.signalsim
def test_signalsim_pcps_detects_e1b_and_e1c(signalsim_validation_result) -> None:
report, _, output = signalsim_validation_result
for component in ("E1B", "E1C"):
result = report["acquisition"][component]
assert result["detected"]
assert result["prn"] == 1
assert abs(result["raw_doppler_error_hz"]) <= 250.0
assert result["detection_metric"] >= 2.5
assert result["overflow_count"] == 0
assert (output / f"acquisition_map_{component.lower()}.png").is_file()
# SignalSim does not export code-delay truth, but the independently acquired
# E1B and E1C components share the same propagation delay.
delta = abs(report["acquisition"]["E1B"]["code_phase_samples"] - report["acquisition"]["E1C"]["code_phase_samples"])
assert delta < 1.0
fixed = report["fixed_point_acquisition"]
assert fixed["doppler_bin_agreement"]
assert fixed["code_phase_agreement"]
assert fixed["acquisition_decision_agreement"]
assert fixed["overflow_count"] == 0
assert fixed["normalized_rms_error"] < 0.005
assert report["rtl_vector_manifest"]["sample_count"] == 5 * 32736
@@ -0,0 +1,47 @@
from __future__ import annotations
from pathlib import Path
import numpy as np
import pytest
from galileo_rx.signal_profiles import galileo_e1_profile
from galileo_rx.signals.signalsim_codes import SignalSimCodeProvider
def _checkout() -> Path:
candidates = [Path("signalsim/SignalSim"), Path("../SignalSim"), Path("../signalsim/SignalSim")]
for candidate in candidates:
if (candidate / "src" / "MemoryCode.dat").is_file():
return candidate.resolve()
pytest.skip("local globsky/SignalSim checkout with src/MemoryCode.dat is unavailable")
@pytest.mark.signalsim
def test_signalsim_e1_code_table_unpacking_and_components() -> None:
provider = SignalSimCodeProvider(_checkout())
profile = galileo_e1_profile()
e1b = provider.primary_code(profile, "E1B", 1)
e1c = provider.primary_code(profile, "E1C", 1)
assert e1b.dtype == np.int8 and e1b.size == 4092
assert e1c.dtype == np.int8 and e1c.size == 4092
assert set(e1b.tolist()) == {-1, 1}
assert not np.array_equal(e1b, e1c)
# First table word is 0xf5d71013 and SignalSim reads it MSB first.
np.testing.assert_array_equal(e1b[:8], [-1, -1, -1, -1, 1, -1, 1, -1])
# Every 32-word storage sector has one discarded padding bit. Chip 1023
# is therefore word 32 bit 31, not word 31 bit 0.
words = provider._e1_words[0] # executable check of the external table layout
expected_chip_1023 = -1 if (int(words[32]) >> 31) & 1 else 1
assert int(e1b[1023]) == expected_chip_1023
@pytest.mark.signalsim
def test_signalsim_e1c_secondary_is_25_symbols_lsb_first() -> None:
provider = SignalSimCodeProvider(_checkout())
secondary = provider.secondary_code(galileo_e1_profile(), "E1C", 1)
assert secondary is not None and secondary.size == 25
# 0x9b501c positions are addressed least-significant bit first.
expected_bits = np.asarray([(0x9B501C >> index) & 1 for index in range(25)])
np.testing.assert_array_equal(secondary, np.where(expected_bits == 0, 1, -1))
assert provider.secondary_code(galileo_e1_profile(), "E1B", 1) is None
@@ -0,0 +1,15 @@
import pytest
@pytest.mark.signalsim
def test_signalsim_e1b_raw_symbols_contain_periodic_inav_sync(signalsim_validation_result) -> None:
report, _, _ = signalsim_validation_result
symbols = report["e1b_symbols"]
assert symbols["recovered_symbols"] >= 450
assert symbols["comparison_scope"] == "known_10_symbol_INAV_sync_words_only"
assert symbols["symbol_errors"] == 0
assert symbols["frame_or_page_sync"]
indices = [match["symbol_index"] for match in symbols["sync_matches"]]
# I/NAV even and odd parts each begin with sync, 250 symbols (1 s) apart.
assert any(right - left == 250 for left, right in zip(indices, indices[1:]))
assert symbols["fec_decoding"] == "not_implemented"
@@ -0,0 +1,13 @@
import pytest
@pytest.mark.signalsim
def test_signalsim_e1c_secondary_sync_and_coherent_gain(signalsim_validation_result) -> None:
report, _, _ = signalsim_validation_result
sync = report["e1c_secondary"]
assert sync["phase"] == 10
assert sync["normalized_correlation"] > 0.9
assert sync["sign_errors"] == 0
assert sync["compared_periods"] >= 100
assert sync["wiped_25_period_coherent_gain_db"] > 10.0
assert sync["wipeoff_gain_improvement_db"] > 10.0
@@ -0,0 +1,19 @@
import pytest
@pytest.mark.signalsim
def test_signalsim_full_multi_vector_regression(signalsim_validation_result) -> None:
report, passed, _ = signalsim_validation_result
assert passed
assert report["passes"]["full_vector_set"]
vectors = report["full_vectors"]
assert vectors["nominal_cn0_single_prn"]["acquisition"][1]["pass"]
assert vectors["nominal_cn0_single_prn"]["tracking_e1c"]["pass"]
assert vectors["negative_doppler"]["acquisition"][27]["pass"]
negative_tracking = vectors["negative_doppler"]["tracking_e1b"]
assert negative_tracking["pass"]
assert negative_tracking["post_pull_in_reacquisition_epoch_count"] == 0
assert negative_tracking["estimated_doppler_hz"] < 0.0
multiple = vectors["multiple_prns"]["acquisition"]
assert set(multiple) == {1, 4, 19, 21, 27}
assert all(result["pass"] for result in multiple.values())
@@ -0,0 +1,53 @@
from __future__ import annotations
import numpy as np
import pytest
from galileo_rx.io.signalsim_reader import SignalSimReader, SignalSimSampleSpec
def test_iq8_signed_i_then_q_and_double_precision(tmp_path) -> None:
path = tmp_path / "iq8.bin"
path.write_bytes(bytes([0x80, 0x7F, 0xFF, 0x01, 0x02, 0xFE]))
samples = SignalSimReader(path, SignalSimSampleSpec("IQ8")).read()
assert samples.dtype == np.complex128
np.testing.assert_array_equal(samples, [-128 + 127j, -1 + 1j, 2 - 2j])
def test_iq16_little_endian_scaling_and_qi_order(tmp_path) -> None:
path = tmp_path / "iq16.bin"
np.asarray([100, -200, -300, 400], dtype="<i2").tofile(path)
spec = SignalSimSampleSpec("IQ16", iq_order="QI", scale_counts_per_unit=100.0)
np.testing.assert_allclose(SignalSimReader(path, spec).read(), [-2 + 1j, 4 - 3j], rtol=0.0, atol=0.0)
def test_iq4_sign_magnitude(tmp_path) -> None:
path = tmp_path / "iq4.bin"
path.write_bytes(bytes([0x17, 0xA9, 0x80]))
np.testing.assert_array_equal(SignalSimReader(path, SignalSimSampleSpec("IQ4")).read(), [1 + 7j, -2 - 1j, 0 + 0j])
def test_iq2_packed_order_and_offset(tmp_path) -> None:
path = tmp_path / "iq2.bin"
# sample 0: I=+1,Q=-1 -> 0xd; sample 1: I=-1,Q=+1 -> 0x7
path.write_bytes(bytes([0x7D]))
reader = SignalSimReader(path, SignalSimSampleSpec("IQ2"))
np.testing.assert_array_equal(reader.read(), [1 - 1j, -1 + 1j])
np.testing.assert_array_equal(reader.read(1, 1), [-1 + 1j])
def test_truncation_odd_iq_and_block_streaming(tmp_path) -> None:
path = tmp_path / "odd.bin"
path.write_bytes(b"\x01\x02\x03")
reader = SignalSimReader(path, SignalSimSampleSpec("IQ8"))
with pytest.raises(ValueError, match="truncated"):
_ = reader.sample_count
complete = tmp_path / "complete.bin"
complete.write_bytes(bytes(range(12)))
reader = SignalSimReader(complete, SignalSimSampleSpec("IQ8"), memory_map=True)
blocks = list(reader.blocks(2, offset_samples=1, count=4))
assert [block.size for block in blocks] == [2, 2]
np.testing.assert_array_equal(np.concatenate(blocks), [2 + 3j, 4 + 5j, 6 + 7j, 8 + 9j])
with pytest.raises(EOFError, match="only 1 remain"):
reader.read(2, 5)
@@ -0,0 +1,15 @@
import pytest
@pytest.mark.signalsim
def test_signalsim_e1b_and_e1c_tracking_lock(signalsim_validation_result) -> None:
report, _, output = signalsim_validation_result
for component in ("E1B", "E1C"):
result = report["tracking"][component]
assert result["final_state"] == "PLL_TRACK"
assert result["reacquisition_epoch_count"] == 0
assert abs(result["doppler_error_hz"]) < 10.0
assert result["doppler_jitter_hz_rms"] < 5.0
assert result["dll_error_chips_rms"] < 0.15
assert result["mean_code_lock_metric"] > 1.0
assert (output / f"tracking_{component.lower()}.csv").is_file()
+103
View File
@@ -0,0 +1,103 @@
import numpy as np
from galileo_rx.acquisition.acquisition_manager import AcquisitionManager
from galileo_rx.config import AcquisitionConfig, TrackingConfig
from galileo_rx.signal_profiles import galileo_e1_profile
from galileo_rx.signals.code_provider import SyntheticCodeProvider
from galileo_rx.signals.signal_generator import ChannelModelConfig, SatelliteSignalConfig, SyntheticSignalGenerator
from galileo_rx.tracking.tracking_channel import TrackingChannel
from galileo_rx.tracking.tracking_manager import TrackingManager
from galileo_rx.types import AcquisitionResult, TrackingState
def test_acquisition_to_tracking_and_multichannel_manager() -> None:
fs = 4.092e6
profile = galileo_e1_profile()
provider = SyntheticCodeProvider()
satellites = [
SatelliteSignalConfig(2, "E1C", code_delay_chips=81.0, doppler_hz=1000.0, carrier_phase_rad=0.1, cn0_db_hz=52.0),
]
generator = SyntheticSignalGenerator(profile, provider, fs, satellites=satellites, channel=ChannelModelConfig(seed=9))
samples, _ = generator.generate(int(0.12 * fs))
acquisition_config = AcquisitionConfig(
doppler_min_hz=0,
doppler_max_hz=2000,
doppler_step_hz=250,
coherent_integration_ms=4,
noncoherent_accumulations=2,
detection_threshold=2.0,
zero_padding_mode="none",
parabolic_interpolation=False,
)
acquisition, _ = AcquisitionManager(profile, provider, fs, acquisition_config).acquire_one(samples, "E1C", 2)
assert acquisition.detected
tracking_config = TrackingConfig(
integration_ms=4.0,
use_fll_assist=True,
pull_in_duration_ms=20.0,
secondary_code_mode="wipeoff",
loss_of_lock_epochs=20,
)
manager = TrackingManager(profile, provider, fs, tracking_config)
channel = manager.add_channel(acquisition)
measurements = channel.process(samples[: 25 * channel.integration_samples])
assert len(measurements) == 25
assert not measurements[-1].reacquisition_requested
assert measurements[-1].state in {TrackingState.PLL_TRACK, TrackingState.WEAK_SIGNAL}
assert abs(measurements[-1].carrier_frequency_hz - 1000.0) < 100.0
assert np.median([abs(item.dll_error_chips) for item in measurements[-10:]]) < 0.1
def test_tracking_under_doppler_rate() -> None:
fs = 4.092e6
profile = galileo_e1_profile()
provider = SyntheticCodeProvider()
generator = SyntheticSignalGenerator(
profile,
provider,
fs,
satellites=[SatelliteSignalConfig(4, "E1B", code_delay_chips=11.0, doppler_hz=-500.0, doppler_rate_hz_s=100.0)],
channel=ChannelModelConfig(add_awgn=False),
)
samples, _ = generator.generate(int(0.08 * fs))
acquisition_config = AcquisitionConfig(
doppler_min_hz=-1000,
doppler_max_hz=0,
doppler_step_hz=250,
coherent_integration_ms=4,
noncoherent_accumulations=1,
detection_threshold=1.5,
zero_padding_mode="none",
)
acquisition, _ = AcquisitionManager(profile, provider, fs, acquisition_config).acquire_one(samples, "E1B", 4)
channel = TrackingChannel(profile, provider, fs, TrackingConfig(pull_in_duration_ms=40.0), acquisition)
measurements = channel.process(samples)
expected_hz = -500.0 + 100.0 * measurements[-1].time_s
assert abs(measurements[-1].carrier_frequency_hz - expected_hz) < 100.0
def test_code_aiding_uses_doppler_not_nonzero_if_center() -> None:
fs = 8.184e6
profile = galileo_e1_profile()
provider = SyntheticCodeProvider()
if_hz = 2.046e6
doppler_hz = 1800.0
acquisition = AcquisitionResult(
detected=True,
prn=1,
component="E1B",
code_phase_samples=0.0,
code_phase_chips=0.0,
doppler_hz=doppler_hz,
carrier_frequency_hz=if_hz + doppler_hz,
peak_power=1.0,
second_peak_power=0.1,
detection_metric=10.0,
estimated_cn0_db_hz=None,
coherent_integration_ms=4.0,
noncoherent_accumulations=1,
)
channel = TrackingChannel(profile, provider, fs, TrackingConfig(use_fll_assist=False), acquisition)
expected_hz = profile.code_rate_hz * (1.0 + doppler_hz / profile.carrier_frequency_hz)
assert abs(channel.code_nco.frequency_hz - expected_hz) < 1e-9
assert channel.carrier_center_hz == if_hz