test(pps): add end-to-end timing validation

This commit is contained in:
2026-07-28 21:13:10 +03:00
parent 156cfc8b77
commit 73db2a8355
5 changed files with 1164 additions and 0 deletions
@@ -0,0 +1,130 @@
# GPS PPS end-to-end Linux validation
This package is a deterministic, causal validation of the project timing path
from synthetic GPS LNAV data to the physical PPS edge. It exists separately
from `pps_python`, whose models concentrate on WLS noise, oscillator behavior,
and adaptive-loop performance.
The validation does not use Linux wall-clock scheduling. The 1 kHz, 50 bit/s,
RF-sample, PCLK, and 125 MHz events advance from synthetic timestamps, making
the signed absolute-time result repeatable on a normal Linux host.
## What is modeled
The model follows the current firmware and RTL conventions:
1. It generates deterministic GPS LNAV subframes with D29*/D30* data inversion
and valid parity for all ten 30-bit words.
2. Each navigation bit is expanded to twenty 1 kHz prompt-sign samples.
3. The `bit_gen.sv` 40 ms `1 -> 0` transition acquisition, majority decision,
20 ms `bit_valid` cadence, and registered ABS_SAMPLE behavior are applied.
4. The tracking wrapper's chronological `2 + 300 + 8 = 310` bit window,
preamble search, parity check, `load_sigs`, and subframe interrupt ordering
are applied.
5. `nav_solve_timer` is initialized from HOW Z-count and counts later bit-valid
events. Navigation-solve interrupt, PPS counter capture, capture sequence,
and a 32-bit RF ABS_SAMPLE wrap are exercised.
6. Firmware reconstructs periodic TOW from the HOW/sample anchor and evaluates
`TOW + POSTPONE_TOW + STARTOFFSET - b/c`.
7. The timestamped 125 MHz PPS peripheral performs coarse/normal mapping,
ignores synthetic CPU/APB delay as phase, acquires lock after five valid
updates, and generates integer-second edges.
8. DUT edges are compared with an independent trusted PPS. Optional DUT and
trusted-output delays make board, cable, and vendor delay sign explicit.
This is a digital epoch-path validation, not an RF waveform simulator. The
existing `pps_python` tests continue to cover WLS, oscillator, PI, adaptive
quality, and holdover behavior.
## Run
From `soc/software`:
```bash
python3 -m pps_end_to_end_validation.pps_end_to_end
python3 -m unittest pps_end_to_end_validation/test_pps_end_to_end.py
```
The default command executes the diagnostic matrix:
```text
scenario bits correction reference error DUT - trusted PPS
current-rtl/160ms 9 160 ms -20.000032 ms +20.000032 ms
current-bit/160ms 8 160 ms -0.000032 ms +0.000032 ms
current-rtl/180ms 9 180 ms -0.000032 ms +0.000032 ms
```
Positive `DUT - trusted PPS` means the project output is late. The default
32 ns remainder is the explicitly configured four-tick PCLK capture pipeline;
it is not part of the LNAV-bit mismatch.
Run one scenario or obtain machine-readable output:
```bash
python3 -m pps_end_to_end_validation.pps_end_to_end \
--mode current-rtl --postpone-ms 160 --json
python3 -m pps_end_to_end_validation.pps_end_to_end \
--mode current-bit \
--dut-output-delay-ns 12 \
--trusted-output-delay-ns 35
```
## 20 ms diagnosis
The current tracking wrapper shifts the current bit into `shift_reg_n`, while
its preamble/parity predicate reads the preceding `shift_reg_p`. On the eighth
following preamble bit, the validator therefore sees only the previous seven
bits. The register contains all eight after the edge, but the bit-valid edge
condition has ended. Validation occurs on the ninth bit, 20 ms later.
The implemented interface is consequently:
```text
current RTL delivery 9 * 20 ms = 180 ms
firmware POSTPONE_TOW 8 * 20 ms = 160 ms
signed epoch mismatch 20 ms
```
With the current bitstream, `POSTPONE_TOW=0.180` is a firmware-only workaround
and must be used consistently by both satellite-state and PPS calculations.
It is 180 ms, not 190 ms. The preferred permanent correction is to validate a
candidate window containing the current eighth bit and retain the intended
160 ms firmware compensation.
The validation intentionally does not modify production RTL or firmware.
## Other regression coverage
The unit tests verify:
- parity-valid LNAV round-trip and rejection after one corrupted data bit;
- 1 kHz prompt-sign to 50 bit/s generation and the registered 1 ms sample lag;
- ninth-bit delivery in current RTL and eighth-bit delivery in the candidate
implementation;
- constant signed PPS error over successive physical edges;
- 180 ms firmware compatibility with current RTL;
- invariance to synthetic CPU/APB stalls when capture timestamps are used;
- DUT-versus-trusted output-delay sign; and
- rejection behavior for the firmware-permitted `NAV_SOLVE_TIMER=40` case.
The `N=40` case divides the GPS week but not a 300-bit subframe. When firmware
refreshes the HOW/sample anchor after six seconds, the timer phase is half a
solve period away from that anchor and the current `N * 20 ms` snap rejects
otherwise valid solves. Production `N=10` divides 300 and is unaffected.
If general timer periods are required, reconstruct elapsed time on the
fundamental 20 ms navigation-bit grid. A narrower compatibility restriction is
to accept only periods for which `300 % N == 0`.
## Hardware confirmation
Before selecting a production correction:
1. Probe `bit_valid`, the final preamble bits, `preamble_ok`, parity result, and
`load_sigs` with ILA.
2. Count whether delivery is on following bit 8 or 9 in the flashed bitstream.
3. Measure DUT and trusted PPS with a common time-interval counter and record
the sign. The diagnosis predicts a DUT edge approximately 20 ms late.
4. Apply only nanosecond-scale output/cable calibration after the epoch
mismatch is removed.
@@ -0,0 +1 @@
"""End-to-end Linux validation for the GPS LNAV-to-PPS timing path."""
@@ -0,0 +1,874 @@
"""Causal Linux model of the GPS LNAV-to-PPS timing path.
This module is intentionally narrower than a general GNSS simulator. It
models the project boundary which can create a constant absolute-time error:
* parity-correct 30-bit GPS LNAV words, including D30* data inversion;
* the 1 kHz prompt-sign transition detector and 20 ms bit-valid generator;
* the tracking wrapper's 310-bit subframe/preamble search window;
* current registered-window and proposed current-bit-aware validation;
* subframe-ready and periodic navigation-solve interrupts;
* RF ABS_SAMPLE anchoring and the timestamped 125 MHz PPS capture;
* the firmware HOW + POSTPONE_TOW + STARTOFFSET - b/c mapping; and
* coarse/normal PPS commits followed by comparison with an independent PPS.
Long stretches between 50 bit/s events are evaluated analytically. No result
depends on the Linux scheduler or wall clock.
"""
from __future__ import annotations
import argparse
import json
import math
from collections import deque
from dataclasses import asdict, dataclass
from enum import Enum
from typing import Iterable, Sequence
C = 299_792_458.0
PREAMBLE = 0x8B
LNAV_WORD_BITS = 30
LNAV_DATA_BITS = 24
LNAV_WORDS_PER_SUBFRAME = 10
LNAV_SUBFRAME_BITS = LNAV_WORD_BITS * LNAV_WORDS_PER_SUBFRAME
LNAV_BIT_PERIOD_S = 0.020
LNAV_BITS_PER_SUBFRAME = 300
LNAV_WINDOW_BITS = 2 + LNAV_SUBFRAME_BITS + 8
TRACKING_INTEGRATION_PERIOD_S = 0.001
TRACKING_INTEGRATIONS_PER_BIT = 20
RF_HZ = 16_367_981.0
PPS_HZ = 125_000_000.0
START_OFFSET_S = 68.802e-3
NAV_TOW_SNAP_TOLERANCE_S = 0.005
Q32 = 1 << 32
TICK_INTEGER_TOLERANCE = 0.01
class DecoderMode(str, Enum):
"""Subframe search ordering relative to the current bit-valid event."""
CURRENT_RTL = "current-rtl"
CURRENT_BIT = "current-bit"
def _bits(value: int, width: int) -> list[int]:
return [(value >> shift) & 1 for shift in range(width - 1, -1, -1)]
def _integer(bits: Iterable[int]) -> int:
value = 0
for bit in bits:
value = (value << 1) | int(bit)
return value
def _xor(values: Iterable[int]) -> int:
result = 0
for value in values:
result ^= int(value)
return result
def lnav_parity(data_bits: Sequence[int], previous_d29: int, previous_d30: int) -> tuple[int, ...]:
"""Return transmitted D25*..D30* using the equations in tracking_wrapper.
``data_bits`` are normalized D1..D24. The equations below are a direct
index-form transcription of ``parity_calc`` in tracking_wrapper.sv.
"""
if len(data_bits) != LNAV_DATA_BITS:
raise ValueError("an LNAV word must contain 24 data bits")
d = [0] + [int(bit) for bit in data_bits] # ICD bit numbers are one based.
return (
_xor((previous_d29, d[1], d[2], d[3], d[5], d[6], d[10], d[11],
d[12], d[13], d[14], d[17], d[18], d[20], d[23])),
_xor((previous_d30, d[2], d[3], d[4], d[6], d[7], d[11], d[12],
d[13], d[14], d[15], d[18], d[19], d[21], d[24])),
_xor((previous_d29, d[1], d[3], d[4], d[5], d[7], d[8], d[12],
d[13], d[14], d[15], d[16], d[19], d[20], d[22])),
_xor((previous_d30, d[2], d[4], d[5], d[6], d[8], d[9], d[13],
d[14], d[15], d[16], d[17], d[20], d[21], d[23])),
_xor((previous_d30, d[1], d[3], d[5], d[6], d[7], d[9], d[10],
d[14], d[15], d[16], d[17], d[18], d[21], d[22], d[24])),
_xor((previous_d29, d[3], d[5], d[6], d[8], d[9], d[10], d[11],
d[13], d[15], d[19], d[22], d[23], d[24])),
)
def encode_lnav_word(
data: int, previous_d29: int, previous_d30: int
) -> tuple[tuple[int, ...], int, int, int]:
"""Encode one normalized 24-bit word and return transmitted bits/state."""
if data < 0 or data >= (1 << LNAV_DATA_BITS):
raise ValueError("LNAV data does not fit in 24 bits")
normalized = _bits(data, LNAV_DATA_BITS)
parity = lnav_parity(normalized, previous_d29, previous_d30)
transmitted_data = tuple(bit ^ previous_d30 for bit in normalized)
transmitted = transmitted_data + parity
normalized_word = (data << 6) | _integer(parity)
return transmitted, parity[4], parity[5], normalized_word
@dataclass(frozen=True)
class EncodedSubframe:
z_count: int
subframe_id: int
transmitted_bits: tuple[int, ...]
normalized_words: tuple[int, ...]
ending_d29: int
ending_d30: int
def encode_lnav_subframe(
z_count: int,
subframe_id: int,
previous_d29: int,
previous_d30: int,
*,
week: int = 2200,
) -> EncodedSubframe:
"""Generate a deterministic, parity-valid LNAV subframe.
The TLM preamble, HOW Z-count/subframe ID, and the modulo-1024 week field
used by this firmware are placed in their implemented bit positions. The
remaining data is deterministic filler so all ten parity checks are live.
"""
if not 0 <= z_count < (1 << 17):
raise ValueError("HOW Z-count must fit in 17 bits")
if not 1 <= subframe_id <= 5:
raise ValueError("subframe ID must be 1..5")
data_words = [0] * LNAV_WORDS_PER_SUBFRAME
data_words[0] = PREAMBLE << 16
data_words[1] = (z_count << 7) | (subframe_id << 2)
if subframe_id == 1:
data_words[2] = (week & 0x3FF) << 14
for index in range(2 if subframe_id != 1 else 3, LNAV_WORDS_PER_SUBFRAME):
# Avoid changing the semantic fields above while exercising parity
# with a stable, non-zero bit pattern.
data_words[index] = (
(0x5A_0000 ^ (z_count * 0x45D9F3B) ^ (subframe_id << 12) ^ index)
& 0xFF_FFFF
)
emitted: list[int] = []
normalized_words: list[int] = []
d29, d30 = int(previous_d29), int(previous_d30)
for data in data_words:
word, d29, d30, normalized_word = encode_lnav_word(data, d29, d30)
emitted.extend(word)
normalized_words.append(normalized_word)
return EncodedSubframe(
z_count=z_count,
subframe_id=subframe_id,
transmitted_bits=tuple(emitted),
normalized_words=tuple(normalized_words),
ending_d29=d29,
ending_d30=d30,
)
@dataclass(frozen=True)
class DecodedSubframe:
z_count: int
subframe_id: int
normalized_words: tuple[int, ...]
def validate_rtl_window(window: Sequence[int]) -> DecodedSubframe | None:
"""Mirror the RTL preamble and ten-word parity predicate.
The chronological window is ``previous D29*, D30*`` followed by the
300-bit candidate subframe and eight bits of the following preamble.
"""
if len(window) != LNAV_WINDOW_BITS:
return None
preamble = tuple(_bits(PREAMBLE, 8))
inverse_preamble = tuple(bit ^ 1 for bit in preamble)
first = tuple(window[2:10])
following = tuple(window[302:310])
if first not in (preamble, inverse_preamble):
return None
if following not in (preamble, inverse_preamble):
return None
d29, d30 = int(window[0]), int(window[1])
normalized_words: list[int] = []
for word_index in range(LNAV_WORDS_PER_SUBFRAME):
start = 2 + word_index * LNAV_WORD_BITS
transmitted = tuple(int(bit) for bit in window[start : start + LNAV_WORD_BITS])
normalized_data = tuple(bit ^ d30 for bit in transmitted[:LNAV_DATA_BITS])
expected_parity = lnav_parity(normalized_data, d29, d30)
if tuple(transmitted[LNAV_DATA_BITS:]) != expected_parity:
return None
normalized_words.append((_integer(normalized_data) << 6) | _integer(expected_parity))
d29, d30 = transmitted[28], transmitted[29]
how_word = normalized_words[1]
return DecodedSubframe(
z_count=(how_word >> 13) & 0x1FFFF,
subframe_id=(how_word >> 8) & 0x7,
normalized_words=tuple(normalized_words),
)
@dataclass(frozen=True)
class SampleStamp:
raw: int
corrected: float
@dataclass(frozen=True)
class BitEvent:
stream_index: int
value: int
transmit_time_s: float
receive_time_s: float
sample_time_s: float
sample: SampleStamp
@dataclass(frozen=True)
class PromptSample:
value: int
receive_time_s: float
sample: SampleStamp
class BitGeneratorModel:
"""Mirror bit_gen.sv at its 1 kHz valid-input clock enables."""
def __init__(self, *, propagation_s: float):
self.propagation_s = propagation_s
self.prompt_history: deque[int] = deque([0] * 40, maxlen=40)
self.record_bits = False
self.loop_counter = 0
self.prompt_ones = 0
self.absolute_sample: SampleStamp | None = None
self.absolute_sample_time_s = 0.0
self.output_index = 0
def feed(self, prompt: PromptSample) -> BitEvent | None:
bit_transition = tuple(self.prompt_history) == (
(1,) * TRACKING_INTEGRATIONS_PER_BIT
+ (0,) * TRACKING_INTEGRATIONS_PER_BIT
)
loop_enabled = self.record_bits
loop_full = self.loop_counter == TRACKING_INTEGRATIONS_PER_BIT - 1
bit_value = int(self.prompt_ones > TRACKING_INTEGRATIONS_PER_BIT / 2 - 1)
sampled_before_edge = self.absolute_sample
sample_time_before_edge = self.absolute_sample_time_s
# All assignments in bit_gen are nonblocking: the output on a full
# loop sees the ABS_SAMPLE registered by the preceding 1 ms input.
self.prompt_history.append(prompt.value)
self.absolute_sample = prompt.sample
self.absolute_sample_time_s = prompt.receive_time_s
if bit_transition:
self.record_bits = True
if loop_enabled:
if loop_full:
self.loop_counter = 0
self.prompt_ones = 0
else:
self.loop_counter += 1
self.prompt_ones += prompt.value
if not loop_full or sampled_before_edge is None:
return None
event = BitEvent(
stream_index=self.output_index,
value=bit_value,
transmit_time_s=prompt.receive_time_s - self.propagation_s,
receive_time_s=prompt.receive_time_s,
sample_time_s=sample_time_before_edge,
sample=sampled_before_edge,
)
self.output_index += 1
return event
@dataclass(frozen=True)
class SubframeInterrupt:
z_count: int
subframe_id: int
words: tuple[int, ...]
bit_index_after_how: int
sample_time_s: float
irq_time_s: float
sample: SampleStamp
class TrackingDecoderModel:
"""Event-level model of tracking_wrapper's shift/search/IRQ ordering."""
def __init__(self, mode: DecoderMode, pps_hz: float = PPS_HZ):
self.mode = DecoderMode(mode)
self.pps_hz = pps_hz
self.history: deque[int] = deque(maxlen=LNAV_WINDOW_BITS)
def feed(self, event: BitEvent) -> SubframeInterrupt | None:
registered = tuple(self.history)
if self.mode == DecoderMode.CURRENT_RTL:
decoded = validate_rtl_window(registered)
self.history.append(event.value)
else:
candidate = (registered + (event.value,))[-LNAV_WINDOW_BITS:]
decoded = validate_rtl_window(candidate)
self.history.append(event.value)
if decoded is None:
return None
how_boundary = decoded.z_count * 6.0
bit_index = int(round((event.transmit_time_s - how_boundary) / LNAV_BIT_PERIOD_S))
# load_sigs is edge-detected before gps_frame_ready. This two-PCLK
# delay affects interrupt latency but not the latched ABS_SAMPLE.
irq_time = event.receive_time_s + 2.0 / self.pps_hz
return SubframeInterrupt(
z_count=decoded.z_count,
subframe_id=decoded.subframe_id,
words=decoded.normalized_words,
bit_index_after_how=bit_index,
sample_time_s=event.sample_time_s,
irq_time_s=irq_time,
sample=event.sample,
)
@dataclass(frozen=True)
class CounterClock:
nominal_hz: float
epoch_time_s: float
oscillator_ppm: float = 0.0
@property
def physical_hz(self) -> float:
return self.nominal_hz * (1.0 + self.oscillator_ppm * 1.0e-6)
def first_tick_at_or_after(self, time_s: float) -> int:
exact = (time_s - self.epoch_time_s) * self.physical_hz
nearest = round(exact)
# Absolute GPS TOW is intentionally retained in the model. Subtracting
# two ~345 ks binary64 values can leave a few thousandths of one 8 ns
# tick even when the synthetic event was constructed exactly on that
# tick. Snap only this numerical residue; genuinely asynchronous
# events still use the following hardware clock edge.
if abs(exact - nearest) <= TICK_INTEGER_TOLERANCE:
exact = float(nearest)
return max(0, int(math.ceil(exact)))
def time_for_tick(self, tick: int) -> float:
return self.epoch_time_s + tick / self.physical_hz
@dataclass(frozen=True)
class NavSolveInterrupt:
sample_time_s: float
irq_time_s: float
bit_index_after_how: int
sample: SampleStamp
pps_counter: int
pps_capture_time_s: float
pps_capture_sequence: int
class NavSolveTimerModel:
"""Mirror the current timer: seed from Z*300, then count later bit pulses."""
def __init__(
self,
period_bits: int,
subframe: SubframeInterrupt,
clock: CounterClock,
*,
capture_pipeline_ticks: int,
):
if period_bits <= 0:
raise ValueError("navigation solve period must be positive")
self.period_bits = period_bits
self.counter = (subframe.z_count * LNAV_BITS_PER_SUBFRAME) % period_bits
self.start_sample_time_s = subframe.sample_time_s
self.how_boundary_s = subframe.z_count * 6.0
self.clock = clock
self.capture_pipeline_ticks = capture_pipeline_ticks
self.sequence = 0
def feed(self, event: BitEvent) -> NavSolveInterrupt | None:
if event.sample_time_s <= self.start_sample_time_s:
return None
reached = self.counter == self.period_bits - 1
self.counter = 0 if reached else self.counter + 1
if not reached:
return None
bit_tick = self.clock.first_tick_at_or_after(event.receive_time_s)
capture_tick = bit_tick + self.capture_pipeline_ticks
self.sequence += 1
return NavSolveInterrupt(
sample_time_s=event.sample_time_s,
irq_time_s=self.clock.time_for_tick(capture_tick + 1),
bit_index_after_how=int(
round((event.transmit_time_s - self.how_boundary_s) / LNAV_BIT_PERIOD_S)
),
sample=event.sample,
pps_counter=capture_tick,
pps_capture_time_s=self.clock.time_for_tick(capture_tick),
pps_capture_sequence=self.sequence,
)
@dataclass(frozen=True)
class FirmwareReference:
nav_bit_index_after_how: int
periodic_elapsed_s: float
transmit_tow_s: float
gps_time_at_capture_s: float
gps_time_at_capture_q32_s: float
true_capture_time_s: float
signed_reference_error_s: float
epoch_counter: int
class FirmwareTimingModel:
"""Mirror HOW sample anchoring and the current PPS receiver-time equation."""
def __init__(
self,
*,
nav_solve_period_bits: int,
postpone_tow_s: float,
propagation_s: float,
):
self.event_period_s = nav_solve_period_bits * LNAV_BIT_PERIOD_S
self.postpone_tow_s = postpone_tow_s
self.propagation_s = propagation_s
# In this project's WLS datum STARTOFFSET - b/c recovers flight time.
self.synthetic_clock_bias_m = (START_OFFSET_S - propagation_s) * C
self.anchor: SubframeInterrupt | None = None
self.rejected_nav_solves = 0
def on_subframe(self, event: SubframeInterrupt) -> None:
self.anchor = event
@staticmethod
def _signed_u32_delta(current: int, previous: int) -> int:
delta = (current - previous) & 0xFFFF_FFFF
return delta - (1 << 32) if delta & 0x8000_0000 else delta
def on_nav_solve(self, event: NavSolveInterrupt) -> FirmwareReference | None:
if self.anchor is None:
raise RuntimeError("navigation solve arrived before a valid HOW anchor")
raw_delta = self._signed_u32_delta(event.sample.raw, self.anchor.sample.raw)
anchor_fraction = self.anchor.sample.corrected - self.anchor.sample.raw
event_fraction = event.sample.corrected - event.sample.raw
elapsed_s = (raw_delta + event_fraction - anchor_fraction) / RF_HZ
event_index = int(math.floor(elapsed_s / self.event_period_s + 0.5))
periodic_elapsed_s = event_index * self.event_period_s
if abs(periodic_elapsed_s - elapsed_s) > NAV_TOW_SNAP_TOLERANCE_S:
# gps_time_from_periodic_sample_anchor() returns false and the
# interrupt handler drops this otherwise coherent NAV solve.
self.rejected_nav_solves += 1
return None
transmit_tow = self.anchor.z_count * 6.0 + periodic_elapsed_s
gps_reference = (
transmit_tow
+ self.postpone_tow_s
+ START_OFFSET_S
- self.synthetic_clock_bias_m / C
)
whole = math.floor(gps_reference)
fraction_q32 = int(math.floor((gps_reference - whole) * Q32 + 0.5))
if fraction_q32 == Q32:
whole += 1
fraction_q32 = 0
gps_reference_q32 = whole + fraction_q32 / Q32
return FirmwareReference(
nav_bit_index_after_how=event.bit_index_after_how,
periodic_elapsed_s=periodic_elapsed_s,
transmit_tow_s=transmit_tow,
gps_time_at_capture_s=gps_reference,
gps_time_at_capture_q32_s=gps_reference_q32,
true_capture_time_s=event.pps_capture_time_s,
signed_reference_error_s=gps_reference_q32 - event.pps_capture_time_s,
epoch_counter=event.pps_counter,
)
@dataclass(frozen=True)
class PpsEdge:
gps_second: int
dut_edge_time_s: float
trusted_edge_time_s: float
dut_minus_trusted_s: float
@dataclass(frozen=True)
class PpsCoreResult:
normal_commits_to_lock: int
lock_time_s: float
commit_age_s: float
edges: tuple[PpsEdge, ...]
def simulate_pps_core(
references: Sequence[FirmwareReference],
clock: CounterClock,
*,
compute_delay_s: float,
core_pipeline_ticks: int,
lock_updates: int,
dut_output_delay_s: float,
trusted_output_delay_s: float,
) -> PpsCoreResult:
"""Model timestamped coarse/normal commits and nominal locked PPS edges."""
if len(references) < lock_updates + 1:
raise RuntimeError("not enough navigation references to acquire PPS lock")
coarse = references[0]
mapping_counter = coarse.epoch_counter
mapping_time = coarse.gps_time_at_capture_q32_s
good_updates = 0
lock_reference: FirmwareReference | None = None
for reference in references[1:]:
mapped_at_capture = mapping_time + (
reference.epoch_counter - mapping_counter
) / clock.nominal_hz
phase_error = reference.gps_time_at_capture_q32_s - mapped_at_capture
if abs(phase_error) <= 8.0 / clock.nominal_hz:
good_updates += 1
else:
good_updates = 0
if good_updates >= lock_updates:
lock_reference = reference
break
if lock_reference is None:
raise RuntimeError("PPS loop did not acquire lock")
commit_time = lock_reference.true_capture_time_s + compute_delay_s
commit_tick = clock.first_tick_at_or_after(commit_time) + core_pipeline_ticks
lock_time = clock.time_for_tick(commit_tick)
commit_age_s = lock_time - lock_reference.true_capture_time_s
lock_counter = commit_tick
mapped_at_lock = mapping_time + (lock_counter - mapping_counter) / clock.nominal_hz
edges: list[PpsEdge] = []
target_second = int(math.floor(mapped_at_lock)) + 1
for _ in range(3):
ticks_exact = (target_second - mapping_time) * clock.nominal_hz
nearest_tick = round(ticks_exact)
if abs(ticks_exact - nearest_tick) <= TICK_INTEGER_TOLERANCE:
ticks_exact = float(nearest_tick)
ticks_from_mapping = int(math.ceil(ticks_exact))
edge_tick = mapping_counter + ticks_from_mapping
dut_edge = clock.time_for_tick(edge_tick) + dut_output_delay_s
trusted_edge = target_second + trusted_output_delay_s
edges.append(
PpsEdge(
gps_second=target_second,
dut_edge_time_s=dut_edge,
trusted_edge_time_s=trusted_edge,
dut_minus_trusted_s=dut_edge - trusted_edge,
)
)
target_second += 1
return PpsCoreResult(
normal_commits_to_lock=good_updates,
lock_time_s=lock_time,
commit_age_s=commit_age_s,
edges=tuple(edges),
)
@dataclass(frozen=True)
class EndToEndConfig:
decoder_mode: DecoderMode = DecoderMode.CURRENT_RTL
postpone_tow_s: float = 0.160
nav_solve_period_bits: int = 10
propagation_s: float = 0.070
pps_capture_pipeline_ticks: int = 4
firmware_compute_delay_s: float = 0.010
core_pipeline_ticks: int = 12
dut_output_delay_s: float = 0.0
trusted_output_delay_s: float = 0.0
first_z_count: int = 57_600
subframes: int = 3
# Put the first HOW capture just before a modulo-32 wrap so the first
# periodic solve also exercises the firmware's signed unwrap rule.
rf_initial_sample: int = (
(1 << 32) - int(10.25 * RF_HZ) - 1_000
)
@dataclass(frozen=True)
class EndToEndResult:
decoder_mode: str
postpone_tow_ms: float
bit_valid_events: int
bit_generator_sample_lag_ms: float
subframe_interrupts: int
nav_solve_interrupts: int
firmware_rejected_nav_solves: int
first_subframe_readout_bits: int
first_nav_solve_bit: int
first_reference_error_ms: float
expected_semantic_offset_ms: float
pps: PpsCoreResult
def _sample_at(time_s: float, epoch_s: float, initial_sample: int) -> SampleStamp:
exact = initial_sample + (time_s - epoch_s) * RF_HZ
raw_unwrapped = int(math.ceil(exact - 1.0e-10))
fractional_delay = raw_unwrapped - exact
raw = raw_unwrapped & 0xFFFF_FFFF
# This is the firmware representation: corrected_abs is near the wrapped
# integer ABS_SAMPLE, not a separately unwrapped floating-point counter.
return SampleStamp(raw=raw, corrected=raw - fractional_delay)
def _build_stream(config: EndToEndConfig) -> tuple[tuple[int, ...], float]:
d29, d30 = 0, 0
stream: list[int] = [d29, d30]
for offset in range(config.subframes):
subframe = encode_lnav_subframe(
config.first_z_count + offset,
offset % 5 + 1,
d29,
d30,
)
stream.extend(subframe.transmitted_bits)
d29, d30 = subframe.ending_d29, subframe.ending_d30
return tuple(stream), config.first_z_count * 6.0
def _generate_bit_events(
stream: Sequence[int],
*,
first_how_boundary_s: float,
propagation_s: float,
counter_epoch_s: float,
rf_initial_sample: int,
) -> tuple[BitEvent, ...]:
"""Expand LNAV bits to the 1 kHz prompt-sign input consumed by bit_gen."""
first_bit_event_tx = first_how_boundary_s - (
LNAV_SUBFRAME_BITS + 1
) * LNAV_BIT_PERIOD_S
first_prompt_receive = (
first_bit_event_tx
+ propagation_s
- LNAV_BIT_PERIOD_S
)
prompt_values = (
[1] * TRACKING_INTEGRATIONS_PER_BIT
+ [0] * TRACKING_INTEGRATIONS_PER_BIT
)
for value in stream:
prompt_values.extend([int(value)] * TRACKING_INTEGRATIONS_PER_BIT)
# The first integration of the following bit makes loop_counter_full true
# and emits the final requested bit, just as in bit_gen.sv.
prompt_values.append(0)
training_duration_s = 2 * LNAV_BIT_PERIOD_S
prompt_start_s = first_prompt_receive - training_duration_s
generator = BitGeneratorModel(propagation_s=propagation_s)
events: list[BitEvent] = []
for prompt_index, value in enumerate(prompt_values):
receive_time = (
prompt_start_s
+ prompt_index * TRACKING_INTEGRATION_PERIOD_S
)
output = generator.feed(
PromptSample(
value=value,
receive_time_s=receive_time,
sample=_sample_at(
receive_time,
counter_epoch_s,
rf_initial_sample,
),
)
)
if output is not None:
events.append(output)
if tuple(event.value for event in events) != tuple(stream):
raise RuntimeError("bit_gen model did not reproduce the requested LNAV stream")
return tuple(events)
def run_end_to_end(config: EndToEndConfig) -> EndToEndResult:
if config.subframes < 2:
raise ValueError("at least two subframes are required for preamble validation")
stream, first_how_boundary = _build_stream(config)
counter_epoch = first_how_boundary - 10.0
# Oscillator/PI behavior is already covered by pps_loop/adaptive_pps. This
# absolute-epoch regression uses an exact clock so a control-loop transient
# cannot hide or imitate an integer navigation-bit error.
clock = CounterClock(PPS_HZ, counter_epoch)
bit_events = _generate_bit_events(
stream,
first_how_boundary_s=first_how_boundary,
propagation_s=config.propagation_s,
counter_epoch_s=counter_epoch,
rf_initial_sample=config.rf_initial_sample,
)
decoder = TrackingDecoderModel(config.decoder_mode)
firmware = FirmwareTimingModel(
nav_solve_period_bits=config.nav_solve_period_bits,
postpone_tow_s=config.postpone_tow_s,
propagation_s=config.propagation_s,
)
timer: NavSolveTimerModel | None = None
subframe_events: list[SubframeInterrupt] = []
nav_events: list[NavSolveInterrupt] = []
references: list[FirmwareReference] = []
for bit_event in bit_events:
# An already-running timer sees the bit even when a later subframe is
# validated on the same event. Firmware handles NAV_BITS before
# NAV_SOLVE when both status flags share one interrupt.
nav_event = timer.feed(bit_event) if timer is not None else None
subframe_event = decoder.feed(bit_event)
if subframe_event is not None:
subframe_events.append(subframe_event)
firmware.on_subframe(subframe_event)
if timer is None:
timer = NavSolveTimerModel(
config.nav_solve_period_bits,
subframe_event,
clock,
capture_pipeline_ticks=config.pps_capture_pipeline_ticks,
)
if nav_event is not None:
nav_events.append(nav_event)
reference = firmware.on_nav_solve(nav_event)
if reference is not None:
references.append(reference)
if not subframe_events or not nav_events or not references:
raise RuntimeError("synthetic stream did not reach the PPS timing path")
pps = simulate_pps_core(
references,
clock,
compute_delay_s=config.firmware_compute_delay_s,
core_pipeline_ticks=config.core_pipeline_ticks,
lock_updates=5,
dut_output_delay_s=config.dut_output_delay_s,
trusted_output_delay_s=config.trusted_output_delay_s,
)
return EndToEndResult(
decoder_mode=config.decoder_mode.value,
postpone_tow_ms=config.postpone_tow_s * 1.0e3,
bit_valid_events=len(bit_events),
bit_generator_sample_lag_ms=(
bit_events[0].receive_time_s - bit_events[0].sample_time_s
)
* 1.0e3,
subframe_interrupts=len(subframe_events),
nav_solve_interrupts=len(nav_events),
firmware_rejected_nav_solves=firmware.rejected_nav_solves,
first_subframe_readout_bits=subframe_events[0].bit_index_after_how,
first_nav_solve_bit=nav_events[0].bit_index_after_how,
first_reference_error_ms=references[0].signed_reference_error_s * 1.0e3,
expected_semantic_offset_ms=(
subframe_events[0].bit_index_after_how * LNAV_BIT_PERIOD_S
- config.postpone_tow_s
)
* 1.0e3,
pps=pps,
)
def default_comparison() -> tuple[EndToEndResult, ...]:
return (
run_end_to_end(EndToEndConfig()),
run_end_to_end(
EndToEndConfig(
decoder_mode=DecoderMode.CURRENT_BIT,
postpone_tow_s=0.160,
)
),
run_end_to_end(
EndToEndConfig(
decoder_mode=DecoderMode.CURRENT_RTL,
postpone_tow_s=0.180,
)
),
)
def _result_dict(result: EndToEndResult) -> dict[str, object]:
value = asdict(result)
value["pps"]["edges"] = [asdict(edge) for edge in result.pps.edges]
return value
def _print_table(results: Sequence[EndToEndResult]) -> None:
print(
"scenario bits correction reference error "
"DUT - trusted PPS"
)
for result in results:
offset_ms = result.pps.edges[0].dut_minus_trusted_s * 1.0e3
scenario = f"{result.decoder_mode}/{result.postpone_tow_ms:.0f}ms"
print(
f"{scenario:<25} {result.first_subframe_readout_bits:>4d} "
f"{result.postpone_tow_ms:>7.0f} ms "
f"{result.first_reference_error_ms:>+12.6f} ms "
f"{offset_ms:>+13.6f} ms"
)
print("\nPositive DUT - trusted means that the project PPS edge is late.")
print(
"The residual tens of nanoseconds are the configured PCLK capture "
"pipeline, not an LNAV bit."
)
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--mode",
choices=("matrix", DecoderMode.CURRENT_RTL.value, DecoderMode.CURRENT_BIT.value),
default="matrix",
help="run the three-way diagnosis matrix or one decoder mode",
)
parser.add_argument("--postpone-ms", type=float, default=160.0)
parser.add_argument("--propagation-ms", type=float, default=70.0)
parser.add_argument("--capture-pipeline-ticks", type=int, default=4)
parser.add_argument("--compute-delay-ms", type=float, default=10.0)
parser.add_argument("--dut-output-delay-ns", type=float, default=0.0)
parser.add_argument("--trusted-output-delay-ns", type=float, default=0.0)
parser.add_argument("--json", action="store_true")
args = parser.parse_args(argv)
if args.mode == "matrix":
results = default_comparison()
else:
results = (
run_end_to_end(
EndToEndConfig(
decoder_mode=DecoderMode(args.mode),
postpone_tow_s=args.postpone_ms * 1.0e-3,
propagation_s=args.propagation_ms * 1.0e-3,
pps_capture_pipeline_ticks=args.capture_pipeline_ticks,
firmware_compute_delay_s=args.compute_delay_ms * 1.0e-3,
dut_output_delay_s=args.dut_output_delay_ns * 1.0e-9,
trusted_output_delay_s=args.trusted_output_delay_ns * 1.0e-9,
)
),
)
if args.json:
print(json.dumps([_result_dict(result) for result in results], indent=2))
else:
_print_table(results)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,137 @@
from __future__ import annotations
import unittest
from pps_end_to_end_validation.pps_end_to_end import (
DecoderMode,
EndToEndConfig,
LNAV_WINDOW_BITS,
encode_lnav_subframe,
run_end_to_end,
validate_rtl_window,
)
class LnavDecoderTests(unittest.TestCase):
def make_window(self):
previous_d29 = 0
previous_d30 = 0
current = encode_lnav_subframe(
57_600, 1, previous_d29, previous_d30
)
following = encode_lnav_subframe(
57_601,
2,
current.ending_d29,
current.ending_d30,
)
window = (
previous_d29,
previous_d30,
*current.transmitted_bits,
*following.transmitted_bits[:8],
)
self.assertEqual(len(window), LNAV_WINDOW_BITS)
return current, window
def test_parity_valid_subframe_round_trips_through_rtl_predicate(self):
current, window = self.make_window()
decoded = validate_rtl_window(window)
self.assertIsNotNone(decoded)
self.assertEqual(decoded.z_count, current.z_count)
self.assertEqual(decoded.subframe_id, current.subframe_id)
self.assertEqual(decoded.normalized_words, current.normalized_words)
def test_single_data_bit_corruption_prevents_subframe_interrupt(self):
_, window = self.make_window()
corrupted = list(window)
corrupted[2 + 73] ^= 1
self.assertIsNone(validate_rtl_window(corrupted))
class PpsEndToEndTests(unittest.TestCase):
def test_bit_generator_recreates_50bps_stream_and_registered_sample_lag(self):
result = run_end_to_end(EndToEndConfig())
self.assertEqual(result.bit_valid_events, 2 + 3 * 300)
self.assertAlmostEqual(result.bit_generator_sample_lag_ms, 1.0, places=6)
def test_current_rtl_validates_on_ninth_following_bit(self):
result = run_end_to_end(EndToEndConfig())
self.assertEqual(result.first_subframe_readout_bits, 9)
self.assertEqual(result.first_nav_solve_bit, 19)
self.assertGreaterEqual(result.subframe_interrupts, 2)
self.assertGreater(result.nav_solve_interrupts, 20)
self.assertEqual(result.firmware_rejected_nav_solves, 0)
def test_permitted_40_bit_period_rejects_after_how_anchor_moves(self):
# Firmware's compile-time rule accepts 40 because it divides the GPS
# week. It does not divide a 300-bit subframe, so refreshing the HOW
# sample anchor changes timer phase by half a solve period. The
# current N*20 ms snap then rejects otherwise valid capture events.
self.assertEqual(30_240_000 % 40, 0)
self.assertNotEqual(300 % 40, 0)
result = run_end_to_end(
EndToEndConfig(nav_solve_period_bits=40, subframes=4)
)
self.assertGreater(result.firmware_rejected_nav_solves, 0)
def test_current_rtl_and_160ms_correction_make_pps_20ms_late(self):
result = run_end_to_end(EndToEndConfig())
offset_s = result.pps.edges[0].dut_minus_trusted_s
self.assertAlmostEqual(result.expected_semantic_offset_ms, 20.0, places=9)
self.assertGreater(offset_s, 0.019999)
self.assertLess(offset_s, 0.020001)
for edge in result.pps.edges:
self.assertAlmostEqual(edge.dut_minus_trusted_s, offset_s, places=9)
def test_current_bit_validation_and_160ms_remove_lnav_bit_error(self):
result = run_end_to_end(
EndToEndConfig(decoder_mode=DecoderMode.CURRENT_BIT)
)
self.assertEqual(result.first_subframe_readout_bits, 8)
self.assertEqual(result.first_nav_solve_bit, 18)
self.assertAlmostEqual(result.expected_semantic_offset_ms, 0.0, places=9)
self.assertLess(abs(result.pps.edges[0].dut_minus_trusted_s), 100.0e-9)
def test_180ms_firmware_workaround_matches_current_rtl(self):
result = run_end_to_end(
EndToEndConfig(postpone_tow_s=0.180)
)
self.assertEqual(result.first_subframe_readout_bits, 9)
self.assertAlmostEqual(result.expected_semantic_offset_ms, 0.0, places=9)
self.assertLess(abs(result.pps.edges[0].dut_minus_trusted_s), 100.0e-9)
def test_timestamped_commit_rejects_cpu_delay_as_phase(self):
fast = run_end_to_end(
EndToEndConfig(firmware_compute_delay_s=0.001)
)
stalled = run_end_to_end(
EndToEndConfig(firmware_compute_delay_s=0.080)
)
self.assertAlmostEqual(
fast.pps.edges[0].dut_minus_trusted_s,
stalled.pps.edges[0].dut_minus_trusted_s,
places=9,
)
self.assertGreater(stalled.pps.commit_age_s, fast.pps.commit_age_s)
def test_vendor_and_dut_fixed_delays_have_expected_sign(self):
result = run_end_to_end(
EndToEndConfig(
decoder_mode=DecoderMode.CURRENT_BIT,
dut_output_delay_s=12.0e-9,
trusted_output_delay_s=35.0e-9,
)
)
baseline = run_end_to_end(
EndToEndConfig(decoder_mode=DecoderMode.CURRENT_BIT)
)
delta = (
result.pps.edges[0].dut_minus_trusted_s
- baseline.pps.edges[0].dut_minus_trusted_s
)
self.assertAlmostEqual(delta, -23.0e-9, places=10)
if __name__ == "__main__":
unittest.main()
+22
View File
@@ -59,6 +59,28 @@ python3 pps_python/simulate_adaptive_pps.py
python3 -m unittest pps_python/test_pps_adaptive.py
```
## Bit-level Linux PPS uçtan uca testi
Bit üretiminden trusted PPS karşılaştırmasına kadar deterministik mutlak zaman
regresyonu ayrı `pps_end_to_end_validation` paketine taşınmıştır. Ayrıntılı
kapsam, 20 ms teşhisi ve donanım doğrulama adımları için
[`pps_end_to_end_validation/README.md`](../pps_end_to_end_validation/README.md)
dosyasına bakın.
Software dizininden çalıştırma:
```bash
python3 -m pps_end_to_end_validation.pps_end_to_end
python3 -m unittest pps_end_to_end_validation/test_pps_end_to_end.py
```
```text
scenario bits correction reference error DUT - trusted PPS
current-rtl/160ms 9 160 ms -20.000032 ms +20.000032 ms
current-bit/160ms 8 160 ms -0.000032 ms +0.000032 ms
current-rtl/180ms 9 180 ms -0.000032 ms +0.000032 ms
```
Kısa veya farklı PI koşusu:
```bash