canopen_host/
nmt.rs

1//! NMT master tooling: heartbeat monitoring and node health.
2//!
3//! [`HeartbeatMonitor`](crate::nmt::HeartbeatMonitor) tracks the heartbeats a
4//! master hears from the nodes on a bus and flags those that have gone silent.
5//! It is transport-agnostic — feed it received frames and a timestamp — so it
6//! builds on every platform; sending NMT node-control commands lives on the
7//! Linux `SocketCan` transport (`send_nmt`).
8
9use std::collections::BTreeMap;
10use std::time::{Duration, Instant};
11
12use canopen_rs::nmt::{decode_heartbeat, HEARTBEAT_COB_BASE};
13use canopen_rs::{NmtState, NodeId};
14
15/// The last-known health of a monitored node.
16#[derive(Debug, Clone, Copy)]
17pub struct NodeHealth {
18    /// The NMT state last reported by the node's heartbeat.
19    pub state: NmtState,
20    /// When that heartbeat was received.
21    pub last_seen: Instant,
22}
23
24/// Tracks per-node heartbeats and flags nodes that stop producing them.
25///
26/// A CANopen node emits a heartbeat on `0x700 + node`; a master watches those
27/// frames to know each node is alive and in which state. Construct with the
28/// consumer timeout, feed every received frame to [`HeartbeatMonitor::on_frame`],
29/// and query [`HeartbeatMonitor::timed_out`] / [`HeartbeatMonitor::is_alive`].
30#[derive(Debug)]
31pub struct HeartbeatMonitor {
32    nodes: BTreeMap<u8, NodeHealth>,
33    timeout: Duration,
34}
35
36impl HeartbeatMonitor {
37    /// Create a monitor that considers a node dead after `timeout` without a
38    /// heartbeat.
39    pub fn new(timeout: Duration) -> Self {
40        Self {
41            nodes: BTreeMap::new(),
42            timeout,
43        }
44    }
45
46    /// Feed a received frame, timestamped `now`.
47    ///
48    /// If it is a heartbeat or boot-up frame, records it and returns the
49    /// `(node, state)` it reported; otherwise returns `None`.
50    pub fn on_frame(
51        &mut self,
52        cob_id: u16,
53        data: &[u8],
54        now: Instant,
55    ) -> Option<(NodeId, NmtState)> {
56        // Heartbeat producers use 0x700 + node, with node in 1..=127.
57        let raw = cob_id.checked_sub(HEARTBEAT_COB_BASE)?;
58        let node = NodeId::new(u8::try_from(raw).ok()?).ok()?;
59        let state = decode_heartbeat(&[*data.first()?]).ok()?;
60        self.nodes.insert(
61            node.raw(),
62            NodeHealth {
63                state,
64                last_seen: now,
65            },
66        );
67        Some((node, state))
68    }
69
70    /// The last state reported by `node`, if it has ever been heard from.
71    pub fn state(&self, node: NodeId) -> Option<NmtState> {
72        self.nodes.get(&node.raw()).map(|h| h.state)
73    }
74
75    /// The full health record for `node`.
76    pub fn health(&self, node: NodeId) -> Option<NodeHealth> {
77        self.nodes.get(&node.raw()).copied()
78    }
79
80    /// Whether `node` produced a heartbeat within the timeout as of `now`.
81    pub fn is_alive(&self, node: NodeId, now: Instant) -> bool {
82        self.nodes
83            .get(&node.raw())
84            .is_some_and(|h| now.saturating_duration_since(h.last_seen) <= self.timeout)
85    }
86
87    /// The nodes that have gone silent — no heartbeat within the timeout.
88    pub fn timed_out(&self, now: Instant) -> impl Iterator<Item = NodeId> + '_ {
89        self.nodes.iter().filter_map(move |(&id, h)| {
90            (now.saturating_duration_since(h.last_seen) > self.timeout)
91                .then(|| NodeId::new(id).ok())
92                .flatten()
93        })
94    }
95
96    /// Every monitored node paired with its health record.
97    pub fn nodes(&self) -> impl Iterator<Item = (NodeId, NodeHealth)> + '_ {
98        self.nodes
99            .iter()
100            .filter_map(|(&id, &h)| NodeId::new(id).ok().map(|n| (n, h)))
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107
108    fn node(id: u8) -> NodeId {
109        NodeId::new(id).unwrap()
110    }
111
112    #[test]
113    fn records_heartbeat_state() {
114        let mut m = HeartbeatMonitor::new(Duration::from_secs(1));
115        let t0 = Instant::now();
116        // Heartbeat from node 5 reporting operational (0x05) on 0x705.
117        assert_eq!(
118            m.on_frame(0x705, &[0x05], t0),
119            Some((node(5), NmtState::Operational))
120        );
121        assert_eq!(m.state(node(5)), Some(NmtState::Operational));
122    }
123
124    #[test]
125    fn decodes_bootup_frame() {
126        let mut m = HeartbeatMonitor::new(Duration::from_secs(1));
127        // Boot-up is 0x700+node with data 0x00 -> Initialising.
128        assert_eq!(
129            m.on_frame(0x70A, &[0x00], Instant::now()),
130            Some((node(10), NmtState::Initialising))
131        );
132    }
133
134    #[test]
135    fn ignores_non_heartbeat_frames() {
136        let mut m = HeartbeatMonitor::new(Duration::from_secs(1));
137        let now = Instant::now();
138        assert_eq!(m.on_frame(0x585, &[0x43, 0, 0, 0], now), None); // an SDO response
139        assert_eq!(m.on_frame(0x000, &[0x01, 0x05], now), None); // an NMT command
140        assert_eq!(m.on_frame(0x700, &[0x00], now), None); // node 0 is not valid
141    }
142
143    #[test]
144    fn liveness_and_timeout() {
145        let mut m = HeartbeatMonitor::new(Duration::from_secs(1));
146        let t0 = Instant::now();
147        m.on_frame(0x705, &[0x05], t0);
148
149        assert!(m.is_alive(node(5), t0));
150        assert!(m.is_alive(node(5), t0 + Duration::from_millis(900)));
151        assert!(!m.is_alive(node(5), t0 + Duration::from_millis(1500)));
152
153        let timed_out: Vec<_> = m.timed_out(t0 + Duration::from_secs(2)).collect();
154        assert_eq!(timed_out, vec![node(5)]);
155        assert!(m.timed_out(t0).next().is_none());
156    }
157
158    #[test]
159    fn unknown_node_is_not_alive() {
160        let m = HeartbeatMonitor::new(Duration::from_secs(1));
161        assert!(!m.is_alive(node(9), Instant::now()));
162        assert_eq!(m.state(node(9)), None);
163    }
164}
canopen_host/nmt.rs

canopen_host/
nmt.rs
