mecha10_behavior_runtime/actions/
set_instance_track_target.rs

1//! Action node that relays live target requests to the `instance-track` node's control topic.
2//!
3//! Mirrors `set_rl_target.rs`, but relays a `bbox` instead of a `target_class`.
4
5use crate::{BehaviorNode, Context, NodeStatus};
6use async_trait::async_trait;
7use mecha10_core::messages::Message;
8use mecha10_core::topics::Topic;
9use mecha10_core::Receiver;
10use serde::{Deserialize, Serialize};
11use serde_json::Value;
12
13#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
14pub(crate) struct BoundingBox {
15    x: f32,
16    y: f32,
17    width: f32,
18    height: f32,
19}
20
21#[derive(Debug, Clone, Serialize, Deserialize)]
22struct InstanceTrackControlCommand {
23    action: String,
24    #[serde(skip_serializing_if = "Option::is_none")]
25    params: Option<Value>,
26}
27
28impl Message for InstanceTrackControlCommand {}
29
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub(crate) struct InstanceTrackTargetRequest {
32    #[serde(default)]
33    pub(crate) bbox: Option<BoundingBox>,
34}
35
36impl Message for InstanceTrackTargetRequest {}
37
38fn default_topic() -> String {
39    "/instance-track/control".to_string()
40}
41
42/// The topic live target requests (clicks/drag-boxes on the dashboard's camera feed) arrive on.
43/// `pub(crate)` so `InstanceTrackTaskNode` can subscribe to the same topic to track whether a
44/// target is currently live, without hardcoding the string a second time.
45pub(crate) fn default_request_topic() -> String {
46    "/instance-track/target/request".to_string()
47}
48
49pub struct SetInstanceTrackTargetNode {
50    topic: String,
51    request_topic: String,
52    target: Option<BoundingBox>,
53    receiver: Option<Receiver<Value>>,
54    published_initial: bool,
55}
56
57impl std::fmt::Debug for SetInstanceTrackTargetNode {
58    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        f.debug_struct("SetInstanceTrackTargetNode")
60            .field("topic", &self.topic)
61            .field("request_topic", &self.request_topic)
62            .field("target", &self.target)
63            .field("subscribed", &self.receiver.is_some())
64            .field("published_initial", &self.published_initial)
65            .finish()
66    }
67}
68
69#[derive(Debug, Serialize, Deserialize)]
70struct SetInstanceTrackTargetConfig {
71    #[serde(default = "default_topic")]
72    topic: String,
73    #[serde(default = "default_request_topic")]
74    request_topic: String,
75    #[serde(default)]
76    bbox: Option<BoundingBox>,
77}
78
79impl SetInstanceTrackTargetNode {
80    /// Create a new SetInstanceTrackTargetNode from JSON configuration. See `SetRlTargetNode::from_json`'s
81    /// doc comment for why `null` is normalized to `{}` here.
82    pub fn from_json(config: Value) -> anyhow::Result<Self> {
83        let config = if config.is_null() {
84            serde_json::json!({})
85        } else {
86            config
87        };
88        let cfg: SetInstanceTrackTargetConfig = serde_json::from_value(config)?;
89        Ok(Self {
90            topic: cfg.topic,
91            request_topic: cfg.request_topic,
92            target: cfg.bbox,
93            receiver: None,
94            published_initial: false,
95        })
96    }
97
98    async fn publish_set_target(&self, ctx: &Context, bbox: BoundingBox) -> anyhow::Result<()> {
99        let command = InstanceTrackControlCommand {
100            action: "set_target".to_string(),
101            params: Some(serde_json::json!({ "bbox": bbox })),
102        };
103
104        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
105        let topic = Topic::<InstanceTrackControlCommand>::new(topic_static);
106
107        ctx.publish_to(topic, &command).await?;
108        tracing::debug!("SetInstanceTrackTargetNode: published set_target({:?})", bbox);
109        Ok(())
110    }
111
112    async fn publish_clear(&self, ctx: &Context) -> anyhow::Result<()> {
113        let command = InstanceTrackControlCommand {
114            action: "clear".to_string(),
115            params: None,
116        };
117
118        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
119        let topic = Topic::<InstanceTrackControlCommand>::new(topic_static);
120
121        ctx.publish_to(topic, &command).await?;
122        tracing::debug!("SetInstanceTrackTargetNode: published clear");
123        Ok(())
124    }
125
126    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
127    /// the most recent request matters, same as `SetRlTargetNode::poll_latest_request`. A
128    /// request with `bbox: null` (or no `bbox` key at all) means "clear".
129    fn poll_latest_request(receiver: &mut Receiver<Value>) -> Option<Option<BoundingBox>> {
130        let mut latest = None;
131        while let Ok(value) = receiver.try_recv() {
132            match serde_json::from_value::<InstanceTrackTargetRequest>(value) {
133                Ok(request) => latest = Some(request.bbox),
134                Err(e) => tracing::warn!("SetInstanceTrackTargetNode: dropping malformed target request: {}", e),
135            }
136        }
137        latest
138    }
139}
140
141#[async_trait]
142impl BehaviorNode for SetInstanceTrackTargetNode {
143    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
144        if self.receiver.is_none() {
145            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
146        }
147
148        if !self.published_initial {
149            self.published_initial = true;
150            if let Some(bbox) = self.target {
151                self.publish_set_target(ctx, bbox).await?;
152            }
153        }
154
155        let latest = Self::poll_latest_request(self.receiver.as_mut().expect("just subscribed above"));
156        if let Some(bbox) = latest {
157            self.target = bbox;
158            match bbox {
159                Some(bbox) => self.publish_set_target(ctx, bbox).await?,
160                None => self.publish_clear(ctx).await?,
161            }
162        }
163
164        Ok(if self.target.is_some() {
165            NodeStatus::Running
166        } else {
167            NodeStatus::Failure
168        })
169    }
170
171    fn name(&self) -> &str {
172        "SetInstanceTrackTargetNode"
173    }
174}
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179    use crate::test_helpers::create_test_context;
180
181    fn sample_bbox() -> BoundingBox {
182        BoundingBox {
183            x: 0.1,
184            y: 0.2,
185            width: 0.3,
186            height: 0.4,
187        }
188    }
189
190    #[test]
191    fn from_json_reads_topic_request_topic_and_bbox() {
192        let node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
193            "topic": "/instance-track/control",
194            "request_topic": "/instance-track/target/request",
195            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
196        }))
197        .unwrap();
198
199        assert_eq!(node.topic, "/instance-track/control");
200        assert_eq!(node.request_topic, "/instance-track/target/request");
201        assert_eq!(node.target, Some(sample_bbox()));
202    }
203
204    #[test]
205    fn from_json_defaults_topics_and_allows_no_target() {
206        let node = SetInstanceTrackTargetNode::from_json(serde_json::json!({})).unwrap();
207
208        assert_eq!(node.topic, "/instance-track/control");
209        assert_eq!(node.request_topic, "/instance-track/target/request");
210        assert_eq!(node.target, None);
211    }
212
213    #[test]
214    fn from_json_accepts_null_the_same_as_an_empty_object() {
215        let node = SetInstanceTrackTargetNode::from_json(Value::Null).unwrap();
216
217        assert_eq!(node.topic, "/instance-track/control");
218        assert_eq!(node.request_topic, "/instance-track/target/request");
219        assert_eq!(node.target, None);
220    }
221
222    #[test]
223    fn poll_latest_request_returns_none_when_empty() {
224        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
225        let mut receiver = Receiver::from(rx);
226        assert_eq!(SetInstanceTrackTargetNode::poll_latest_request(&mut receiver), None);
227    }
228
229    #[test]
230    fn poll_latest_request_keeps_only_the_most_recent_value() {
231        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
232        let mut receiver = Receiver::from(rx);
233        tx.send(serde_json::json!({ "bbox": { "x": 0.0, "y": 0.0, "width": 0.1, "height": 0.1 } }))
234            .unwrap();
235        tx.send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
236            .unwrap();
237
238        assert_eq!(
239            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
240            Some(Some(sample_bbox()))
241        );
242    }
243
244    #[test]
245    fn poll_latest_request_accepts_a_null_bbox_as_a_clear_request() {
246        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
247        let mut receiver = Receiver::from(rx);
248        tx.send(serde_json::json!({ "bbox": null })).unwrap();
249
250        assert_eq!(
251            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
252            Some(None)
253        );
254    }
255
256    #[test]
257    fn poll_latest_request_accepts_an_absent_bbox_key_as_a_clear_request() {
258        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
259        let mut receiver = Receiver::from(rx);
260        tx.send(serde_json::json!({})).unwrap();
261
262        assert_eq!(
263            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
264            Some(None)
265        );
266    }
267
268    #[test]
269    fn poll_latest_request_drops_malformed_requests_and_keeps_polling() {
270        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
271        let mut receiver = Receiver::from(rx);
272        tx.send(serde_json::json!({ "bbox": "not-a-bbox" })).unwrap();
273        tx.send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
274            .unwrap();
275
276        assert_eq!(
277            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
278            Some(Some(sample_bbox()))
279        );
280    }
281
282    #[tokio::test]
283    #[ignore]
284    async fn tick_publishes_configured_target_once_then_stays_running() {
285        let ctx = create_test_context("test_set_instance_track_target").await;
286
287        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
288            "topic": "/instance-track/control",
289            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
290        }))
291        .unwrap();
292
293        let mut receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
294
295        let status = node.tick(&ctx).await.unwrap();
296        assert_eq!(status, NodeStatus::Running);
297
298        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
299            .await
300            .expect("timed out waiting for published command")
301            .expect("channel closed without a message");
302
303        assert_eq!(received["action"], "set_target");
304        assert_eq!(received["params"]["bbox"]["x"], 0.1);
305
306        let status = node.tick(&ctx).await.unwrap();
307        assert_eq!(status, NodeStatus::Running);
308        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
309        assert!(timed_out.is_err(), "expected no second publish without a new request");
310    }
311
312    #[tokio::test]
313    #[ignore]
314    async fn tick_relays_a_live_request_published_on_request_topic() {
315        let ctx = create_test_context("test_set_instance_track_target_request").await;
316
317        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
318            "topic": "/instance-track/control",
319            "request_topic": "/instance-track/target/request"
320        }))
321        .unwrap();
322
323        node.tick(&ctx).await.unwrap();
324
325        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
326        ctx.publish_raw(
327            "/instance-track/target/request",
328            &serde_json::json!({ "bbox": { "x": 0.5, "y": 0.5, "width": 0.1, "height": 0.1 } }),
329        )
330        .await
331        .unwrap();
332
333        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
334        node.tick(&ctx).await.unwrap();
335
336        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
337            .await
338            .expect("timed out waiting for relayed command")
339            .expect("channel closed without a message");
340
341        assert_eq!(received["action"], "set_target");
342        assert_eq!(received["params"]["bbox"]["x"], 0.5);
343        assert_eq!(
344            node.tick(&ctx).await.unwrap(),
345            NodeStatus::Running,
346            "a live target must report Running"
347        );
348    }
349
350    #[tokio::test]
351    #[ignore = "Requires Redis infrastructure"]
352    async fn tick_returns_failure_when_no_target_has_ever_been_set() {
353        let ctx = create_test_context("test_set_instance_track_target_idle").await;
354
355        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
356            "request_topic": "/instance-track/target/request"
357        }))
358        .unwrap();
359
360        let status = node.tick(&ctx).await.unwrap();
361        assert_eq!(
362            status,
363            NodeStatus::Failure,
364            "no target yet means cede control, not Success"
365        );
366    }
367
368    #[tokio::test]
369    #[ignore = "Requires Redis infrastructure"]
370    async fn tick_returns_failure_after_a_clear_request_even_with_a_configured_initial_target() {
371        let ctx = create_test_context("test_set_instance_track_target_cleared").await;
372
373        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
374            "request_topic": "/instance-track/target/request",
375            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
376        }))
377        .unwrap();
378
379        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Running);
380
381        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
382            .await
383            .unwrap();
384        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
385
386        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Failure);
387    }
388
389    #[tokio::test]
390    #[ignore]
391    async fn tick_relays_a_clear_request_published_on_request_topic() {
392        let ctx = create_test_context("test_set_instance_track_target_clear").await;
393
394        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
395            "topic": "/instance-track/control",
396            "request_topic": "/instance-track/target/request"
397        }))
398        .unwrap();
399
400        node.tick(&ctx).await.unwrap();
401
402        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
403        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
404            .await
405            .unwrap();
406
407        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
408        node.tick(&ctx).await.unwrap();
409
410        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
411            .await
412            .expect("timed out waiting for relayed command")
413            .expect("channel closed without a message");
414
415        assert_eq!(received["action"], "clear");
416    }
417}
mecha10_behavior_runtime/actions/set_instance_track_target.rs

mecha10_behavior_runtime/actions/
set_instance_track_target.rs
