lesavka/server/src/calibration.rs

468 lines
18 KiB
Rust
Raw Normal View History

2026-04-30 08:16:57 -03:00
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use anyhow::{Context, Result};
use chrono::Utc;
use lesavka_common::lesavka::{
CalibrationAction, CalibrationRequest, CalibrationState as ProtoCalibrationState,
};
use crate::upstream_media_runtime::UpstreamMediaRuntime;
pub const FACTORY_MJPEG_AUDIO_OFFSET_US: i64 = -45_000;
pub const FACTORY_MJPEG_VIDEO_OFFSET_US: i64 = 0;
const PROFILE: &str = "mjpeg";
const FACTORY_CONFIDENCE: &str = "factory";
const OFFSET_LIMIT_US: i64 = 500_000;
#[derive(Debug, Clone, PartialEq, Eq)]
struct CalibrationSnapshot {
profile: String,
factory_audio_offset_us: i64,
factory_video_offset_us: i64,
default_audio_offset_us: i64,
default_video_offset_us: i64,
active_audio_offset_us: i64,
active_video_offset_us: i64,
source: String,
confidence: String,
updated_at: String,
detail: String,
}
#[derive(Debug)]
pub struct CalibrationStore {
path: PathBuf,
runtime: Arc<UpstreamMediaRuntime>,
state: Mutex<CalibrationSnapshot>,
}
impl CalibrationStore {
pub fn load(runtime: Arc<UpstreamMediaRuntime>) -> Self {
let path = calibration_path();
let state = std::fs::read_to_string(&path)
.ok()
.map(|raw| parse_snapshot(&raw))
.unwrap_or_else(snapshot_from_env);
runtime.set_playout_offsets(state.active_video_offset_us, state.active_audio_offset_us);
Self {
path,
runtime,
state: Mutex::new(state),
}
}
pub fn current(&self) -> ProtoCalibrationState {
self.state
.lock()
.expect("calibration mutex poisoned")
.to_proto()
}
pub fn apply(&self, request: CalibrationRequest) -> Result<ProtoCalibrationState> {
let mut state = self.state.lock().expect("calibration mutex poisoned");
let action =
CalibrationAction::try_from(request.action).unwrap_or(CalibrationAction::Unspecified);
match action {
CalibrationAction::Unspecified => {}
CalibrationAction::RestoreDefault => {
state.active_audio_offset_us = state.default_audio_offset_us;
state.active_video_offset_us = state.default_video_offset_us;
state.source = "default".to_string();
state.confidence = "saved-default".to_string();
state.detail = "restored saved upstream A/V calibration".to_string();
touch(&mut state);
}
CalibrationAction::RestoreFactory => {
state.active_audio_offset_us = state.factory_audio_offset_us;
state.active_video_offset_us = state.factory_video_offset_us;
state.source = "factory".to_string();
state.confidence = FACTORY_CONFIDENCE.to_string();
state.detail = "restored release-shipped MJPEG upstream calibration".to_string();
touch(&mut state);
}
CalibrationAction::AdjustActive => {
state.active_audio_offset_us = clamp_offset(
state
.active_audio_offset_us
.saturating_add(request.audio_delta_us),
);
state.active_video_offset_us = clamp_offset(
state
.active_video_offset_us
.saturating_add(request.video_delta_us),
);
state.source = "manual".to_string();
state.confidence = "manual".to_string();
state.detail = format!(
"manual upstream A/V calibration nudge: audio {:+.1}ms, video {:+.1}ms",
request.audio_delta_us as f64 / 1000.0,
request.video_delta_us as f64 / 1000.0
);
touch(&mut state);
}
CalibrationAction::BlindEstimate => {
state.active_audio_offset_us = clamp_offset(
state
.active_audio_offset_us
.saturating_add(request.audio_delta_us),
);
state.active_video_offset_us = clamp_offset(
state
.active_video_offset_us
.saturating_add(request.video_delta_us),
);
state.source = "blind".to_string();
state.confidence = "estimated".to_string();
state.detail = if request.note.trim().is_empty() {
format!(
"blind estimate applied from relay telemetry: delivery skew {:.1}ms, enqueue skew {:.1}ms",
request.observed_delivery_skew_ms, request.observed_enqueue_skew_ms
)
} else {
request.note
};
touch(&mut state);
}
CalibrationAction::SaveActiveAsDefault => {
state.default_audio_offset_us = state.active_audio_offset_us;
state.default_video_offset_us = state.active_video_offset_us;
state.source = "default".to_string();
state.confidence = "measured".to_string();
state.detail = "saved current upstream A/V calibration as site default".to_string();
touch(&mut state);
}
}
self.runtime
.set_playout_offsets(state.active_video_offset_us, state.active_audio_offset_us);
persist_snapshot(&self.path, &state)?;
Ok(state.to_proto())
}
}
impl CalibrationSnapshot {
fn to_proto(&self) -> ProtoCalibrationState {
ProtoCalibrationState {
profile: self.profile.clone(),
factory_audio_offset_us: self.factory_audio_offset_us,
factory_video_offset_us: self.factory_video_offset_us,
default_audio_offset_us: self.default_audio_offset_us,
default_video_offset_us: self.default_video_offset_us,
active_audio_offset_us: self.active_audio_offset_us,
active_video_offset_us: self.active_video_offset_us,
source: self.source.clone(),
confidence: self.confidence.clone(),
updated_at: self.updated_at.clone(),
detail: self.detail.clone(),
}
}
}
pub fn calibration_path() -> PathBuf {
std::env::var("LESAVKA_CALIBRATION_PATH")
.ok()
.filter(|path| !path.trim().is_empty())
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from("/var/lib/lesavka/calibration.toml"))
}
fn snapshot_from_env() -> CalibrationSnapshot {
let env_audio = env_i64("LESAVKA_UPSTREAM_AUDIO_PLAYOUT_OFFSET_US");
let env_video = env_i64("LESAVKA_UPSTREAM_VIDEO_PLAYOUT_OFFSET_US");
let default_audio_offset_us = env_audio.unwrap_or(FACTORY_MJPEG_AUDIO_OFFSET_US);
let default_video_offset_us = env_video.unwrap_or(FACTORY_MJPEG_VIDEO_OFFSET_US);
let source = if env_audio.is_some() || env_video.is_some() {
"env".to_string()
} else {
"factory".to_string()
};
let confidence = if source == "factory" {
FACTORY_CONFIDENCE.to_string()
} else {
"configured".to_string()
};
CalibrationSnapshot {
profile: PROFILE.to_string(),
factory_audio_offset_us: FACTORY_MJPEG_AUDIO_OFFSET_US,
factory_video_offset_us: FACTORY_MJPEG_VIDEO_OFFSET_US,
default_audio_offset_us,
default_video_offset_us,
active_audio_offset_us: default_audio_offset_us,
active_video_offset_us: default_video_offset_us,
source,
confidence,
updated_at: Utc::now().to_rfc3339(),
detail: "loaded upstream A/V calibration defaults".to_string(),
}
}
fn parse_snapshot(raw: &str) -> CalibrationSnapshot {
let fallback = snapshot_from_env();
let value = |key: &str| -> Option<String> {
raw.lines().find_map(|line| {
let trimmed = line.trim();
let (left, right) = trimmed.split_once('=')?;
(left.trim() == key).then(|| right.trim().trim_matches('"').to_string())
})
};
let number = |key: &str, default: i64| -> i64 {
value(key)
.and_then(|raw| raw.parse::<i64>().ok())
.map(clamp_offset)
.unwrap_or(default)
};
CalibrationSnapshot {
profile: value("profile").unwrap_or(fallback.profile),
factory_audio_offset_us: FACTORY_MJPEG_AUDIO_OFFSET_US,
factory_video_offset_us: FACTORY_MJPEG_VIDEO_OFFSET_US,
default_audio_offset_us: number(
"default_audio_offset_us",
fallback.default_audio_offset_us,
),
default_video_offset_us: number(
"default_video_offset_us",
fallback.default_video_offset_us,
),
active_audio_offset_us: number("active_audio_offset_us", fallback.active_audio_offset_us),
active_video_offset_us: number("active_video_offset_us", fallback.active_video_offset_us),
source: value("source").unwrap_or(fallback.source),
confidence: value("confidence").unwrap_or(fallback.confidence),
updated_at: value("updated_at").unwrap_or(fallback.updated_at),
detail: value("detail").unwrap_or(fallback.detail),
}
}
fn persist_snapshot(path: &PathBuf, state: &CalibrationSnapshot) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("creating calibration directory {}", parent.display()))?;
}
std::fs::write(path, serialize_snapshot(state))
.with_context(|| format!("writing calibration state {}", path.display()))
}
fn serialize_snapshot(state: &CalibrationSnapshot) -> String {
format!(
"profile=\"{}\"\ndefault_audio_offset_us={}\ndefault_video_offset_us={}\nactive_audio_offset_us={}\nactive_video_offset_us={}\nsource=\"{}\"\nconfidence=\"{}\"\nupdated_at=\"{}\"\ndetail=\"{}\"\n",
escape_value(&state.profile),
state.default_audio_offset_us,
state.default_video_offset_us,
state.active_audio_offset_us,
state.active_video_offset_us,
escape_value(&state.source),
escape_value(&state.confidence),
escape_value(&state.updated_at),
escape_value(&state.detail),
)
}
fn touch(state: &mut CalibrationSnapshot) {
state.updated_at = Utc::now().to_rfc3339();
}
fn env_i64(name: &str) -> Option<i64> {
std::env::var(name)
.ok()
.and_then(|value| value.trim().parse::<i64>().ok())
.map(clamp_offset)
}
fn clamp_offset(value: i64) -> i64 {
value.clamp(-OFFSET_LIMIT_US, OFFSET_LIMIT_US)
}
fn escape_value(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"")
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::NamedTempFile;
#[test]
fn default_snapshot_uses_factory_mjpeg_calibration() {
temp_env::with_vars(
[
("LESAVKA_UPSTREAM_AUDIO_PLAYOUT_OFFSET_US", None::<&str>),
("LESAVKA_UPSTREAM_VIDEO_PLAYOUT_OFFSET_US", None::<&str>),
],
|| {
let state = snapshot_from_env();
assert_eq!(state.default_audio_offset_us, -45_000);
assert_eq!(state.active_video_offset_us, 0);
assert_eq!(state.source, "factory");
},
);
}
#[test]
fn store_persists_manual_adjustments_and_updates_runtime() {
let file = NamedTempFile::new().expect("temp calibration file");
let path = file.path().to_string_lossy().to_string();
temp_env::with_var("LESAVKA_CALIBRATION_PATH", Some(path.as_str()), || {
let runtime = Arc::new(UpstreamMediaRuntime::new());
let store = CalibrationStore::load(runtime.clone());
let state = store
.apply(CalibrationRequest {
action: CalibrationAction::AdjustActive as i32,
audio_delta_us: -5_000,
video_delta_us: 0,
observed_delivery_skew_ms: 0.0,
observed_enqueue_skew_ms: 0.0,
note: String::new(),
})
.expect("manual adjust applies");
assert_eq!(state.active_audio_offset_us, -50_000);
assert_eq!(runtime.playout_offsets(), (0, -50_000));
let raw = std::fs::read_to_string(file.path()).expect("persisted calibration");
assert!(raw.contains("active_audio_offset_us=-50000"));
});
}
#[test]
fn calibration_path_uses_default_for_blank_override() {
temp_env::with_var("LESAVKA_CALIBRATION_PATH", Some(""), || {
assert_eq!(
calibration_path(),
PathBuf::from("/var/lib/lesavka/calibration.toml")
);
});
}
#[test]
fn snapshot_from_env_uses_configured_offsets_and_clamps_extremes() {
temp_env::with_vars(
[
("LESAVKA_UPSTREAM_AUDIO_PLAYOUT_OFFSET_US", Some("-999999")),
("LESAVKA_UPSTREAM_VIDEO_PLAYOUT_OFFSET_US", Some("12345")),
],
|| {
let state = snapshot_from_env();
assert_eq!(state.default_audio_offset_us, -500_000);
assert_eq!(state.default_video_offset_us, 12_345);
assert_eq!(state.source, "env");
assert_eq!(state.confidence, "configured");
},
);
}
#[test]
fn parse_snapshot_falls_back_for_missing_and_malformed_fields() {
temp_env::with_vars(
[
("LESAVKA_UPSTREAM_AUDIO_PLAYOUT_OFFSET_US", None::<&str>),
("LESAVKA_UPSTREAM_VIDEO_PLAYOUT_OFFSET_US", None::<&str>),
],
|| {
let state = parse_snapshot(
r#"
profile="mjpeg"
default_audio_offset_us=bad
default_video_offset_us=2500
active_audio_offset_us=-600000
source="saved"
detail="loaded \"quoted\" value"
"#,
);
assert_eq!(state.default_audio_offset_us, FACTORY_MJPEG_AUDIO_OFFSET_US);
assert_eq!(state.default_video_offset_us, 2_500);
assert_eq!(state.active_audio_offset_us, -500_000);
assert_eq!(state.active_video_offset_us, FACTORY_MJPEG_VIDEO_OFFSET_US);
assert_eq!(state.source, "saved");
assert_eq!(state.confidence, FACTORY_CONFIDENCE);
},
);
}
#[test]
fn store_applies_all_calibration_actions_and_persists_defaults() {
let dir = tempfile::tempdir().expect("calibration dir");
let path = dir.path().join("calibration.toml");
let path_string = path.to_string_lossy().to_string();
temp_env::with_var(
"LESAVKA_CALIBRATION_PATH",
Some(path_string.as_str()),
|| {
let runtime = Arc::new(UpstreamMediaRuntime::new());
let store = CalibrationStore::load(runtime.clone());
let blind = store
.apply(CalibrationRequest {
action: CalibrationAction::BlindEstimate as i32,
audio_delta_us: 5_000,
video_delta_us: -2_000,
observed_delivery_skew_ms: 44.0,
observed_enqueue_skew_ms: 3.5,
note: String::new(),
})
.expect("blind estimate");
assert_eq!(blind.source, "blind");
assert!(blind.detail.contains("delivery skew 44.0ms"));
assert_eq!(runtime.playout_offsets(), (-2_000, -40_000));
let manual = store
.apply(CalibrationRequest {
action: CalibrationAction::AdjustActive as i32,
audio_delta_us: 999_999,
video_delta_us: 0,
observed_delivery_skew_ms: 0.0,
observed_enqueue_skew_ms: 0.0,
note: String::new(),
})
.expect("manual clamp");
assert_eq!(manual.active_audio_offset_us, 500_000);
let saved = store
.apply(CalibrationRequest {
action: CalibrationAction::SaveActiveAsDefault as i32,
..CalibrationRequest::default()
})
.expect("save default");
assert_eq!(saved.default_audio_offset_us, saved.active_audio_offset_us);
assert_eq!(saved.confidence, "measured");
let factory = store
.apply(CalibrationRequest {
action: CalibrationAction::RestoreFactory as i32,
..CalibrationRequest::default()
})
.expect("factory restore");
assert_eq!(
factory.active_audio_offset_us,
FACTORY_MJPEG_AUDIO_OFFSET_US
);
assert_eq!(factory.source, "factory");
let restored = store
.apply(CalibrationRequest {
action: CalibrationAction::RestoreDefault as i32,
..CalibrationRequest::default()
})
.expect("default restore");
assert_eq!(
restored.active_audio_offset_us,
restored.default_audio_offset_us
);
assert_eq!(
store.current().active_audio_offset_us,
restored.active_audio_offset_us
);
let no_op = store
.apply(CalibrationRequest::default())
.expect("unspecified action is ok");
assert_eq!(
no_op.active_audio_offset_us,
restored.active_audio_offset_us
);
let raw = std::fs::read_to_string(&path).expect("persisted actions");
assert!(raw.contains("confidence="));
assert!(raw.contains("detail="));
assert_eq!(escape_value("a\\b\"c"), "a\\\\b\\\"c");
},
);
}
}