920 lines
37 KiB
Rust
920 lines
37 KiB
Rust
//! GPS pipeline: exiftool track extraction (background sweep), shared track math,
|
||
//! and the Stage-1 "Viterbi" smoother with its preview/keep/discard endpoint.
|
||
//!
|
||
//! Track representation everywhere: `[[lat, lon, t_s], …]` (t_s = seconds since the
|
||
//! first GPS sample). The RAW extracted track (`videos.gps_track`) is immutable once
|
||
//! scanned; a kept correction lives in `gps_track_corrected` and is preferred by the
|
||
//! map + chainage. Raw is never modified — discarding a correction restores raw.
|
||
//!
|
||
//! Lifecycle (`videos.gps_status`): 'pending' → 'ok' | 'none' (no GPS metadata) |
|
||
//! 'error' (exiftool/mount failure, message in `gps_error`). The sweep processes ONE
|
||
//! video at a time (exiftool -ee reads the whole file over NFS — never hammer the NAS).
|
||
|
||
use crate::auth::AuthUser;
|
||
use crate::{err, ApiResult, AppState};
|
||
use axum::{
|
||
extract::{Path as AxPath, State},
|
||
http::StatusCode,
|
||
Json,
|
||
};
|
||
use serde::Deserialize;
|
||
use serde_json::{json, Value};
|
||
use std::collections::BTreeMap;
|
||
use std::time::Duration;
|
||
|
||
/// One GPS sample: [lat, lon, t_s].
|
||
pub type Pt = [f64; 3];
|
||
|
||
/// Max samples stored per track (1 Hz ⇒ ~83 min of video at full fidelity).
|
||
const MAX_TRACK_PTS: usize = 5000;
|
||
/// Hard deadline for one exiftool scan (-ee reads the whole file over NFS).
|
||
const EXIFTOOL_TIMEOUT: Duration = Duration::from_secs(600);
|
||
/// Sweep pacing: idle poll when nothing is pending / pause between videos / back-off
|
||
/// after a failure (so an unreachable NAS isn't probed in a hot loop).
|
||
const SWEEP_IDLE: Duration = Duration::from_secs(60);
|
||
const SWEEP_BETWEEN: Duration = Duration::from_millis(500);
|
||
const SWEEP_AFTER_ERROR: Duration = Duration::from_secs(5);
|
||
|
||
// ---- track math (shared with chainage) ----
|
||
|
||
/// WGS84 ellipsoid: semi-major axis + first eccentricity squared.
|
||
const WGS84_A: f64 = 6_378_137.0;
|
||
const WGS84_E2: f64 = 0.006_694_379_990_14;
|
||
|
||
/// Meters per degree of latitude / longitude at a given latitude, from the WGS84
|
||
/// meridian (M) and prime-vertical (N) radii. Distance and snapping both use this,
|
||
/// so every meter in the system is measured the same way.
|
||
pub fn wgs84_m_per_deg(lat_deg: f64) -> (f64, f64) {
|
||
let phi = lat_deg.to_radians();
|
||
let w = (1.0 - WGS84_E2 * phi.sin().powi(2)).sqrt();
|
||
let m = WGS84_A * (1.0 - WGS84_E2) / (w * w * w);
|
||
let n = WGS84_A / w;
|
||
let rad = std::f64::consts::PI / 180.0;
|
||
(m * rad, n * phi.cos() * rad)
|
||
}
|
||
|
||
/// WGS84 local-ellipsoid distance between two nearby points. Essentially exact at
|
||
/// GPS sample spacing (meters–tens of meters) and <2e-5 relative for spans up to
|
||
/// ~100 km — unlike spherical haversine, whose 0.1–0.3% systematic bias would land
|
||
/// directly in sr_chainage. (Not valid across the ±180° meridian — irrelevant here.)
|
||
pub fn dist_m(lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 {
|
||
let (mlat, mlon) = wgs84_m_per_deg((lat1 + lat2) * 0.5);
|
||
let dy = (lat2 - lat1) * mlat;
|
||
let dx = (lon2 - lon1) * mlon;
|
||
(dx * dx + dy * dy).sqrt()
|
||
}
|
||
|
||
pub fn polyline_len_m(track: &[Pt]) -> f64 {
|
||
track
|
||
.windows(2)
|
||
.map(|w| dist_m(w[0][0], w[0][1], w[1][0], w[1][1]))
|
||
.sum()
|
||
}
|
||
|
||
/// Parse a stored JSONB track (`[[lat,lon,t_s],…]`) back into points. Tolerant:
|
||
/// skips malformed entries.
|
||
pub fn parse_track(v: &Value) -> Vec<Pt> {
|
||
v.as_array()
|
||
.map(|a| {
|
||
a.iter()
|
||
.filter_map(|p| {
|
||
let p = p.as_array()?;
|
||
Some([p.first()?.as_f64()?, p.get(1)?.as_f64()?, p.get(2).and_then(|t| t.as_f64()).unwrap_or(0.0)])
|
||
})
|
||
.collect()
|
||
})
|
||
.unwrap_or_default()
|
||
}
|
||
|
||
/// Serialize a track for storage/transport, rounding to keep the JSON small
|
||
/// (1e-6 deg ≈ 0.1 m; 0.1 s time resolution).
|
||
pub fn track_value(track: &[Pt]) -> Value {
|
||
Value::Array(
|
||
track
|
||
.iter()
|
||
.map(|p| {
|
||
json!([
|
||
(p[0] * 1e6).round() / 1e6,
|
||
(p[1] * 1e6).round() / 1e6,
|
||
(p[2] * 10.0).round() / 10.0
|
||
])
|
||
})
|
||
.collect(),
|
||
)
|
||
}
|
||
|
||
/// Average + maximum speed over a track (m/s), from the full-resolution samples.
|
||
/// Max is measured over single hops (dt clamped ≥ 0.5 s so duplicate timestamps
|
||
/// can't fabricate teleports). None when the track has no elapsed time.
|
||
pub fn speed_stats(track: &[Pt]) -> (Option<f64>, Option<f64>) {
|
||
if track.len() < 2 {
|
||
return (None, None);
|
||
}
|
||
let elapsed = track.last().unwrap()[2] - track[0][2];
|
||
let avg = (elapsed > 0.0).then(|| polyline_len_m(track) / elapsed);
|
||
let max = track
|
||
.windows(2)
|
||
.map(|w| dist_m(w[0][0], w[0][1], w[1][0], w[1][1]) / (w[1][2] - w[0][2]).max(0.5))
|
||
.fold(None::<f64>, |acc, v| Some(acc.map_or(v, |a| a.max(v))));
|
||
(avg, max)
|
||
}
|
||
|
||
/// Cap a track at `max` points (uniform stride, always keeping the last point).
|
||
pub fn downsample(track: Vec<Pt>, max: usize) -> Vec<Pt> {
|
||
if track.len() <= max || max < 2 {
|
||
return track;
|
||
}
|
||
let stride = track.len().div_ceil(max);
|
||
let last = *track.last().unwrap();
|
||
let mut out: Vec<Pt> = track.into_iter().step_by(stride).collect();
|
||
if out.last() != Some(&last) {
|
||
out.push(last);
|
||
}
|
||
out
|
||
}
|
||
|
||
// ---- exiftool extraction ----
|
||
|
||
/// What one exiftool scan yields: the GPS samples (t_s relative to the first fix),
|
||
/// the first fix's absolute epoch (None when the camera wrote no GPSDateTime), the
|
||
/// container's media duration in seconds, and the video frame rate. Empty samples =
|
||
/// no GPS metadata. Duration + fps let us do frame→position math on raw NAS videos
|
||
/// that were never annotated (no sibling JSON ⇒ no fps in our DB otherwise).
|
||
struct ScanResult {
|
||
samples: Vec<Pt>,
|
||
t0_abs: Option<f64>,
|
||
duration_s: Option<f64>,
|
||
fps: Option<f64>,
|
||
}
|
||
|
||
/// Run exiftool over a video and collect its embedded GPS time-series (Doc{N}
|
||
/// groups, the format dashcams/DJI write at ~1 Hz). Same flags as the proven
|
||
/// desktop extractor, plus container Duration + VideoFrameRate.
|
||
async fn run_exiftool(path: &std::path::Path) -> Result<ScanResult, String> {
|
||
let child = tokio::process::Command::new("exiftool")
|
||
.args([
|
||
"-ee", "-api", "LargeFileSupport=1", "-G3", "-json", "-n",
|
||
"-GPSLatitude", "-GPSLongitude", "-GPSDateTime", "-Duration", "-VideoFrameRate",
|
||
])
|
||
.arg(path)
|
||
.stdout(std::process::Stdio::piped())
|
||
.stderr(std::process::Stdio::piped())
|
||
.kill_on_drop(true)
|
||
.spawn()
|
||
.map_err(|e| format!("failed to run exiftool (is it installed?): {e}"))?;
|
||
|
||
let output = match tokio::time::timeout(EXIFTOOL_TIMEOUT, child.wait_with_output()).await {
|
||
Ok(r) => r.map_err(|e| format!("exiftool failed: {e}"))?,
|
||
// kill_on_drop terminates the scan when the timeout drops the child.
|
||
Err(_) => return Err("exiftool timed out (file too slow to read?)".into()),
|
||
};
|
||
|
||
let stdout = String::from_utf8_lossy(&output.stdout);
|
||
let parsed: Value = serde_json::from_str(stdout.trim()).map_err(|_| {
|
||
let se = String::from_utf8_lossy(&output.stderr);
|
||
let msg: String = se.trim().chars().take(300).collect();
|
||
if msg.is_empty() {
|
||
"exiftool produced no output".to_string()
|
||
} else {
|
||
format!("exiftool: {msg}")
|
||
}
|
||
})?;
|
||
|
||
let obj = parsed
|
||
.as_array()
|
||
.and_then(|a| a.first())
|
||
.and_then(|v| v.as_object())
|
||
.ok_or_else(|| "exiftool output was not a non-empty array".to_string())?;
|
||
|
||
// exiftool reports per-file problems inline (e.g. "Error: File is empty").
|
||
if let Some((_, e)) = obj.iter().find(|(k, _)| *k == "Error" || k.ends_with(":Error")) {
|
||
if let Some(msg) = e.as_str() {
|
||
return Err(format!("exiftool: {msg}"));
|
||
}
|
||
}
|
||
|
||
// Numeric field across any -G3 group (container + per-track may each report one);
|
||
// `reduce` picks the max, which is the sensible choice for both Duration and fps.
|
||
let numeric = |name: &str| -> Option<f64> {
|
||
obj.iter()
|
||
.filter(|(k, _)| k.as_str() == name || k.ends_with(&format!(":{name}")))
|
||
.filter_map(|(_, v)| v.as_f64().or_else(|| v.as_str().and_then(|s| s.parse().ok())))
|
||
.reduce(f64::max)
|
||
};
|
||
let duration = numeric("Duration");
|
||
let fps = numeric("VideoFrameRate").filter(|f| *f > 0.0);
|
||
|
||
let (samples, t0_abs) = collect_samples(obj);
|
||
Ok(ScanResult { samples, t0_abs, duration_s: duration, fps })
|
||
}
|
||
|
||
/// Group `Doc{N}:GPSLatitude/GPSLongitude/GPSDateTime` keys into ordered samples
|
||
/// with relative timestamps (falls back to 1 Hz when GPSDateTime is missing).
|
||
/// Also returns the first fix's ABSOLUTE epoch — GPS time is UTC on every camera,
|
||
/// so consecutive files can be matched by timestamp (start-of-video stitching).
|
||
fn collect_samples(obj: &serde_json::Map<String, Value>) -> (Vec<Pt>, Option<f64>) {
|
||
#[derive(Default)]
|
||
struct Partial {
|
||
lat: Option<f64>,
|
||
lon: Option<f64>,
|
||
t: Option<f64>,
|
||
}
|
||
|
||
let mut by_doc: BTreeMap<u32, Partial> = BTreeMap::new();
|
||
for (k, v) in obj {
|
||
let Some((doc_idx, field)) = split_doc_key(k) else { continue };
|
||
let entry = by_doc.entry(doc_idx).or_default();
|
||
match field {
|
||
"GPSLatitude" => entry.lat = v.as_f64(),
|
||
"GPSLongitude" => entry.lon = v.as_f64(),
|
||
"GPSDateTime" => entry.t = v.as_str().and_then(parse_gps_datetime),
|
||
_ => {}
|
||
}
|
||
}
|
||
|
||
let mut samples: Vec<Pt> = Vec::with_capacity(by_doc.len());
|
||
let mut t0: Option<f64> = None;
|
||
let mut t0_abs: Option<f64> = None;
|
||
for (idx, p) in by_doc {
|
||
let (Some(lat), Some(lon)) = (p.lat, p.lon) else { continue };
|
||
if t0.is_none() {
|
||
t0_abs = p.t; // real epoch only when the camera stamped it
|
||
}
|
||
let t_abs = p.t.unwrap_or(idx as f64);
|
||
let t_rel = t_abs - *t0.get_or_insert(t_abs);
|
||
samples.push([lat, lon, t_rel]);
|
||
}
|
||
(samples, t0_abs)
|
||
}
|
||
|
||
fn split_doc_key(k: &str) -> Option<(u32, &str)> {
|
||
let (group, field) = k.split_once(':')?;
|
||
let idx: u32 = group.strip_prefix("Doc")?.parse().ok()?;
|
||
Some((idx, field))
|
||
}
|
||
|
||
/// "2025:08:02 09:33:31Z" → seconds since epoch (approx; only relative deltas are
|
||
/// used). Civil-to-days via Howard Hinnant's algorithm — no chrono parse needed for
|
||
/// exiftool's colon-separated dates.
|
||
fn parse_gps_datetime(s: &str) -> Option<f64> {
|
||
let s = s.trim_end_matches('Z').trim();
|
||
let (date, time) = s.split_once(' ')?;
|
||
let mut dp = date.split(':');
|
||
let y: i64 = dp.next()?.parse().ok()?;
|
||
let mo: i64 = dp.next()?.parse().ok()?;
|
||
let d: i64 = dp.next()?.parse().ok()?;
|
||
let mut tp = time.split(':');
|
||
let h: i64 = tp.next()?.parse().ok()?;
|
||
let mi: i64 = tp.next()?.parse().ok()?;
|
||
let sec: f64 = tp.next()?.parse().ok()?;
|
||
let y_adj = if mo <= 2 { y - 1 } else { y };
|
||
let era = y_adj.div_euclid(400);
|
||
let yoe = y_adj - era * 400;
|
||
let m = if mo > 2 { mo - 3 } else { mo + 9 };
|
||
let doy = (153 * m + 2) / 5 + d - 1;
|
||
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
|
||
let days = era * 146097 + doe - 719468;
|
||
Some(days as f64 * 86400.0 + (h * 3600 + mi * 60) as f64 + sec)
|
||
}
|
||
|
||
// ---- background sweep ----
|
||
|
||
/// Forever-loop: pick ONE pending video, scan it, store the result, repeat.
|
||
/// Serial by design (requirement: throttle — never scan the NAS in parallel).
|
||
pub async fn sweep(state: AppState) {
|
||
loop {
|
||
let next: Option<(i32, String, String, String)> = sqlx::query_as(
|
||
r#"SELECT v.id, v.rel_path, p.source_path, p.source_kind
|
||
FROM videos v JOIN projects p ON v.project_id = p.id
|
||
WHERE v.gps_status = 'pending'
|
||
ORDER BY v.id LIMIT 1"#,
|
||
)
|
||
.fetch_optional(&state.db)
|
||
.await
|
||
.unwrap_or(None);
|
||
|
||
let Some((id, rel_path, source_path, source_kind)) = next else {
|
||
tokio::time::sleep(SWEEP_IDLE).await;
|
||
continue;
|
||
};
|
||
|
||
let ok = scan_one(&state, id, &rel_path, &source_path, &source_kind).await;
|
||
tokio::time::sleep(if ok { SWEEP_BETWEEN } else { SWEEP_AFTER_ERROR }).await;
|
||
}
|
||
}
|
||
|
||
/// Scan a single video and persist the outcome. Returns false on error (the video
|
||
/// is marked 'error' either way, so the sweep never re-picks it until a re-scan).
|
||
async fn scan_one(state: &AppState, id: i32, rel_path: &str, source_path: &str, source_kind: &str) -> bool {
|
||
let (sp, kind, host, opts) =
|
||
(source_path.to_string(), source_kind.to_string(), state.nfs_host.clone(), state.nfs_opts.clone());
|
||
let dir = match tokio::task::spawn_blocking(move || crate::nfs::resolve_dir(&sp, &kind, &host, &opts)).await {
|
||
Ok(Ok(d)) => d,
|
||
Ok(Err(e)) => return mark_error(&state.db, id, &e).await,
|
||
Err(e) => return mark_error(&state.db, id, &format!("mount task failed: {e}")).await,
|
||
};
|
||
|
||
match run_exiftool(&dir.join(rel_path)).await {
|
||
Ok(ScanResult { samples, t0_abs, duration_s, fps }) if samples.len() >= 2 => {
|
||
// Data-quality stats are measured BEFORE the storage cap: the raw fix
|
||
// count and the largest inter-fix gap (dropouts) must reflect reality.
|
||
let sample_count = samples.len() as i32;
|
||
let max_gap = samples.windows(2).map(|w| w[1][2] - w[0][2]).fold(0.0_f64, f64::max);
|
||
let end_epoch = t0_abs.map(|t0| t0 + samples.last().unwrap()[2]);
|
||
let samples = downsample(samples, MAX_TRACK_PTS);
|
||
let len = polyline_len_m(&samples);
|
||
let res = sqlx::query(
|
||
r#"UPDATE videos SET gps_status='ok', gps_track=$2, gps_len_m=$3, gps_error=NULL,
|
||
gps_scanned_at=now(),
|
||
gps_start_epoch=$4, gps_end_epoch=$5, gps_sample_count=$6, gps_max_gap_s=$7,
|
||
duration_s=$8,
|
||
-- fill fps for raw NAS videos (no sibling JSON) so frame→position
|
||
-- math works; never clobber a JSON/desktop-derived fps.
|
||
fps=COALESCE(fps, $9),
|
||
-- a fresh raw track invalidates any correction derived from the old one
|
||
gps_track_corrected=NULL, gps_corrected_by=NULL, gps_corrected_at=NULL,
|
||
gps_corrected_note=NULL
|
||
WHERE id=$1"#,
|
||
)
|
||
.bind(id)
|
||
.bind(track_value(&samples))
|
||
.bind(len)
|
||
.bind(t0_abs)
|
||
.bind(end_epoch)
|
||
.bind(sample_count)
|
||
.bind(max_gap)
|
||
.bind(duration_s)
|
||
.bind(fps)
|
||
.execute(&state.db)
|
||
.await;
|
||
match res {
|
||
Ok(_) => {
|
||
tracing::info!("gps: video {id} ok — {sample_count} samples, {len:.0} m, max gap {max_gap:.0}s");
|
||
// Auto-correct this video (spikes + start stitch), then its
|
||
// temporal successor — its start stitch may only now be possible.
|
||
auto_correct(&state.db, id).await;
|
||
if let Some(end) = end_epoch {
|
||
let succ: Option<i32> = sqlx::query_scalar(
|
||
r#"SELECT v.id FROM videos v, videos me
|
||
WHERE me.id=$1 AND v.id<>$1 AND v.project_id=me.project_id
|
||
AND v.gps_status='ok' AND v.gps_start_epoch IS NOT NULL
|
||
AND v.gps_start_epoch > $2 AND v.gps_start_epoch - $2 <= $3
|
||
AND (CASE WHEN v.rel_path LIKE '%/%' THEN regexp_replace(v.rel_path,'/[^/]*$','') ELSE '' END)
|
||
= (CASE WHEN me.rel_path LIKE '%/%' THEN regexp_replace(me.rel_path,'/[^/]*$','') ELSE '' END)
|
||
ORDER BY v.gps_start_epoch LIMIT 1"#,
|
||
)
|
||
.bind(id)
|
||
.bind(end)
|
||
.bind(STITCH_MAX_GAP_S)
|
||
.fetch_optional(&state.db)
|
||
.await
|
||
.unwrap_or(None);
|
||
if let Some(succ) = succ {
|
||
auto_correct(&state.db, succ).await;
|
||
}
|
||
}
|
||
true
|
||
}
|
||
Err(e) => mark_error(&state.db, id, &format!("db write failed: {e}")).await,
|
||
}
|
||
}
|
||
Ok(_) => {
|
||
let _ = sqlx::query(
|
||
"UPDATE videos SET gps_status='none', gps_error=NULL, gps_scanned_at=now() WHERE id=$1",
|
||
)
|
||
.bind(id)
|
||
.execute(&state.db)
|
||
.await;
|
||
tracing::info!("gps: video {id} has no GPS metadata");
|
||
true
|
||
}
|
||
Err(e) => mark_error(&state.db, id, &e).await,
|
||
}
|
||
}
|
||
|
||
async fn mark_error(db: &sqlx::PgPool, id: i32, msg: &str) -> bool {
|
||
tracing::warn!("gps: video {id} scan failed: {msg}");
|
||
let _ = sqlx::query(
|
||
"UPDATE videos SET gps_status='error', gps_error=$2, gps_scanned_at=now() WHERE id=$1",
|
||
)
|
||
.bind(id)
|
||
.bind(msg)
|
||
.execute(db)
|
||
.await;
|
||
false
|
||
}
|
||
|
||
// ---- Stage-1 "Viterbi" smoother ----
|
||
|
||
/// Speed gate: hops implying more than this are implausible (≈160 km/h).
|
||
const MAX_SPEED_MPS: f64 = 45.0;
|
||
/// Cost of dropping (re-interpolating) one sample. An outlier is dropped when
|
||
/// keeping it costs more than this.
|
||
const DROP_PENALTY: f64 = 30.0;
|
||
/// Max consecutive dropped samples + 1 (DP transition window).
|
||
const VITERBI_WINDOW: usize = 12;
|
||
|
||
/// Min-cost keep/drop path over the samples (Viterbi over a keep-lattice):
|
||
/// transition cost between consecutive KEPT samples grows quadratically with the
|
||
/// implied speed above `MAX_SPEED_MPS`; every dropped sample costs `DROP_PENALTY`.
|
||
/// Dropped samples are re-interpolated (by time) between their kept neighbors —
|
||
/// principled outlier rejection with no road network. Returns the corrected track
|
||
/// (same length + timestamps as the input) and how many samples were replaced.
|
||
pub fn viterbi_smooth(track: &[Pt]) -> (Vec<Pt>, usize) {
|
||
let n = track.len();
|
||
if n < 3 {
|
||
return (track.to_vec(), 0);
|
||
}
|
||
|
||
let trans_cost = |a: &Pt, b: &Pt| -> f64 {
|
||
let dt = (b[2] - a[2]).max(0.5);
|
||
let v = dist_m(a[0], a[1], b[0], b[1]) / dt;
|
||
let excess = (v - MAX_SPEED_MPS).max(0.0);
|
||
(excess / 5.0).powi(2)
|
||
};
|
||
|
||
// cost[i] = min cost of a path whose last kept sample is i.
|
||
let mut cost = vec![f64::INFINITY; n];
|
||
let mut prev = vec![usize::MAX; n];
|
||
for i in 0..n {
|
||
if i <= VITERBI_WINDOW {
|
||
// Start the path at i, paying for the dropped leading samples.
|
||
cost[i] = DROP_PENALTY * i as f64;
|
||
}
|
||
for j in i.saturating_sub(VITERBI_WINDOW)..i {
|
||
if cost[j].is_finite() {
|
||
let c = cost[j] + DROP_PENALTY * (i - j - 1) as f64 + trans_cost(&track[j], &track[i]);
|
||
if c < cost[i] {
|
||
cost[i] = c;
|
||
prev[i] = j;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// End the path at the sample minimizing total cost incl. dropped trailing ones.
|
||
let mut end = n - 1;
|
||
let mut best = f64::INFINITY;
|
||
for i in n.saturating_sub(VITERBI_WINDOW + 1)..n {
|
||
let c = cost[i] + DROP_PENALTY * (n - 1 - i) as f64;
|
||
if c < best {
|
||
best = c;
|
||
end = i;
|
||
}
|
||
}
|
||
|
||
let mut kept = vec![false; n];
|
||
let mut i = end;
|
||
loop {
|
||
kept[i] = true;
|
||
if prev[i] == usize::MAX {
|
||
break;
|
||
}
|
||
i = prev[i];
|
||
}
|
||
|
||
// Rebuild: kept samples stay; dropped ones are re-interpolated between the
|
||
// nearest kept neighbors (clamped to the first/last kept at the edges).
|
||
let kept_idx: Vec<usize> = (0..n).filter(|&k| kept[k]).collect();
|
||
let mut out = track.to_vec();
|
||
let mut replaced = 0usize;
|
||
for k in 0..n {
|
||
if kept[k] {
|
||
continue;
|
||
}
|
||
let after = kept_idx.partition_point(|&ki| ki < k);
|
||
let (lo, hi) = match (after.checked_sub(1).map(|a| kept_idx[a]), kept_idx.get(after)) {
|
||
(Some(lo), Some(&hi)) => (lo, hi),
|
||
(None, Some(&hi)) => (hi, hi),
|
||
(Some(lo), None) => (lo, lo),
|
||
(None, None) => continue, // unreachable: at least one sample is kept
|
||
};
|
||
let (a, b) = (&track[lo], &track[hi]);
|
||
let f = if b[2] > a[2] { ((track[k][2] - a[2]) / (b[2] - a[2])).clamp(0.0, 1.0) } else { 0.0 };
|
||
out[k] = [a[0] + (b[0] - a[0]) * f, a[1] + (b[1] - a[1]) * f, track[k][2]];
|
||
replaced += 1;
|
||
}
|
||
(out, replaced)
|
||
}
|
||
|
||
// ---- auto-correction: spikes + start-of-video stitching ----
|
||
|
||
/// Two files are one continuous recording only when the GPS gap between them is at
|
||
/// most this many seconds.
|
||
const STITCH_MAX_GAP_S: f64 = 180.0;
|
||
/// Gaps at or below this are the normal ~1 Hz cadence across a file boundary —
|
||
/// nothing is missing.
|
||
const STITCH_MIN_GAP_S: f64 = 2.0;
|
||
|
||
/// Decide the time-shift for a missing-start stitch. `gap_s` = seconds between the
|
||
/// predecessor's last fix and this video's first fix; `span_s` = this track's own
|
||
/// first→last duration; `duration_s` = the video's real length when known
|
||
/// (frame_count / fps); `jump_m` = distance from the predecessor's last fix to this
|
||
/// video's first fix. Returns the seconds this track starts late (its samples are
|
||
/// shifted by this), or None when nothing should be stitched.
|
||
fn plan_stitch(gap_s: f64, span_s: f64, duration_s: Option<f64>, jump_m: f64) -> Option<f64> {
|
||
if gap_s <= STITCH_MIN_GAP_S || gap_s > STITCH_MAX_GAP_S {
|
||
return None; // healthy rollover, or not a continuous recording
|
||
}
|
||
if jump_m / gap_s > MAX_SPEED_MPS {
|
||
return None; // the vehicle can't have covered the jump — different session
|
||
}
|
||
let mut shift = gap_s - 1.0; // one normal sample period isn't missing data
|
||
if let Some(dur) = duration_s {
|
||
// The track already covers the whole file ⇒ the gap was the PREDECESSOR's
|
||
// problem (missing end), not ours — shifting would misalign every frame.
|
||
let missing = dur - span_s;
|
||
if missing < STITCH_MIN_GAP_S {
|
||
return None;
|
||
}
|
||
shift = shift.min(missing);
|
||
}
|
||
(shift > 0.5).then_some(shift)
|
||
}
|
||
|
||
/// The unified correction result: Viterbi spike removal + (when a temporal
|
||
/// predecessor exists in the same folder) a missing-start stitch that re-bases the
|
||
/// clock so t = 0 is the actual video start. Raw is never modified.
|
||
pub struct CorrectionOutcome {
|
||
pub raw: Vec<Pt>,
|
||
pub corrected: Vec<Pt>,
|
||
pub replaced: usize,
|
||
pub stitched_from: Option<String>,
|
||
pub shift_s: f64,
|
||
pub note: String,
|
||
}
|
||
|
||
impl CorrectionOutcome {
|
||
pub fn changed(&self) -> bool {
|
||
self.replaced > 0 || self.stitched_from.is_some()
|
||
}
|
||
}
|
||
|
||
/// Compute (but do not store) the correction for a video. Errors when the video has
|
||
/// no usable raw track.
|
||
pub async fn compute_correction(db: &sqlx::PgPool, video_id: i32) -> Result<CorrectionOutcome, String> {
|
||
#[allow(clippy::type_complexity)]
|
||
let row: Option<(Option<Value>, Option<f64>, Option<f64>, Option<i32>, Option<f64>, i32, String)> = sqlx::query_as(
|
||
r#"SELECT gps_track, gps_start_epoch, fps, frame_count, duration_s, project_id, rel_path
|
||
FROM videos WHERE id=$1 AND gps_status='ok'"#,
|
||
)
|
||
.bind(video_id)
|
||
.fetch_optional(db)
|
||
.await
|
||
.map_err(|e| e.to_string())?;
|
||
let Some((track, start_epoch, fps, frame_count, duration_s, project_id, rel_path)) = row else {
|
||
return Err("video has no GPS track".into());
|
||
};
|
||
let raw = track.as_ref().map(parse_track).unwrap_or_default();
|
||
if raw.len() < 2 {
|
||
return Err("video has no GPS track".into());
|
||
}
|
||
|
||
let (mut corrected, replaced) = viterbi_smooth(&raw);
|
||
let mut notes: Vec<String> = Vec::new();
|
||
if replaced > 0 {
|
||
notes.push(format!("{replaced} spike(s) re-interpolated"));
|
||
}
|
||
|
||
// Missing-start stitch: the nearest earlier-ending video in the same folder,
|
||
// matched by absolute GPS time (UTC on every camera — no timezone games).
|
||
let mut stitched_from = None;
|
||
let mut shift_s = 0.0;
|
||
if let Some(start_epoch) = start_epoch {
|
||
let folder = match rel_path.rfind('/') {
|
||
Some(i) => rel_path[..i].to_string(),
|
||
None => String::new(),
|
||
};
|
||
let prev: Option<(String, f64, Value)> = sqlx::query_as(
|
||
r#"SELECT file_name, gps_end_epoch, gps_track FROM videos
|
||
WHERE project_id=$1 AND id<>$2 AND gps_track IS NOT NULL AND gps_end_epoch IS NOT NULL
|
||
AND gps_end_epoch < $3 AND $3 - gps_end_epoch <= $4
|
||
AND (CASE WHEN rel_path LIKE '%/%' THEN regexp_replace(rel_path,'/[^/]*$','') ELSE '' END) = $5
|
||
ORDER BY gps_end_epoch DESC LIMIT 1"#,
|
||
)
|
||
.bind(project_id)
|
||
.bind(video_id)
|
||
.bind(start_epoch)
|
||
.bind(STITCH_MAX_GAP_S)
|
||
.bind(&folder)
|
||
.fetch_optional(db)
|
||
.await
|
||
.map_err(|e| e.to_string())?;
|
||
if let Some((prev_name, prev_end, prev_track)) = prev {
|
||
// Borrow from the predecessor's RAW track (clean provenance — its own
|
||
// correction may already contain stitched points).
|
||
let prev_pts = parse_track(&prev_track);
|
||
if let Some(anchor) = prev_pts.last() {
|
||
let gap = start_epoch - prev_end;
|
||
let span = corrected.last().unwrap()[2] - corrected[0][2];
|
||
// Container duration (from the scan) first; the annotation-doc
|
||
// fps/frame_count only as a fallback for pre-upgrade rows.
|
||
let duration = duration_s.or(match (fps, frame_count) {
|
||
(Some(f), Some(n)) if f > 0.0 => Some(n as f64 / f),
|
||
_ => None,
|
||
});
|
||
let jump = dist_m(anchor[0], anchor[1], corrected[0][0], corrected[0][1]);
|
||
if let Some(shift) = plan_stitch(gap, span, duration, jump) {
|
||
for p in corrected.iter_mut() {
|
||
p[2] += shift;
|
||
}
|
||
corrected.insert(0, [anchor[0], anchor[1], 0.0]);
|
||
shift_s = shift;
|
||
notes.push(format!(
|
||
"start stitched from {prev_name} (first {shift:.1}s of GPS were missing — clock re-based)"
|
||
));
|
||
stitched_from = Some(prev_name);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
Ok(CorrectionOutcome {
|
||
raw,
|
||
corrected,
|
||
replaced,
|
||
stitched_from,
|
||
shift_s,
|
||
note: notes.join(" · "),
|
||
})
|
||
}
|
||
|
||
/// Compute + store the correction as 'auto' — but never over a human decision:
|
||
/// only when the video is untouched or its current correction is itself 'auto'.
|
||
/// A no-op correction clears a stale auto one. Best-effort (sweep path).
|
||
pub async fn auto_correct(db: &sqlx::PgPool, video_id: i32) {
|
||
let eligible: Option<bool> = sqlx::query_scalar(
|
||
"SELECT (gps_corrected_by IS NULL OR gps_corrected_by='auto') FROM videos WHERE id=$1 AND gps_status='ok'",
|
||
)
|
||
.bind(video_id)
|
||
.fetch_optional(db)
|
||
.await
|
||
.ok()
|
||
.flatten();
|
||
if eligible != Some(true) {
|
||
return;
|
||
}
|
||
match compute_correction(db, video_id).await {
|
||
Ok(o) if o.changed() => {
|
||
let len = polyline_len_m(&o.corrected);
|
||
let res = sqlx::query(
|
||
r#"UPDATE videos SET gps_track_corrected=$2, gps_corrected_by='auto',
|
||
gps_corrected_at=now(), gps_corrected_note=$3, gps_len_m=$4 WHERE id=$1"#,
|
||
)
|
||
.bind(video_id)
|
||
.bind(track_value(&o.corrected))
|
||
.bind(&o.note)
|
||
.bind(len)
|
||
.execute(db)
|
||
.await;
|
||
match res {
|
||
Ok(_) => tracing::info!("gps: video {video_id} auto-corrected — {}", o.note),
|
||
Err(e) => tracing::warn!("gps: video {video_id} auto-correct store failed: {e}"),
|
||
}
|
||
}
|
||
Ok(o) => {
|
||
// Nothing to fix (anymore) — drop a stale auto correction if present.
|
||
let _ = sqlx::query(
|
||
r#"UPDATE videos SET gps_track_corrected=NULL, gps_corrected_by=NULL,
|
||
gps_corrected_at=NULL, gps_corrected_note=NULL, gps_len_m=$2
|
||
WHERE id=$1 AND gps_corrected_by='auto'"#,
|
||
)
|
||
.bind(video_id)
|
||
.bind(polyline_len_m(&o.raw))
|
||
.execute(db)
|
||
.await;
|
||
}
|
||
Err(e) => tracing::debug!("gps: video {video_id} auto-correct skipped: {e}"),
|
||
}
|
||
}
|
||
|
||
// ---- endpoints ----
|
||
|
||
/// `POST /api/videos/:id/gps/rescan` — queue a video for a fresh metadata scan
|
||
/// (admin). The sweep picks it up; any previous error is cleared.
|
||
pub async fn rescan(State(s): State<AppState>, user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Value> {
|
||
user.require_admin()?;
|
||
let updated: Option<i32> =
|
||
sqlx::query_scalar("UPDATE videos SET gps_status='pending', gps_error=NULL WHERE id=$1 RETURNING id")
|
||
.bind(id)
|
||
.fetch_optional(&s.db)
|
||
.await
|
||
.map_err(err)?;
|
||
if updated.is_none() {
|
||
return Err((StatusCode::NOT_FOUND, "video not found".into()));
|
||
}
|
||
Ok(Json(json!({ "queued": true })))
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
pub struct CorrectRequest {
|
||
/// 'preview' (compute, don't save) | 'keep' (save corrected) | 'discard' (revert to raw).
|
||
pub action: String,
|
||
}
|
||
|
||
/// `POST /api/videos/:id/gps/correct {action}` — the keep-or-discard correction flow
|
||
/// (admin), running the SAME pipeline the scan sweep auto-applies (spike removal +
|
||
/// missing-start stitch). Preview returns original + corrected tracks so the map can
|
||
/// overlay them; keep persists (attributed to the admin); discard reverts to raw AND
|
||
/// tombstones the video so the sweep won't re-auto-correct it (a re-scan resets that).
|
||
pub async fn correct(
|
||
State(s): State<AppState>,
|
||
user: AuthUser,
|
||
AxPath(id): AxPath<i32>,
|
||
Json(req): Json<CorrectRequest>,
|
||
) -> ApiResult<Value> {
|
||
user.require_admin()?;
|
||
let exists: Option<String> = sqlx::query_scalar("SELECT gps_status FROM videos WHERE id=$1")
|
||
.bind(id)
|
||
.fetch_optional(&s.db)
|
||
.await
|
||
.map_err(err)?;
|
||
let Some(gps_status) = exists else {
|
||
return Err((StatusCode::NOT_FOUND, "video not found".into()));
|
||
};
|
||
if gps_status != "ok" {
|
||
return Err((StatusCode::BAD_REQUEST, "video has no GPS track".into()));
|
||
}
|
||
|
||
match req.action.as_str() {
|
||
"discard" => {
|
||
let raw_track: Option<Value> = sqlx::query_scalar("SELECT gps_track FROM videos WHERE id=$1")
|
||
.bind(id)
|
||
.fetch_one(&s.db)
|
||
.await
|
||
.map_err(err)?;
|
||
let len = raw_track.as_ref().map(|t| polyline_len_m(&parse_track(t))).unwrap_or_default();
|
||
// Tombstone: gps_corrected_by keeps the admin's name with a NULL track,
|
||
// so auto_correct (which only touches NULL/'auto') leaves this video raw.
|
||
sqlx::query(
|
||
r#"UPDATE videos SET gps_track_corrected=NULL,
|
||
gps_corrected_by=$2, gps_corrected_at=now(),
|
||
gps_corrected_note='reverted to raw — auto-correction off (re-scan to reset)',
|
||
gps_len_m=$3
|
||
WHERE id=$1"#,
|
||
)
|
||
.bind(id)
|
||
.bind(&user.username)
|
||
.bind(len)
|
||
.execute(&s.db)
|
||
.await
|
||
.map_err(err)?;
|
||
Ok(Json(json!({ "action": "discard", "len_m": len })))
|
||
}
|
||
"preview" | "keep" => {
|
||
let o = compute_correction(&s.db, id).await.map_err(|e| (StatusCode::BAD_REQUEST, e))?;
|
||
let len_before = polyline_len_m(&o.raw);
|
||
let len_after = polyline_len_m(&o.corrected);
|
||
if req.action == "keep" {
|
||
sqlx::query(
|
||
r#"UPDATE videos SET gps_track_corrected=$2, gps_corrected_by=$3,
|
||
gps_corrected_at=now(), gps_corrected_note=$4, gps_len_m=$5 WHERE id=$1"#,
|
||
)
|
||
.bind(id)
|
||
.bind(track_value(&o.corrected))
|
||
.bind(&user.username)
|
||
.bind(&o.note)
|
||
.bind(len_after)
|
||
.execute(&s.db)
|
||
.await
|
||
.map_err(err)?;
|
||
crate::worker::log_event(&s.db, id, &user.username, "gps_corrected",
|
||
json!({ "replaced": o.replaced, "stitched": o.stitched_from.is_some() })).await;
|
||
}
|
||
Ok(Json(json!({
|
||
"action": req.action,
|
||
"points": o.raw.len(),
|
||
"replaced": o.replaced,
|
||
"stitched_from": o.stitched_from,
|
||
"shift_s": o.shift_s,
|
||
"note": o.note,
|
||
"len_before_m": len_before,
|
||
"len_after_m": len_after,
|
||
"original": track_value(&o.raw),
|
||
"corrected": track_value(&o.corrected),
|
||
})))
|
||
}
|
||
_ => Err((StatusCode::BAD_REQUEST, "action must be preview, keep or discard".into())),
|
||
}
|
||
}
|
||
|
||
/// `POST /api/projects/:id/gps/rescan_all` — queue EVERY already-scanned video in the
|
||
/// project for a fresh scan (admin). Needed once after upgrading: pre-upgrade scans
|
||
/// have no absolute timestamps, so stitching/data-quality stats only exist after a
|
||
/// re-scan. The serial sweep works through them in the background.
|
||
pub async fn rescan_all(State(s): State<AppState>, user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Value> {
|
||
user.require_admin()?;
|
||
let n = sqlx::query(
|
||
"UPDATE videos SET gps_status='pending', gps_error=NULL WHERE project_id=$1 AND gps_status <> 'pending'",
|
||
)
|
||
.bind(id)
|
||
.execute(&s.db)
|
||
.await
|
||
.map_err(err)?
|
||
.rows_affected();
|
||
crate::admin::audit(&s.db, &user.username, "gps_rescan_all", json!({ "project_id": id, "queued": n })).await;
|
||
Ok(Json(json!({ "queued": n })))
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
/// A straight east-west track at the equator: 1 pt/s, ~11.1 m apart (40 km/h).
|
||
fn straight(n: usize) -> Vec<Pt> {
|
||
(0..n).map(|i| [0.0, i as f64 * 0.0001, i as f64]).collect()
|
||
}
|
||
|
||
#[test]
|
||
fn smoother_removes_a_jump() {
|
||
let mut t = straight(60);
|
||
// one wild point 0.01° (~1.1 km) off the line
|
||
t[30][0] = 0.01;
|
||
let (fixed, replaced) = viterbi_smooth(&t);
|
||
assert_eq!(replaced, 1);
|
||
assert!(fixed[30][0].abs() < 1e-6, "outlier not re-interpolated: {}", fixed[30][0]);
|
||
// everything else untouched
|
||
assert_eq!(fixed[29], t[29]);
|
||
assert_eq!(fixed[31], t[31]);
|
||
}
|
||
|
||
#[test]
|
||
fn smoother_keeps_a_clean_track() {
|
||
let t = straight(60);
|
||
let (fixed, replaced) = viterbi_smooth(&t);
|
||
assert_eq!(replaced, 0);
|
||
assert_eq!(fixed, t);
|
||
}
|
||
|
||
#[test]
|
||
fn smoother_removes_a_burst() {
|
||
let mut t = straight(80);
|
||
for k in 40..44 {
|
||
t[k][0] = 0.02; // 2.2 km off for 4 consecutive samples
|
||
}
|
||
let (fixed, replaced) = viterbi_smooth(&t);
|
||
assert_eq!(replaced, 4);
|
||
for k in 40..44 {
|
||
assert!(fixed[k][0].abs() < 1e-6);
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn downsample_caps_and_keeps_last() {
|
||
let t = straight(12000);
|
||
let d = downsample(t.clone(), 5000);
|
||
assert!(d.len() <= 5001);
|
||
assert_eq!(d.last(), t.last());
|
||
}
|
||
|
||
#[test]
|
||
fn speed_stats_measures_avg_and_max() {
|
||
// ~11.13 m per second for 60 s → avg ≈ max ≈ 11.1 m/s (40 km/h).
|
||
let t = straight(60);
|
||
let (avg, max) = speed_stats(&t);
|
||
assert!((avg.unwrap() - 11.13).abs() < 0.1, "{avg:?}");
|
||
assert!((max.unwrap() - 11.13).abs() < 0.1, "{max:?}");
|
||
// One fast hop dominates max but barely moves avg.
|
||
let mut t2 = straight(60);
|
||
t2[30][1] += 0.0002; // extra ~22 m in that second
|
||
let (avg2, max2) = speed_stats(&t2);
|
||
assert!(max2.unwrap() > 25.0, "{max2:?}");
|
||
assert!(avg2.unwrap() < 13.0, "{avg2:?}");
|
||
}
|
||
|
||
#[test]
|
||
fn stitch_planning_rules() {
|
||
// Healthy 1 Hz rollover (1 s gap) → nothing missing.
|
||
assert_eq!(plan_stitch(1.0, 299.0, Some(300.0), 20.0), None);
|
||
// 9 s gap, duration unknown → shift 8 s.
|
||
assert_eq!(plan_stitch(9.0, 291.0, None, 100.0), Some(8.0));
|
||
// 9 s gap but the track already covers the whole file → the PREDECESSOR was
|
||
// missing its end; shifting us would misalign every frame.
|
||
assert_eq!(plan_stitch(9.0, 299.0, Some(300.0), 100.0), None);
|
||
// Duration caps the shift: 20 s gap but only 5 s missing from this file.
|
||
assert_eq!(plan_stitch(20.0, 295.0, Some(300.0), 100.0), Some(5.0));
|
||
// Teleport guard: 5 km jump in 9 s can't be the same vehicle.
|
||
assert_eq!(plan_stitch(9.0, 291.0, None, 5000.0), None);
|
||
// Different session: 10 minutes apart.
|
||
assert_eq!(plan_stitch(600.0, 291.0, None, 100.0), None);
|
||
}
|
||
|
||
#[test]
|
||
fn wgs84_distance_matches_reference_arcs() {
|
||
// Textbook WGS84 arc lengths: 1° latitude ≈ 110.574 km (equator) /
|
||
// 111.132 km (45°); 1° longitude at the equator ≈ 111.320 km.
|
||
assert!((dist_m(0.0, 0.0, 1.0, 0.0) - 110_574.0).abs() < 200.0);
|
||
assert!((dist_m(0.0, 0.0, 0.0, 1.0) - 111_320.0).abs() < 30.0);
|
||
assert!((dist_m(44.5, 0.0, 45.5, 0.0) - 111_132.0).abs() < 200.0);
|
||
}
|
||
|
||
#[test]
|
||
fn track_roundtrip() {
|
||
let t = straight(10);
|
||
let v = track_value(&t);
|
||
let back = parse_track(&v);
|
||
assert_eq!(back.len(), 10);
|
||
assert!((back[9][1] - t[9][1]).abs() < 1e-5);
|
||
}
|
||
}
|