mecha10_behavior_runtime/actions/
sensing.rs

1//! Action nodes that inspect sensor state.
2
3use crate::{BehaviorNode, Context, NodeStatus, TreeStatusSnapshot};
4use async_trait::async_trait;
5use mecha10_core::Receiver;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9pub struct SensorCheckNode {
10    topic: String,
11    field: String,
12    operator: ComparisonOperator,
13    threshold: f64,
14    receiver: Option<Receiver<Value>>,
15    latest: Option<Value>,
16    last_status: Option<NodeStatus>,
17}
18
19impl std::fmt::Debug for SensorCheckNode {
20    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        f.debug_struct("SensorCheckNode")
22            .field("topic", &self.topic)
23            .field("field", &self.field)
24            .field("operator", &self.operator)
25            .field("threshold", &self.threshold)
26            .field("subscribed", &self.receiver.is_some())
27            .field("latest", &self.latest)
28            .finish()
29    }
30}
31
32#[derive(Debug, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34enum ComparisonOperator {
35    LessThan,
36    GreaterThan,
37    Equal,
38}
39
40impl ComparisonOperator {
41    /// Evaluate `actual <op> threshold`.
42    fn evaluate(&self, actual: f64, threshold: f64) -> bool {
43        match self {
44            ComparisonOperator::LessThan => actual < threshold,
45            ComparisonOperator::GreaterThan => actual > threshold,
46            ComparisonOperator::Equal => (actual - threshold).abs() < f64::EPSILON,
47        }
48    }
49}
50
51#[derive(Debug, Serialize, Deserialize)]
52struct SensorCheckConfig {
53    topic: String,
54    field: String,
55    operator: ComparisonOperator,
56    threshold: f64,
57}
58
59impl SensorCheckNode {
60    /// Create a new SensorCheckNode from JSON configuration
61    pub fn from_json(config: Value) -> anyhow::Result<Self> {
62        let cfg: SensorCheckConfig = serde_json::from_value(config)?;
63        Ok(Self {
64            topic: cfg.topic,
65            field: cfg.field,
66            operator: cfg.operator,
67            threshold: cfg.threshold,
68            receiver: None,
69            latest: None,
70            last_status: None,
71        })
72    }
73
74    /// Drain any pending sensor messages (keeping only the most recent) and evaluate the
75    /// configured condition against the latest known reading.
76    fn evaluate(&mut self) -> NodeStatus {
77        if let Some(receiver) = self.receiver.as_mut() {
78            while let Ok(msg) = receiver.try_recv() {
79                self.latest = Some(msg);
80            }
81        }
82
83        let Some(latest) = self.latest.as_ref() else {
84            tracing::debug!(topic = %self.topic, "SensorCheckNode: waiting for first sensor reading");
85            return NodeStatus::Running;
86        };
87
88        let Some(actual) = latest.get(&self.field).and_then(Value::as_f64) else {
89            tracing::warn!(
90                topic = %self.topic,
91                field = %self.field,
92                message = %latest,
93                "SensorCheckNode: field missing or not numeric in sensor message"
94            );
95            return NodeStatus::Failure;
96        };
97
98        let met = self.operator.evaluate(actual, self.threshold);
99        tracing::debug!(
100            "SensorCheckNode: checking {} {} {} -> {}",
101            actual,
102            match self.operator {
103                ComparisonOperator::LessThan => "<",
104                ComparisonOperator::GreaterThan => ">",
105                ComparisonOperator::Equal => "==",
106            },
107            self.threshold,
108            met
109        );
110
111        if met {
112            NodeStatus::Success
113        } else {
114            NodeStatus::Failure
115        }
116    }
117
118    /// Construct a node with an already-connected receiver, bypassing live subscription.
119    ///
120    /// Used by unit tests to exercise `evaluate()` without requiring a real `Context`/Redis.
121    #[cfg(test)]
122    fn with_receiver(field: &str, operator: ComparisonOperator, threshold: f64, receiver: Receiver<Value>) -> Self {
123        Self {
124            topic: "test/topic".to_string(),
125            field: field.to_string(),
126            operator,
127            threshold,
128            receiver: Some(receiver),
129            latest: None,
130            last_status: None,
131        }
132    }
133}
134
135#[async_trait]
136impl BehaviorNode for SensorCheckNode {
137    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
138        if self.receiver.is_none() {
139            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.topic).await?);
140        }
141        let status = self.evaluate();
142        self.last_status = Some(status);
143        Ok(status)
144    }
145
146    fn name(&self) -> &str {
147        "SensorCheckNode"
148    }
149
150    fn status_snapshot(&self) -> TreeStatusSnapshot {
151        TreeStatusSnapshot {
152            name: self.name().to_string(),
153            status: self.last_status,
154            active_capability: self.active_capability().map(String::from),
155            children: Vec::new(),
156        }
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163    use serde_json::json;
164    use tokio::sync::mpsc;
165
166    fn node_with_channel(
167        field: &str,
168        operator: ComparisonOperator,
169        threshold: f64,
170    ) -> (SensorCheckNode, mpsc::UnboundedSender<Value>) {
171        let (tx, rx) = mpsc::unbounded_channel();
172        let node = SensorCheckNode::with_receiver(field, operator, threshold, Receiver::from(rx));
173        (node, tx)
174    }
175
176    #[test]
177    fn no_message_received_returns_running() {
178        let (mut node, _tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
179        assert_eq!(node.evaluate(), NodeStatus::Running);
180    }
181
182    #[test]
183    fn missing_field_returns_failure() {
184        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
185        tx.send(json!({ "other_field": 1.0 })).unwrap();
186        assert_eq!(node.evaluate(), NodeStatus::Failure);
187    }
188
189    #[test]
190    fn non_numeric_field_returns_failure() {
191        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
192        tx.send(json!({ "distance": "close" })).unwrap();
193        assert_eq!(node.evaluate(), NodeStatus::Failure);
194    }
195
196    #[test]
197    fn less_than_success_when_below_threshold() {
198        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
199        tx.send(json!({ "distance": 0.2 })).unwrap();
200        assert_eq!(node.evaluate(), NodeStatus::Success);
201    }
202
203    #[test]
204    fn less_than_failure_when_at_or_above_threshold() {
205        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
206        tx.send(json!({ "distance": 0.5 })).unwrap();
207        assert_eq!(node.evaluate(), NodeStatus::Failure);
208
209        tx.send(json!({ "distance": 1.0 })).unwrap();
210        assert_eq!(node.evaluate(), NodeStatus::Failure);
211    }
212
213    #[test]
214    fn greater_than_success_when_above_threshold() {
215        let (mut node, tx) = node_with_channel("battery", ComparisonOperator::GreaterThan, 20.0);
216        tx.send(json!({ "battery": 55.0 })).unwrap();
217        assert_eq!(node.evaluate(), NodeStatus::Success);
218    }
219
220    #[test]
221    fn greater_than_failure_when_at_or_below_threshold() {
222        let (mut node, tx) = node_with_channel("battery", ComparisonOperator::GreaterThan, 20.0);
223        tx.send(json!({ "battery": 20.0 })).unwrap();
224        assert_eq!(node.evaluate(), NodeStatus::Failure);
225
226        tx.send(json!({ "battery": 5.0 })).unwrap();
227        assert_eq!(node.evaluate(), NodeStatus::Failure);
228    }
229
230    #[test]
231    fn equal_success_when_matching() {
232        let (mut node, tx) = node_with_channel("state", ComparisonOperator::Equal, 1.0);
233        tx.send(json!({ "state": 1.0 })).unwrap();
234        assert_eq!(node.evaluate(), NodeStatus::Success);
235    }
236
237    #[test]
238    fn equal_failure_when_not_matching() {
239        let (mut node, tx) = node_with_channel("state", ComparisonOperator::Equal, 1.0);
240        tx.send(json!({ "state": 0.0 })).unwrap();
241        assert_eq!(node.evaluate(), NodeStatus::Failure);
242    }
243
244    #[test]
245    fn stale_topic_reuses_last_known_reading() {
246        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
247
248        tx.send(json!({ "distance": 0.1 })).unwrap();
249        assert_eq!(node.evaluate(), NodeStatus::Success);
250
251        assert_eq!(node.evaluate(), NodeStatus::Success);
252        assert_eq!(node.evaluate(), NodeStatus::Success);
253    }
254
255    #[test]
256    fn later_message_overrides_earlier_one() {
257        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
258
259        tx.send(json!({ "distance": 0.1 })).unwrap();
260        tx.send(json!({ "distance": 1.0 })).unwrap();
261
262        assert_eq!(node.evaluate(), NodeStatus::Failure);
263    }
264
265    #[test]
266    fn from_json_parses_all_operators() {
267        for operator in ["less_than", "greater_than", "equal"] {
268            let config = json!({
269                "topic": "/sensors/distance",
270                "field": "value",
271                "operator": operator,
272                "threshold": 0.5
273            });
274            assert!(SensorCheckNode::from_json(config).is_ok());
275        }
276    }
277
278    #[test]
279    fn from_json_rejects_unknown_operator() {
280        let config = json!({
281            "topic": "/sensors/distance",
282            "field": "value",
283            "operator": "not_a_real_operator",
284            "threshold": 0.5
285        });
286        assert!(SensorCheckNode::from_json(config).is_err());
287    }
288}
mecha10_behavior_runtime/actions/sensing.rs

mecha10_behavior_runtime/actions/
sensing.rs
