mecha10_behavior_runtime/actions/
sensing.rs

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

mecha10_behavior_runtime/actions/
sensing.rs
