ados_offload/
session.rs

1//! The streaming offload session.
2//!
3//! A behaviour opens a session in one of three modes and streams frames to the
4//! compute node; the node streams detections and/or drift-corrected poses back.
5//! The session does not move frames itself (the transport is injected by the
6//! consumer), it owns the SAFETY: it tracks each return stream's freshness and
7//! runs the lock-state gate so a behaviour commands only on fresh, locked
8//! results and stops the moment any required stream goes stale or the link
9//! drops. The fast control loop and the guided-setpoint math stay on the drone;
10//! this gates the slow remote layer that corrects it.
11
12use serde::{Deserialize, Serialize};
13
14use crate::freshness::{FreshnessGate, GateState, LockGate, LockState};
15
16/// What the node returns for a session.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(rename_all = "snake_case")]
19pub enum OffloadMode {
20    /// Detections + tracking (a target / TargetRef). Consumers: follow, orbit,
21    /// inspection / SAR detect.
22    VisionOnly,
23    /// Drift-corrected poses + occupancy. Consumers: navigation, avoidance,
24    /// the live world model.
25    SlamOnly,
26    /// Both, from one frame stream, an NPU-less drone on a full autonomous
27    /// mission. Any required stream going stale trips the lock.
28    Full,
29}
30
31/// A snapshot of a session's safety state for one cycle.
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub struct SessionStatus {
34    /// The lock state after this cycle.
35    pub lock: LockState,
36    /// The detection/target stream's freshness.
37    pub target: GateState,
38    /// The pose stream's freshness.
39    pub pose: GateState,
40    /// Whether the behaviour may command the FC this cycle (locked AND every
41    /// stream the mode requires is fresh). When false, the behaviour stops/holds.
42    pub commanding: bool,
43}
44
45/// A streaming perception/SLAM offload session with per-stream freshness gates
46/// and a lock-state safety gate.
47#[derive(Debug, Clone)]
48pub struct OffloadSession {
49    mode: OffloadMode,
50    target_gate: FreshnessGate,
51    pose_gate: FreshnessGate,
52    lock: LockGate,
53}
54
55impl OffloadSession {
56    /// A session in `mode` with the target-age and pose-age freshness budgets.
57    pub fn new(mode: OffloadMode, target_budget_ms: i64, pose_budget_ms: i64) -> Self {
58        Self {
59            mode,
60            target_gate: FreshnessGate::new(target_budget_ms),
61            pose_gate: FreshnessGate::new(pose_budget_ms),
62            lock: LockGate::new(),
63        }
64    }
65
66    /// The mode the session was opened in.
67    pub fn mode(&self) -> OffloadMode {
68        self.mode
69    }
70
71    /// A detection/target result arrived from the node. `now_ms` is the local
72    /// monotonic time of arrival (the same clock [`tick`](Self::tick) reads), not
73    /// the result's own timestamp, see [`FreshnessGate`].
74    pub fn on_detection(&mut self, now_ms: i64) {
75        self.target_gate.record(now_ms);
76    }
77
78    /// A drift-corrected pose arrived from the node.
79    /// `now_ms` is the local monotonic time of arrival.
80    pub fn on_pose(&mut self, now_ms: i64) {
81        self.pose_gate.record(now_ms);
82    }
83
84    /// The link state changed (a drop trips every stream to lost).
85    pub fn set_link(&mut self, up: bool) {
86        self.target_gate.set_link(up);
87        self.pose_gate.set_link(up);
88    }
89
90    /// Acquire (or re-acquire) the lock, an operator designate / click-to-track.
91    /// The only way out of a lost lock.
92    pub fn designate(&mut self) {
93        self.lock.lock();
94    }
95
96    /// Release the lock (back to idle).
97    pub fn drop_lock(&mut self) {
98        self.lock.unlock();
99    }
100
101    /// The current lock state.
102    pub fn lock_state(&self) -> LockState {
103        self.lock.state()
104    }
105
106    /// Whether every return stream the mode requires is fresh at `now_ms`.
107    fn streams_usable(&self, now_ms: i64) -> bool {
108        match self.mode {
109            OffloadMode::VisionOnly => self.target_gate.is_usable(now_ms),
110            OffloadMode::SlamOnly => self.pose_gate.is_usable(now_ms),
111            OffloadMode::Full => {
112                self.target_gate.is_usable(now_ms) && self.pose_gate.is_usable(now_ms)
113            }
114        }
115    }
116
117    /// Advance the session one cycle. Updates the lock from the stream freshness
118    /// (a locked behaviour whose required stream went stale drops to lost and
119    /// must be re-designated) and returns the safety snapshot the behaviour acts
120    /// on. `commanding` is true only when locked AND fresh.
121    pub fn tick(&mut self, now_ms: i64) -> SessionStatus {
122        let usable = self.streams_usable(now_ms);
123        self.lock.update(usable);
124        SessionStatus {
125            lock: self.lock.state(),
126            target: self.target_gate.state(now_ms),
127            pose: self.pose_gate.state(now_ms),
128            // Command only when locked AND the required streams are fresh this
129            // cycle. A still-acquiring lock (locked, no fresh reading yet) holds.
130            commanding: self.lock.is_locked() && usable,
131        }
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138
139    #[test]
140    fn a_full_session_commands_when_locked_and_both_streams_fresh() {
141        let mut s = OffloadSession::new(OffloadMode::Full, 500, 500);
142        s.designate();
143        s.on_detection(1000);
144        s.on_pose(1000);
145        let status = s.tick(1100);
146        assert_eq!(status.lock, LockState::Locked);
147        assert!(status.commanding);
148    }
149
150    #[test]
151    fn a_full_session_stops_when_one_required_stream_goes_stale() {
152        let mut s = OffloadSession::new(OffloadMode::Full, 500, 500);
153        s.designate();
154        s.on_detection(1000);
155        s.on_pose(1000);
156        assert!(s.tick(1100).commanding);
157        // The pose stream stops updating; it goes stale past the budget.
158        s.on_detection(1500); // detections keep coming...
159        let status = s.tick(1700); // ...but the pose is now 700ms old (> 500)
160        assert_eq!(status.pose, GateState::Stale);
161        assert_eq!(status.lock, LockState::Lost);
162        assert!(!status.commanding);
163    }
164
165    #[test]
166    fn a_link_drop_stops_the_session() {
167        let mut s = OffloadSession::new(OffloadMode::VisionOnly, 500, 500);
168        s.designate();
169        s.on_detection(1000);
170        assert!(s.tick(1100).commanding);
171        s.set_link(false);
172        let status = s.tick(1150);
173        assert_eq!(status.lock, LockState::Lost);
174        assert!(!status.commanding);
175    }
176
177    #[test]
178    fn a_session_never_auto_re_acquires_after_a_stale_drop() {
179        let mut s = OffloadSession::new(OffloadMode::VisionOnly, 500, 500);
180        s.designate();
181        s.on_detection(1000);
182        assert!(s.tick(1100).commanding);
183        // Go stale -> lost.
184        assert_eq!(s.tick(2000).lock, LockState::Lost);
185        // Fresh detections resume, but the lock must NOT come back by itself.
186        s.on_detection(2100);
187        s.on_detection(2200);
188        let status = s.tick(2250);
189        assert_eq!(status.target, GateState::Fresh);
190        assert_eq!(status.lock, LockState::Lost); // still lost
191        assert!(!status.commanding);
192        // Only a re-designate re-acquires.
193        s.designate();
194        s.on_detection(2300);
195        assert!(s.tick(2350).commanding);
196    }
197
198    #[test]
199    fn slam_only_gates_on_the_pose_stream() {
200        let mut s = OffloadSession::new(OffloadMode::SlamOnly, 500, 500);
201        s.designate();
202        // A detection without a pose does not make a SLAM-only session command.
203        s.on_detection(1000);
204        assert!(!s.tick(1100).commanding);
205        s.on_pose(1100);
206        assert!(s.tick(1200).commanding);
207    }
208
209    #[test]
210    fn a_session_never_auto_re_acquires_after_a_link_drop() {
211        // The link path closed the same way as the stale path: a recovered link
212        // + fresh results must NOT re-acquire a dropped lock on its own.
213        let mut s = OffloadSession::new(OffloadMode::VisionOnly, 500, 500);
214        s.designate();
215        s.on_detection(1000);
216        assert!(s.tick(1100).commanding);
217        s.set_link(false);
218        assert_eq!(s.tick(1150).lock, LockState::Lost);
219        // Link recovers, fresh detections resume, but the lock stays Lost.
220        s.set_link(true);
221        s.on_detection(1200);
222        let status = s.tick(1250);
223        assert_eq!(status.target, GateState::Fresh);
224        assert_eq!(status.lock, LockState::Lost);
225        assert!(!status.commanding);
226    }
227
228    #[test]
229    fn a_full_session_holds_while_only_one_stream_is_fresh() {
230        // Full needs BOTH streams; with only the target fresh it stays acquiring
231        // (locked, holding) and never commands, it does not trip to Lost (it has
232        // not yet tracked) and it does not command on half the data.
233        let mut s = OffloadSession::new(OffloadMode::Full, 500, 500);
234        s.designate();
235        s.on_detection(1000); // target only, no pose yet
236        let status = s.tick(1100);
237        assert_eq!(status.lock, LockState::Locked); // still acquiring
238        assert!(!status.commanding);
239    }
240
241    #[test]
242    fn an_un_designated_session_never_commands() {
243        let mut s = OffloadSession::new(OffloadMode::Full, 500, 500);
244        s.on_detection(1000);
245        s.on_pose(1000);
246        // Fresh streams, but no designate -> idle, not commanding.
247        let status = s.tick(1100);
248        assert_eq!(status.lock, LockState::Unlocked);
249        assert!(!status.commanding);
250    }
251}
ados_offload/session.rs

ados_offload/
session.rs
