Align `test` to `reference` and score the match. Cross-correlates to recover the capture latency, trims both signals to their overlap, and reports SNR and Pearson correlation over that overlap. SNR is amplitude-invariant here because both signals are normalized to unit RMS before th
(reference, test)
| 88 | |
| 89 | |
| 90 | def align_and_score(reference, test): |
| 91 | """Align `test` to `reference` and score the match. |
| 92 | |
| 93 | Cross-correlates to recover the capture latency, trims both signals to |
| 94 | their overlap, and reports SNR and Pearson correlation over that |
| 95 | overlap. SNR is amplitude-invariant here because both signals are |
| 96 | normalized to unit RMS before the residual is taken. |
| 97 | |
| 98 | Returns a dict: lag, overlap, snr_db, correlation. |
| 99 | """ |
| 100 | ref = _normalize(np.asarray(reference, dtype=np.float64)) |
| 101 | tst = _normalize(np.asarray(test, dtype=np.float64)) |
| 102 | if ref.size == 0 or tst.size == 0: |
| 103 | return {"lag": 0, "overlap": 0, "snr_db": -np.inf, "correlation": 0.0} |
| 104 | |
| 105 | lag = _best_lag(ref, tst) |
| 106 | |
| 107 | if lag >= 0: |
| 108 | ref_a = ref[lag:] |
| 109 | tst_a = tst[: ref_a.size] |
| 110 | else: |
| 111 | tst_a = tst[-lag:] |
| 112 | ref_a = ref[: tst_a.size] |
| 113 | |
| 114 | overlap = min(ref_a.size, tst_a.size) |
| 115 | ref_a = ref_a[:overlap] |
| 116 | tst_a = tst_a[:overlap] |
| 117 | if overlap == 0: |
| 118 | return {"lag": lag, "overlap": 0, "snr_db": -np.inf, "correlation": 0.0} |
| 119 | |
| 120 | # Re-normalize on the overlap so the score is not diluted by trimmed tails |
| 121 | ref_a = _normalize(ref_a) |
| 122 | tst_a = _normalize(tst_a) |
| 123 | |
| 124 | residual = ref_a - tst_a |
| 125 | signal_power = float(np.sum(ref_a * ref_a)) |
| 126 | noise_power = float(np.sum(residual * residual)) |
| 127 | snr_db = 10.0 * np.log10(signal_power / noise_power) if noise_power > 0 else np.inf |
| 128 | |
| 129 | denom = np.sqrt(np.sum(ref_a * ref_a) * np.sum(tst_a * tst_a)) |
| 130 | correlation = float(np.sum(ref_a * tst_a) / denom) if denom > 0 else 0.0 |
| 131 | |
| 132 | return { |
| 133 | "lag": lag, |
| 134 | "overlap": overlap, |
| 135 | "snr_db": float(snr_db), |
| 136 | "correlation": correlation, |
| 137 | } |
| 138 | |
| 139 | |
| 140 | def analyze_cadence(elapsed, expected_rate): |