mecha10_core/
topic_utils.rs

1//! Topic utility functions
2//!
3//! Provides common utilities for working with topics including:
4//! - Topic path parsing and categorization
5//! - Time formatting for topic metadata
6//! - Topic metadata extraction from Redis streams
7//! - Per-role publish/subscribe topic configuration for node configs
8
9use schemars::JsonSchema;
10use serde::{Deserialize, Serialize};
11use std::collections::HashMap;
12
13/// Extract category from topic path (first segment)
14///
15/// # Examples
16/// ```
17/// use mecha10_core::topic_utils::extract_category;
18///
19/// assert_eq!(extract_category("/sensor/camera"), "/sensor");
20/// assert_eq!(extract_category("/actuator/motor/left"), "/actuator");
21/// assert_eq!(extract_category("unknown"), "/unknown");
22/// assert_eq!(extract_category(""), "uncategorized");
23/// ```
24pub fn extract_category(topic: &str) -> String {
25    let parts: Vec<&str> = topic.trim_start_matches('/').split('/').collect();
26    if parts.is_empty() || parts[0].is_empty() {
27        "uncategorized".to_string()
28    } else {
29        format!("/{}", parts[0])
30    }
31}
32
33/// Format elapsed time in human-readable format
34///
35/// # Examples
36/// ```
37/// use mecha10_core::topic_utils::format_elapsed_time;
38///
39/// assert_eq!(format_elapsed_time(500), "500ms ago");
40/// assert_eq!(format_elapsed_time(5000), "5s ago");
41/// assert_eq!(format_elapsed_time(65000), "1m ago");
42/// ```
43pub fn format_elapsed_time(elapsed_ms: u64) -> String {
44    if elapsed_ms < 1000 {
45        format!("{}ms ago", elapsed_ms)
46    } else if elapsed_ms < 60_000 {
47        format!("{}s ago", elapsed_ms / 1000)
48    } else if elapsed_ms < 3_600_000 {
49        format!("{}m ago", elapsed_ms / 60_000)
50    } else if elapsed_ms < 86_400_000 {
51        format!("{}h ago", elapsed_ms / 3_600_000)
52    } else {
53        format!("{}d ago", elapsed_ms / 86_400_000)
54    }
55}
56
57#[derive(Debug, Clone)]
58pub struct TopicMetadata {
59    pub topic: String,
60    pub message_count: Option<usize>,
61    pub last_update_ms: Option<u64>,
62    pub first_timestamp_ms: Option<u64>,
63    pub last_timestamp_ms: Option<u64>,
64}
65
66impl TopicMetadata {
67    /// Create new topic metadata
68    pub fn new(topic: String) -> Self {
69        Self {
70            topic,
71            message_count: None,
72            last_update_ms: None,
73            first_timestamp_ms: None,
74            last_timestamp_ms: None,
75        }
76    }
77
78    crate::builder_setter!(
79        /// Set message count
80        into_some, with_message_count, message_count, usize
81    );
82
83    crate::builder_setter!(
84        /// Set last update timestamp
85        into_some, with_last_update, last_update_ms, u64
86    );
87
88    /// Set timestamp range
89    pub fn with_timestamp_range(mut self, first_ms: u64, last_ms: u64) -> Self {
90        self.first_timestamp_ms = Some(first_ms);
91        self.last_timestamp_ms = Some(last_ms);
92        self
93    }
94
95    /// Calculate message rate (messages per second)
96    pub fn calculate_rate(&self) -> Option<f64> {
97        if let (Some(first_ms), Some(last_ms), Some(count)) =
98            (self.first_timestamp_ms, self.last_timestamp_ms, self.message_count)
99        {
100            let duration_ms = last_ms.saturating_sub(first_ms);
101            if duration_ms > 1000 && count > 0 {
102                return Some((count as f64 * 1000.0) / duration_ms as f64);
103            }
104        }
105        None
106    }
107
108    /// Get topic category
109    pub fn category(&self) -> String {
110        extract_category(&self.topic)
111    }
112}
113
114#[derive(Debug, Clone, Serialize, Deserialize, Default, JsonSchema, PartialEq)]
115pub struct Topics {
116    #[serde(default)]
117    pub publishes: Vec<HashMap<String, String>>,
118
119    #[serde(default)]
120    pub subscribes: Vec<HashMap<String, String>>,
121}
122
123impl Topics {
124    /// Look up a subscribed topic by role, falling back to `default` if not configured
125    pub fn subscribed_topic(&self, role: &str, default: &str) -> String {
126        Self::lookup(&self.subscribes, role, default)
127    }
128
129    /// Look up a published topic by role, falling back to `default` if not configured
130    pub fn published_topic(&self, role: &str, default: &str) -> String {
131        Self::lookup(&self.publishes, role, default)
132    }
133
134    fn lookup(topics: &[HashMap<String, String>], role: &str, default: &str) -> String {
135        topics
136            .iter()
137            .find_map(|t| t.get(role))
138            .cloned()
139            .unwrap_or_else(|| default.to_string())
140    }
141
142    /// Default topic wiring for a vision node: subscribes to `/camera/rgb`
143    /// input and a `control` topic, publishes a single `output` topic.
144    ///
145    /// Shared by node configs (`image-classifier`, `object-detector`) whose
146    /// `Default` impl wires up this same shape with different topic paths.
147    pub fn vision_defaults(output_topic: &str, control_topic: &str) -> Self {
148        Self {
149            publishes: vec![HashMap::from([("output".to_string(), output_topic.to_string())])],
150            subscribes: vec![
151                HashMap::from([("input".to_string(), "/camera/rgb".to_string())]),
152                HashMap::from([("control".to_string(), control_topic.to_string())]),
153            ],
154        }
155    }
156}
mecha10_core/topic_utils.rs

mecha10_core/
topic_utils.rs
