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    fn evaluate(&mut self) -> NodeStatus {
75        if let Some(receiver) = self.receiver.as_mut() {
76            while let Ok(msg) = receiver.try_recv() {
77                self.latest = Some(msg);
78            }
79        }
80
81        let Some(latest) = self.latest.as_ref() else {
82            tracing::debug!(topic = %self.topic, "SensorCheckNode: waiting for first sensor reading");
83            return NodeStatus::Running;
84        };
85
86        let Some(actual) = latest.get(&self.field).and_then(Value::as_f64) else {
87            tracing::warn!(
88                topic = %self.topic,
89                field = %self.field,
90                message = %latest,
91                "SensorCheckNode: field missing or not numeric in sensor message"
92            );
93            return NodeStatus::Failure;
94        };
95
96        let met = self.operator.evaluate(actual, self.threshold);
97        tracing::debug!(
98            "SensorCheckNode: checking {} {} {} -> {}",
99            actual,
100            match self.operator {
101                ComparisonOperator::LessThan => "<",
102                ComparisonOperator::GreaterThan => ">",
103                ComparisonOperator::Equal => "==",
104            },
105            self.threshold,
106            met
107        );
108
109        if met {
110            NodeStatus::Success
111        } else {
112            NodeStatus::Failure
113        }
114    }
115
116    /// Construct a node with an already-connected receiver, bypassing live subscription.
117    ///
118    /// Used by unit tests to exercise `evaluate()` without requiring a real `Context`/Redis.
119    #[cfg(test)]
120    fn with_receiver(field: &str, operator: ComparisonOperator, threshold: f64, receiver: Receiver<Value>) -> Self {
121        Self {
122            topic: "test/topic".to_string(),
123            field: field.to_string(),
124            operator,
125            threshold,
126            receiver: Some(receiver),
127            latest: None,
128        }
129    }
130}
131
132#[async_trait]
133impl BehaviorNode for SensorCheckNode {
134    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
135        if self.receiver.is_none() {
136            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.topic).await?);
137        }
138        Ok(self.evaluate())
139    }
140
141    fn name(&self) -> &str {
142        "SensorCheckNode"
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149    use serde_json::json;
150    use tokio::sync::mpsc;
151
152    fn node_with_channel(
153        field: &str,
154        operator: ComparisonOperator,
155        threshold: f64,
156    ) -> (SensorCheckNode, mpsc::UnboundedSender<Value>) {
157        let (tx, rx) = mpsc::unbounded_channel();
158        let node = SensorCheckNode::with_receiver(field, operator, threshold, Receiver::from(rx));
159        (node, tx)
160    }
161
162    #[test]
163    fn no_message_received_returns_running() {
164        let (mut node, _tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
165        assert_eq!(node.evaluate(), NodeStatus::Running);
166    }
167
168    #[test]
169    fn missing_field_returns_failure() {
170        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
171        tx.send(json!({ "other_field": 1.0 })).unwrap();
172        assert_eq!(node.evaluate(), NodeStatus::Failure);
173    }
174
175    #[test]
176    fn non_numeric_field_returns_failure() {
177        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
178        tx.send(json!({ "distance": "close" })).unwrap();
179        assert_eq!(node.evaluate(), NodeStatus::Failure);
180    }
181
182    #[test]
183    fn less_than_success_when_below_threshold() {
184        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
185        tx.send(json!({ "distance": 0.2 })).unwrap();
186        assert_eq!(node.evaluate(), NodeStatus::Success);
187    }
188
189    #[test]
190    fn less_than_failure_when_at_or_above_threshold() {
191        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
192        tx.send(json!({ "distance": 0.5 })).unwrap();
193        assert_eq!(node.evaluate(), NodeStatus::Failure);
194
195        tx.send(json!({ "distance": 1.0 })).unwrap();
196        assert_eq!(node.evaluate(), NodeStatus::Failure);
197    }
198
199    #[test]
200    fn greater_than_success_when_above_threshold() {
201        let (mut node, tx) = node_with_channel("battery", ComparisonOperator::GreaterThan, 20.0);
202        tx.send(json!({ "battery": 55.0 })).unwrap();
203        assert_eq!(node.evaluate(), NodeStatus::Success);
204    }
205
206    #[test]
207    fn greater_than_failure_when_at_or_below_threshold() {
208        let (mut node, tx) = node_with_channel("battery", ComparisonOperator::GreaterThan, 20.0);
209        tx.send(json!({ "battery": 20.0 })).unwrap();
210        assert_eq!(node.evaluate(), NodeStatus::Failure);
211
212        tx.send(json!({ "battery": 5.0 })).unwrap();
213        assert_eq!(node.evaluate(), NodeStatus::Failure);
214    }
215
216    #[test]
217    fn equal_success_when_matching() {
218        let (mut node, tx) = node_with_channel("state", ComparisonOperator::Equal, 1.0);
219        tx.send(json!({ "state": 1.0 })).unwrap();
220        assert_eq!(node.evaluate(), NodeStatus::Success);
221    }
222
223    #[test]
224    fn equal_failure_when_not_matching() {
225        let (mut node, tx) = node_with_channel("state", ComparisonOperator::Equal, 1.0);
226        tx.send(json!({ "state": 0.0 })).unwrap();
227        assert_eq!(node.evaluate(), NodeStatus::Failure);
228    }
229
230    #[test]
231    fn stale_topic_reuses_last_known_reading() {
232        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
233
234        tx.send(json!({ "distance": 0.1 })).unwrap();
235        assert_eq!(node.evaluate(), NodeStatus::Success);
236
237        assert_eq!(node.evaluate(), NodeStatus::Success);
238        assert_eq!(node.evaluate(), NodeStatus::Success);
239    }
240
241    #[test]
242    fn later_message_overrides_earlier_one() {
243        let (mut node, tx) = node_with_channel("distance", ComparisonOperator::LessThan, 0.5);
244
245        tx.send(json!({ "distance": 0.1 })).unwrap();
246        tx.send(json!({ "distance": 1.0 })).unwrap();
247
248        assert_eq!(node.evaluate(), NodeStatus::Failure);
249    }
250
251    #[test]
252    fn from_json_parses_all_operators() {
253        for operator in ["less_than", "greater_than", "equal"] {
254            let config = json!({
255                "topic": "/sensors/distance",
256                "field": "value",
257                "operator": operator,
258                "threshold": 0.5
259            });
260            assert!(SensorCheckNode::from_json(config).is_ok());
261        }
262    }
263
264    #[test]
265    fn from_json_rejects_unknown_operator() {
266        let config = json!({
267            "topic": "/sensors/distance",
268            "field": "value",
269            "operator": "not_a_real_operator",
270            "threshold": 0.5
271        });
272        assert!(SensorCheckNode::from_json(config).is_err());
273    }
274}
mecha10_behavior_runtime/actions/sensing.rs

mecha10_behavior_runtime/actions/
sensing.rs
