109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Monte-Carlo sweep for noisy/missing WLS adaptive PPS scenarios."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import random
|
|
import statistics
|
|
|
|
try:
|
|
from .adaptive_pps import Observation, Profile, QualityMetrics, run_observations
|
|
except ImportError:
|
|
from adaptive_pps import Observation, Profile, QualityMetrics, run_observations
|
|
|
|
|
|
def metrics(profile: Profile) -> QualityMetrics:
|
|
if profile == Profile.PRECISION:
|
|
return QualityMetrics(1.7, 3.2, 18.0, 38.0, 27.0, 0.35, 0.25, 7)
|
|
if profile == Profile.ROBUST:
|
|
return QualityMetrics(5.0, 8.0, 65.0, 27.0, 18.0, 0.85, 1.8, 5)
|
|
return QualityMetrics(12.0, 18.0, 180.0, 18.0, 10.0, 2.0, 4.0, 4)
|
|
|
|
|
|
def scenario(seed: int, duration_s: float = 180.0) -> list[Observation]:
|
|
rng = random.Random(seed)
|
|
result = []
|
|
for update in range(round(duration_s / 0.2)):
|
|
time_s = update * 0.2
|
|
if 70.0 <= time_s < 82.0:
|
|
continue
|
|
profile = Profile.PRECISION
|
|
if 35.0 <= time_s < 45.0 or 110.0 <= time_s < 125.0:
|
|
profile = Profile.ROBUST
|
|
noise_sigma = 320.0e-9 if profile == Profile.PRECISION else 650.0e-9
|
|
noise_s = rng.gauss(0.0, noise_sigma)
|
|
if update % 41 == 0:
|
|
noise_s += rng.choice((-1.0, 1.0)) * 900.0e-9
|
|
result.append(
|
|
Observation(
|
|
time_s,
|
|
noise_s,
|
|
rng.gauss(0.0, 0.3),
|
|
metrics(profile),
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def summarize(name: str, traces) -> None:
|
|
first_locks = [
|
|
trace.first_lock_time()
|
|
for trace in traces
|
|
if trace.first_lock_time() is not None
|
|
]
|
|
tail_rms_ns = []
|
|
jitter_rms_ns = []
|
|
for trace in traces:
|
|
tail = trace.phase_error_s[len(trace.phase_error_s) // 2 :]
|
|
if tail:
|
|
tail_rms_ns.append(
|
|
math.sqrt(sum(value * value for value in tail) / len(tail))
|
|
* 1.0e9
|
|
)
|
|
jitter = trace.one_second_jitter_rms(after_s=90.0)
|
|
if math.isfinite(jitter):
|
|
jitter_rms_ns.append(jitter * 1.0e9)
|
|
first_lock_p95 = (
|
|
statistics.quantiles(first_locks, n=20)[18]
|
|
if len(first_locks) >= 2
|
|
else (first_locks[0] if first_locks else math.nan)
|
|
)
|
|
tail_rms_median = (
|
|
statistics.median(tail_rms_ns) if tail_rms_ns else math.nan
|
|
)
|
|
jitter_rms_median = (
|
|
statistics.median(jitter_rms_ns) if jitter_rms_ns else math.nan
|
|
)
|
|
print(
|
|
f"{name}: lock_success={len(first_locks)}/{len(traces)} "
|
|
f"first_lock_p95={first_lock_p95:.2f}s "
|
|
f"tail_rms_median={tail_rms_median:.2f}ns "
|
|
f"one_second_jitter_median={jitter_rms_median:.2f}ns "
|
|
f"unlock_total={sum(trace.unlock_events for trace in traces)}"
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
seeds = range(1, 51)
|
|
raw = [
|
|
run_observations(
|
|
scenario(seed),
|
|
finish_time_s=180.0,
|
|
adaptive=False,
|
|
software_filter=False,
|
|
)
|
|
for seed in seeds
|
|
]
|
|
adaptive = [
|
|
run_observations(scenario(seed), finish_time_s=180.0)
|
|
for seed in seeds
|
|
]
|
|
summarize("raw fixed PI", raw)
|
|
summarize("adaptive fixed-point PPS", adaptive)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|