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