1#[cfg(not(feature = "std"))]
6extern crate alloc;
7
8use ConfigGraphs::{Missions, Simple};
9#[cfg(feature = "std")]
10use core::fmt;
11#[cfg(feature = "std")]
12use core::fmt::Display;
13use cu29_traits::{CuError, CuResult};
14use hashbrown::HashMap;
15pub use petgraph::Direction::Incoming;
16pub use petgraph::Direction::Outgoing;
17use petgraph::stable_graph::{EdgeIndex, NodeIndex, StableDiGraph};
18#[cfg(feature = "std")]
19use petgraph::visit::IntoEdgeReferences;
20use petgraph::visit::{Bfs, EdgeRef};
21use ron::Options;
22use ron::extensions::Extensions;
23use serde::{Deserialize, Deserializer, Serialize, Serializer};
24
25#[cfg(not(feature = "std"))]
26use alloc::boxed::Box;
27#[cfg(not(feature = "std"))]
28use alloc::collections::BTreeMap;
29#[cfg(not(feature = "std"))]
30use alloc::vec;
31#[cfg(feature = "std")]
32use std::collections::BTreeMap;
33
34#[cfg(not(feature = "std"))]
35mod imp {
36 pub use alloc::borrow::ToOwned;
37 pub use alloc::format;
38 pub use alloc::string::String;
39 pub use alloc::string::ToString;
40 pub use alloc::vec::Vec;
41}
42
43#[cfg(feature = "std")]
44mod imp {
45 pub use html_escape::encode_text;
46 pub use std::fs::read_to_string;
47}
48
49use imp::*;
50
51mod value;
52
53pub use value::{
54 ComponentConfig, ConfigError, ConstantConfig, ConstantNumber, ConstantStorage, Value,
55};
56
57pub type NodeId = u32;
60pub const DEFAULT_MISSION_ID: &str = "default";
61#[doc(hidden)]
63pub const DEFAULT_COPPERLIST_COUNT: usize = 2;
64
65#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
74#[repr(u8)]
75pub enum HandleContent {
76 #[serde(rename = "all", alias = "All")]
78 #[default]
79 All = 0,
80 #[serde(rename = "touched_only", alias = "TouchedOnly")]
82 TouchedOnly = 1,
83 #[serde(rename = "none", alias = "None")]
85 None = 2,
86}
87
88impl HandleContent {
89 #[allow(dead_code)] pub fn from_u8(v: u8) -> Self {
93 match v {
94 1 => HandleContent::TouchedOnly,
95 2 => HandleContent::None,
96 _ => HandleContent::All,
97 }
98 }
99}
100
101#[derive(Serialize, Deserialize, Debug, Clone)]
103pub struct NodeLogging {
104 #[serde(default = "default_as_true")]
105 enabled: bool,
106 #[serde(skip_serializing_if = "Option::is_none")]
107 codec: Option<String>,
108 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
109 #[serde(serialize_with = "serialize_ordered_map")]
110 codecs: HashMap<String, String>,
111 #[serde(default, skip_serializing_if = "is_default_handle_content")]
114 handle_content: HandleContent,
115}
116
117fn is_default_handle_content(c: &HandleContent) -> bool {
118 *c == HandleContent::default()
119}
120
121impl NodeLogging {
122 #[allow(dead_code)]
123 pub fn enabled(&self) -> bool {
124 self.enabled
125 }
126
127 #[allow(dead_code)]
128 pub fn codec(&self) -> Option<&str> {
129 self.codec.as_deref()
130 }
131
132 #[allow(dead_code)]
133 pub fn codecs(&self) -> &HashMap<String, String> {
134 &self.codecs
135 }
136
137 #[allow(dead_code)]
138 pub fn codec_for_msg_type(&self, msg_type: &str) -> Option<&str> {
139 self.codecs
140 .get(msg_type)
141 .map(String::as_str)
142 .or(self.codec.as_deref())
143 }
144
145 pub fn handle_content(&self) -> HandleContent {
148 self.handle_content
149 }
150}
151
152impl Default for NodeLogging {
153 fn default() -> Self {
154 Self {
155 enabled: true,
156 codec: None,
157 codecs: HashMap::new(),
158 handle_content: HandleContent::default(),
159 }
160 }
161}
162
163#[derive(Default, Debug, Copy, Clone, PartialEq, Eq)]
166pub enum Flavor {
167 #[default]
168 Task,
169 Bridge,
170}
171
172#[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq)]
177pub enum TaskKind {
178 #[serde(rename = "source", alias = "src")]
179 Source,
180 #[serde(rename = "task", alias = "regular", alias = "cutask")]
181 Regular,
182 #[serde(rename = "stateless_task", alias = "stateless")]
185 Stateless,
186 #[serde(rename = "sink", alias = "snk")]
187 Sink,
188}
189
190impl TaskKind {
191 #[allow(dead_code)]
192 pub fn as_str(&self) -> &'static str {
193 match self {
194 TaskKind::Source => "source",
195 TaskKind::Regular => "task",
196 TaskKind::Stateless => "stateless_task",
197 TaskKind::Sink => "sink",
198 }
199 }
200}
201
202pub const DEFAULT_BACKGROUND_POOL: &str = "background";
204
205#[allow(dead_code)] pub const RT_POOL: &str = "rt";
209
210#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
216#[serde(untagged)]
217pub enum BackgroundConfig {
218 Flag(bool),
220 Pool { pool: String },
222}
223
224#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
238pub struct AnytimeConfig {
239 #[serde(skip_serializing_if = "Option::is_none")]
244 pub time_budget_ms: Option<f64>,
245
246 #[serde(skip_serializing_if = "Option::is_none")]
250 pub max_age_ms: Option<f64>,
251
252 #[serde(skip_serializing_if = "Option::is_none")]
256 pub quality_target: Option<f32>,
257
258 #[serde(skip_serializing_if = "Option::is_none")]
262 pub quality_floor: Option<f32>,
263
264 #[serde(skip_serializing_if = "Option::is_none")]
266 pub max_refines: Option<u32>,
267
268 #[serde(skip_serializing_if = "Option::is_none")]
271 pub max_stall: Option<u32>,
272}
273
274impl AnytimeConfig {
275 fn validate(&self, task_id: &str) -> CuResult<()> {
282 if let Some(budget) = self.time_budget_ms {
283 let valid = budget.is_finite() && budget > 0.0;
284 if !valid {
285 return Err(CuError::from(format!(
286 "Task '{task_id}': anytime.time_budget_ms must be a positive number of milliseconds (got {budget})."
287 )));
288 }
289 }
290 if let Some(age) = self.max_age_ms {
291 let valid = age.is_finite() && age > 0.0;
292 if !valid {
293 return Err(CuError::from(format!(
294 "Task '{task_id}': anytime.max_age_ms must be a positive number of milliseconds (got {age})."
295 )));
296 }
297 }
298 if let Some(target) = self.quality_target {
299 let valid = target > 0.0 && target <= 1.0;
300 if !valid {
301 return Err(CuError::from(format!(
302 "Task '{task_id}': anytime.quality_target must be within (0.0, 1.0] (got {target})."
303 )));
304 }
305 }
306 if let Some(floor) = self.quality_floor {
307 let valid = floor > 0.0 && floor < 1.0;
308 if !valid {
309 return Err(CuError::from(format!(
310 "Task '{task_id}': anytime.quality_floor must be within (0.0, 1.0) (got {floor})."
311 )));
312 }
313 }
314 if let Some(refines) = self.max_refines
315 && refines == 0
316 {
317 return Err(CuError::from(format!(
318 "Task '{task_id}': anytime.max_refines must be at least 1."
319 )));
320 }
321 if let Some(stall) = self.max_stall
322 && stall == 0
323 {
324 return Err(CuError::from(format!(
325 "Task '{task_id}': anytime.max_stall must be at least 1."
326 )));
327 }
328 if let (Some(floor), Some(target)) = (self.quality_floor, self.quality_target)
329 && floor > target
330 {
331 return Err(CuError::from(format!(
332 "Task '{task_id}': anytime.quality_floor ({floor}) must not exceed anytime.quality_target ({target}): refinement could stop at the target and then always discard the result."
333 )));
334 }
335 if self.time_budget_ms.is_none() && self.max_age_ms.is_none() && self.max_refines.is_none()
336 {
337 return Err(CuError::from(format!(
338 "Task '{task_id}': anytime needs at least one hard bound: set time_budget_ms, max_age_ms or max_refines. quality_target, max_stall and quality_floor alone leave refinement unbounded."
339 )));
340 }
341 Ok(())
342 }
343}
344
345#[derive(Serialize, Deserialize, Debug, Clone, Copy, Default, PartialEq, Eq)]
347#[serde(rename_all = "snake_case")]
348pub enum StreamReplay {
349 #[default]
350 Capture,
351 Reconstruct,
352}
353
354#[derive(Serialize, Deserialize, Debug, Clone, Copy, Default)]
356#[serde(deny_unknown_fields)]
357pub struct NodeStreaming {
358 #[serde(default)]
359 pub replay: StreamReplay,
360}
361
362#[derive(Serialize, Deserialize, Debug, Clone)]
365pub struct Node {
366 id: String,
368
369 #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
371 type_: Option<String>,
372
373 #[serde(skip_serializing_if = "Option::is_none")]
376 kind: Option<TaskKind>,
377
378 #[serde(skip_serializing_if = "Option::is_none")]
380 config: Option<ComponentConfig>,
381
382 #[serde(skip_serializing_if = "Option::is_none")]
384 #[serde(serialize_with = "serialize_optional_ordered_map")]
385 resources: Option<HashMap<String, String>>,
386
387 missions: Option<Vec<String>>,
389
390 #[serde(skip_serializing_if = "Option::is_none")]
397 background: Option<BackgroundConfig>,
398
399 #[serde(skip_serializing_if = "Option::is_none")]
402 background_process_empty: Option<bool>,
403
404 #[serde(skip_serializing_if = "Option::is_none")]
409 anytime: Option<AnytimeConfig>,
410
411 #[serde(skip_serializing_if = "Option::is_none")]
417 run_in_sim: Option<bool>,
418
419 #[serde(skip_serializing_if = "Option::is_none")]
421 logging: Option<NodeLogging>,
422
423 #[serde(skip_serializing_if = "Option::is_none")]
424 streaming: Option<NodeStreaming>,
425
426 #[serde(skip, default)]
428 flavor: Flavor,
429 #[serde(skip, default)]
431 nc_outputs: Vec<String>,
432 #[serde(skip, default)]
434 nc_output_orders: Vec<usize>,
435}
436
437impl Node {
438 #[allow(dead_code)]
439 pub fn new(id: &str, ptype: &str) -> Self {
440 Node {
441 id: id.to_string(),
442 type_: Some(ptype.to_string()),
443 kind: None,
444 config: None,
445 resources: None,
446 missions: None,
447 background: None,
448 background_process_empty: None,
449 anytime: None,
450 run_in_sim: None,
451 logging: None,
452 streaming: None,
453 flavor: Flavor::Task,
454 nc_outputs: Vec::new(),
455 nc_output_orders: Vec::new(),
456 }
457 }
458
459 #[allow(dead_code)]
460 pub fn new_with_flavor(id: &str, ptype: &str, flavor: Flavor) -> Self {
461 let mut node = Self::new(id, ptype);
462 node.flavor = flavor;
463 node
464 }
465
466 #[allow(dead_code)]
467 pub fn get_id(&self) -> String {
468 self.id.clone()
469 }
470
471 #[allow(dead_code)]
472 pub fn get_type(&self) -> &str {
473 self.type_.as_ref().unwrap()
474 }
475
476 #[allow(dead_code)]
477 pub fn set_type(mut self, name: Option<String>) -> Self {
478 self.type_ = name;
479 self
480 }
481
482 #[allow(dead_code)]
483 pub fn get_declared_task_kind(&self) -> Option<TaskKind> {
484 self.kind
485 }
486
487 #[allow(dead_code)]
488 pub fn set_task_kind(&mut self, kind: Option<TaskKind>) {
489 self.kind = kind;
490 }
491
492 #[allow(dead_code)]
493 pub fn set_resources<I>(&mut self, resources: Option<I>)
494 where
495 I: IntoIterator<Item = (String, String)>,
496 {
497 self.resources = resources.map(|iter| iter.into_iter().collect());
498 }
499
500 #[allow(dead_code)]
501 pub fn is_background(&self) -> bool {
502 match &self.background {
503 Some(BackgroundConfig::Flag(flag)) => *flag,
504 Some(BackgroundConfig::Pool { .. }) => true,
505 None => false,
506 }
507 }
508
509 #[doc(hidden)]
511 #[allow(dead_code)]
512 pub fn background_process_empty(&self) -> bool {
513 self.background_process_empty.unwrap_or(false)
514 }
515
516 #[allow(dead_code)]
519 pub fn background_pool(&self) -> &str {
520 match &self.background {
521 Some(BackgroundConfig::Pool { pool }) => pool.as_str(),
522 _ => DEFAULT_BACKGROUND_POOL,
523 }
524 }
525
526 #[allow(dead_code)]
527 pub fn is_anytime(&self) -> bool {
528 self.anytime.is_some()
529 }
530
531 #[allow(dead_code)]
533 pub fn anytime(&self) -> Option<&AnytimeConfig> {
534 self.anytime.as_ref()
535 }
536
537 #[allow(dead_code)]
539 pub fn set_anytime(&mut self, anytime: Option<AnytimeConfig>) {
540 self.anytime = anytime;
541 }
542
543 #[allow(dead_code)]
544 pub fn get_instance_config(&self) -> Option<&ComponentConfig> {
545 self.config.as_ref()
546 }
547
548 #[allow(dead_code)]
549 pub fn get_resources(&self) -> Option<&HashMap<String, String>> {
550 self.resources.as_ref()
551 }
552
553 #[allow(dead_code)]
556 pub fn is_run_in_sim(&self) -> bool {
557 self.run_in_sim.unwrap_or(false)
558 }
559
560 #[allow(dead_code)]
561 pub fn is_logging_enabled(&self) -> bool {
562 if let Some(logging) = &self.logging {
563 logging.enabled()
564 } else {
565 true
566 }
567 }
568
569 #[allow(dead_code)]
573 pub fn handle_content_policy(&self) -> HandleContent {
574 self.logging
575 .as_ref()
576 .map(NodeLogging::handle_content)
577 .unwrap_or_default()
578 }
579
580 #[allow(dead_code)]
581 pub fn streaming(&self) -> NodeStreaming {
582 self.streaming.unwrap_or_default()
583 }
584
585 #[allow(dead_code)]
586 pub fn get_logging(&self) -> Option<&NodeLogging> {
587 self.logging.as_ref()
588 }
589
590 #[allow(dead_code)]
591 pub fn get_param<T>(&self, key: &str) -> Result<Option<T>, ConfigError>
592 where
593 T: for<'a> TryFrom<&'a Value, Error = ConfigError>,
594 {
595 let pc = match self.config.as_ref() {
596 Some(pc) => pc,
597 None => return Ok(None),
598 };
599 let ComponentConfig(pc) = pc;
600 match pc.get(key) {
601 Some(v) => T::try_from(v).map(Some),
602 None => Ok(None),
603 }
604 }
605
606 #[allow(dead_code)]
607 pub fn set_param<T: Into<Value>>(&mut self, key: &str, value: T) {
608 if self.config.is_none() {
609 self.config = Some(ComponentConfig(HashMap::new()));
610 }
611 let ComponentConfig(config) = self.config.as_mut().unwrap();
612 config.insert(key.to_string(), value.into());
613 }
614
615 #[allow(dead_code)]
617 pub fn get_flavor(&self) -> Flavor {
618 self.flavor
619 }
620
621 #[allow(dead_code)]
623 pub fn set_flavor(&mut self, flavor: Flavor) {
624 self.flavor = flavor;
625 }
626
627 #[allow(dead_code)]
629 pub fn add_nc_output(&mut self, msg_type: &str, order: usize) {
630 if let Some(pos) = self
631 .nc_outputs
632 .iter()
633 .position(|existing| existing == msg_type)
634 {
635 if order < self.nc_output_orders[pos] {
636 self.nc_output_orders[pos] = order;
637 }
638 return;
639 }
640 self.nc_outputs.push(msg_type.to_string());
641 self.nc_output_orders.push(order);
642 }
643
644 #[allow(dead_code)]
646 pub fn nc_outputs(&self) -> &[String] {
647 &self.nc_outputs
648 }
649
650 #[allow(dead_code)]
652 pub fn nc_outputs_with_order(&self) -> impl Iterator<Item = (&String, usize)> {
653 self.nc_outputs
654 .iter()
655 .zip(self.nc_output_orders.iter().copied())
656 }
657}
658
659#[derive(Serialize, Deserialize, Debug, Clone)]
661pub enum BridgeChannelConfigRepresentation {
662 Rx {
664 id: String,
665 #[serde(skip_serializing_if = "Option::is_none")]
667 route: Option<String>,
668 #[serde(skip_serializing_if = "Option::is_none")]
670 config: Option<ComponentConfig>,
671 },
672 Tx {
674 id: String,
675 #[serde(skip_serializing_if = "Option::is_none")]
677 route: Option<String>,
678 #[serde(skip_serializing_if = "Option::is_none")]
680 config: Option<ComponentConfig>,
681 },
682}
683
684impl BridgeChannelConfigRepresentation {
685 #[allow(dead_code)]
687 pub fn id(&self) -> &str {
688 match self {
689 BridgeChannelConfigRepresentation::Rx { id, .. }
690 | BridgeChannelConfigRepresentation::Tx { id, .. } => id,
691 }
692 }
693
694 #[allow(dead_code)]
696 pub fn route(&self) -> Option<&str> {
697 match self {
698 BridgeChannelConfigRepresentation::Rx { route, .. }
699 | BridgeChannelConfigRepresentation::Tx { route, .. } => route.as_deref(),
700 }
701 }
702}
703
704enum EndpointRole {
705 Source,
706 Destination,
707}
708
709fn validate_bridge_channel(
710 bridge: &BridgeConfig,
711 channel_id: &str,
712 role: EndpointRole,
713) -> Result<(), String> {
714 let channel = bridge
715 .channels
716 .iter()
717 .find(|ch| ch.id() == channel_id)
718 .ok_or_else(|| {
719 format!(
720 "Bridge '{}' does not declare a channel named '{}'",
721 bridge.id, channel_id
722 )
723 })?;
724
725 match (role, channel) {
726 (EndpointRole::Source, BridgeChannelConfigRepresentation::Rx { .. }) => Ok(()),
727 (EndpointRole::Destination, BridgeChannelConfigRepresentation::Tx { .. }) => Ok(()),
728 (EndpointRole::Source, BridgeChannelConfigRepresentation::Tx { .. }) => Err(format!(
729 "Bridge '{}' channel '{}' is Tx and cannot act as a source",
730 bridge.id, channel_id
731 )),
732 (EndpointRole::Destination, BridgeChannelConfigRepresentation::Rx { .. }) => Err(format!(
733 "Bridge '{}' channel '{}' is Rx and cannot act as a destination",
734 bridge.id, channel_id
735 )),
736 }
737}
738
739#[derive(Serialize, Deserialize, Debug, Clone)]
741pub struct ResourceBundleConfig {
742 #[serde(default, skip_serializing_if = "Option::is_none")]
744 #[serde(serialize_with = "serialize_optional_ordered_map")]
745 pub resources: Option<HashMap<String, String>>,
746 pub id: String,
747 #[serde(rename = "provider")]
748 pub provider: String,
749 #[serde(skip_serializing_if = "Option::is_none")]
750 pub config: Option<ComponentConfig>,
751 #[serde(skip_serializing_if = "Option::is_none")]
752 pub missions: Option<Vec<String>>,
753}
754
755#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
757#[serde(deny_unknown_fields)]
758pub struct LogStreamingConfig {
759 pub destinations: Vec<LogStreamDestinationConfig>,
760}
761
762#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
764#[serde(deny_unknown_fields)]
765pub struct LogStreamDestinationConfig {
766 #[serde(default, skip_serializing_if = "Option::is_none")]
767 pub feedback: Option<LogStreamFeedbackConfig>,
768 pub id: String,
769 pub transport: LogStreamTransportConfig,
770 pub link: LogStreamLinkConfig,
771 pub fec: LogStreamFecConfig,
772 pub recovery_interval: u32,
774 pub max_record_bytes: u64,
775}
776
777#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
779#[serde(deny_unknown_fields)]
780pub struct LogStreamFeedbackConfig {
781 pub transport: LogStreamTransportConfig,
782 pub report_interval_ms: u32,
783 pub timeout_ms: u32,
784 #[serde(default, skip_serializing_if = "Option::is_none")]
785 pub adaptation: Option<LogStreamAdaptationConfig>,
786}
787
788#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
790#[serde(deny_unknown_fields)]
791pub struct LogStreamAdaptationConfig {
792 pub min_repair_every_source_symbols: u16,
793 pub max_repair_every_source_symbols: u16,
794}
795
796#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
798#[serde(deny_unknown_fields)]
799pub struct LogStreamTransportConfig {
800 #[serde(rename = "type")]
801 pub type_: String,
802 pub resource: String,
803}
804
805#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
807#[serde(deny_unknown_fields)]
808pub struct LogStreamLinkConfig {
809 pub mtu_bytes: u16,
810 pub bitrate_bps: u64,
811 pub memory_budget_kib: u32,
812 pub max_latency_ms: u32,
813 pub burst_packets: u32,
814}
815
816#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
819#[serde(deny_unknown_fields)]
820pub struct LogStreamFecConfig {
821 pub continuous: LogStreamContinuousFecConfig,
822 pub objects: LogStreamObjectFecConfig,
823}
824
825#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
826#[serde(deny_unknown_fields)]
827pub struct LogStreamContinuousFecConfig {
828 pub field: LogStreamRlcField,
829 pub window_symbols: u16,
830 pub repair_every_source_symbols: u16,
831 pub repair_density: LogStreamRepairDensity,
832}
833
834#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
835pub enum LogStreamRlcField {
836 Gf2,
837 Gf256,
838}
839
840#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
841pub enum LogStreamRepairDensity {
842 Full,
843 Threshold(u8),
844}
845
846impl LogStreamRepairDensity {
847 pub const fn threshold(self) -> u8 {
848 match self {
849 Self::Full => 15,
850 Self::Threshold(value) => value,
851 }
852 }
853}
854
855#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
856#[serde(deny_unknown_fields)]
857pub struct LogStreamObjectFecConfig {
858 pub max_object_bytes: u64,
859 pub repair_symbols_per_block: u32,
860}
861
862#[derive(Serialize, Deserialize, Debug, Clone)]
864pub struct BridgeConfig {
865 pub id: String,
866 #[serde(rename = "type")]
867 pub type_: String,
868 #[serde(skip_serializing_if = "Option::is_none")]
869 pub config: Option<ComponentConfig>,
870 #[serde(skip_serializing_if = "Option::is_none")]
871 #[serde(serialize_with = "serialize_optional_ordered_map")]
872 pub resources: Option<HashMap<String, String>>,
873 #[serde(skip_serializing_if = "Option::is_none")]
874 pub missions: Option<Vec<String>>,
875 #[serde(skip_serializing_if = "Option::is_none")]
880 pub run_in_sim: Option<bool>,
881 pub channels: Vec<BridgeChannelConfigRepresentation>,
883}
884
885impl BridgeConfig {
886 #[allow(dead_code)]
888 pub fn is_run_in_sim(&self) -> bool {
889 self.run_in_sim.unwrap_or(true)
890 }
891
892 fn to_node(&self) -> Node {
893 let mut node = Node::new_with_flavor(&self.id, &self.type_, Flavor::Bridge);
894 node.config = self.config.clone();
895 node.resources = self.resources.clone();
896 node.missions = self.missions.clone();
897 node
898 }
899}
900
901fn insert_bridge_node(graph: &mut CuGraph, bridge: &BridgeConfig) -> Result<(), String> {
902 if graph.get_node_id_by_name(bridge.id.as_str()).is_some() {
903 return Err(format!(
904 "Bridge '{}' reuses an existing node id. Bridge ids must be unique.",
905 bridge.id
906 ));
907 }
908 graph
909 .add_node(bridge.to_node())
910 .map(|_| ())
911 .map_err(|e| e.to_string())
912}
913
914#[derive(Serialize, Deserialize, Debug, Clone)]
916struct SerializedCnx {
917 src: String,
918 dst: String,
919 msg: String,
920 missions: Option<Vec<String>>,
921}
922
923pub const NC_ENDPOINT: &str = "__nc__";
925
926#[derive(Debug, Clone)]
928pub struct Cnx {
929 pub src: String,
931 pub dst: String,
933 pub msg: String,
935 pub missions: Option<Vec<String>>,
937 pub src_channel: Option<String>,
939 pub dst_channel: Option<String>,
941 pub order: usize,
943}
944
945impl From<&Cnx> for SerializedCnx {
946 fn from(cnx: &Cnx) -> Self {
947 SerializedCnx {
948 src: format_endpoint(&cnx.src, cnx.src_channel.as_deref()),
949 dst: format_endpoint(&cnx.dst, cnx.dst_channel.as_deref()),
950 msg: cnx.msg.clone(),
951 missions: cnx.missions.clone(),
952 }
953 }
954}
955
956fn format_endpoint(node: &str, channel: Option<&str>) -> String {
957 match channel {
958 Some(ch) => format!("{node}/{ch}"),
959 None => node.to_string(),
960 }
961}
962
963fn parse_endpoint(
964 endpoint: &str,
965 role: EndpointRole,
966 bridges: &HashMap<&str, &BridgeConfig>,
967) -> Result<(String, Option<String>), String> {
968 if let Some((node, channel)) = endpoint.split_once('/') {
969 if let Some(bridge) = bridges.get(node) {
970 validate_bridge_channel(bridge, channel, role)?;
971 return Ok((node.to_string(), Some(channel.to_string())));
972 } else {
973 return Err(format!(
974 "Endpoint '{endpoint}' references an unknown bridge '{node}'"
975 ));
976 }
977 }
978
979 if let Some(bridge) = bridges.get(endpoint) {
980 return Err(format!(
981 "Bridge '{}' connections must reference a channel using '{}/<channel>'",
982 bridge.id, bridge.id
983 ));
984 }
985
986 Ok((endpoint.to_string(), None))
987}
988
989fn build_bridge_lookup(bridges: Option<&Vec<BridgeConfig>>) -> HashMap<&str, &BridgeConfig> {
990 let mut map = HashMap::new();
991 if let Some(bridges) = bridges {
992 for bridge in bridges {
993 map.insert(bridge.id.as_str(), bridge);
994 }
995 }
996 map
997}
998
999fn mission_applies(missions: &Option<Vec<String>>, mission_id: &str) -> bool {
1000 missions
1001 .as_ref()
1002 .map(|mission_list| mission_list.iter().any(|m| m == mission_id))
1003 .unwrap_or(true)
1004}
1005
1006fn merge_connection_missions(existing: &mut Option<Vec<String>>, incoming: &Option<Vec<String>>) {
1007 if incoming.is_none() {
1008 *existing = None;
1009 return;
1010 }
1011 if existing.is_none() {
1012 return;
1013 }
1014
1015 if let (Some(existing_missions), Some(incoming_missions)) =
1016 (existing.as_mut(), incoming.as_ref())
1017 {
1018 for mission in incoming_missions {
1019 if !existing_missions
1020 .iter()
1021 .any(|existing_mission| existing_mission == mission)
1022 {
1023 existing_missions.push(mission.clone());
1024 }
1025 }
1026 existing_missions.sort();
1027 existing_missions.dedup();
1028 }
1029}
1030
1031fn register_nc_output<E>(
1032 graph: &mut CuGraph,
1033 src_endpoint: &str,
1034 msg_type: &str,
1035 order: usize,
1036 bridge_lookup: &HashMap<&str, &BridgeConfig>,
1037) -> Result<(), E>
1038where
1039 E: From<String>,
1040{
1041 let (src_name, src_channel) =
1042 parse_endpoint(src_endpoint, EndpointRole::Source, bridge_lookup).map_err(E::from)?;
1043 if src_channel.is_some() {
1044 return Err(E::from(format!(
1045 "NC destination '{}' does not support bridge channels in source endpoint '{}'",
1046 NC_ENDPOINT, src_endpoint
1047 )));
1048 }
1049
1050 let src = graph
1051 .get_node_id_by_name(src_name.as_str())
1052 .ok_or_else(|| E::from(format!("Source node not found: {src_endpoint}")))?;
1053 let src_node = graph
1054 .get_node_mut(src)
1055 .ok_or_else(|| E::from(format!("Source node id {src} not found for NC output")))?;
1056 if src_node.get_flavor() != Flavor::Task {
1057 return Err(E::from(format!(
1058 "NC destination '{}' is only supported for task outputs (source '{}')",
1059 NC_ENDPOINT, src_endpoint
1060 )));
1061 }
1062 src_node.add_nc_output(msg_type, order);
1063 Ok(())
1064}
1065
1066#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1069pub enum CuDirection {
1070 Outgoing,
1071 Incoming,
1072}
1073
1074impl From<CuDirection> for petgraph::Direction {
1075 fn from(dir: CuDirection) -> Self {
1076 match dir {
1077 CuDirection::Outgoing => petgraph::Direction::Outgoing,
1078 CuDirection::Incoming => petgraph::Direction::Incoming,
1079 }
1080 }
1081}
1082
1083#[derive(Default, Debug, Clone)]
1084pub struct CuGraph(pub StableDiGraph<Node, Cnx, NodeId>);
1085
1086impl CuGraph {
1087 #[allow(dead_code)]
1088 pub fn get_all_nodes(&self) -> Vec<(NodeId, &Node)> {
1089 self.0
1090 .node_indices()
1091 .map(|index| (index.index() as u32, &self.0[index]))
1092 .collect()
1093 }
1094
1095 #[allow(dead_code)]
1096 pub fn get_neighbor_ids(&self, node_id: NodeId, dir: CuDirection) -> Vec<NodeId> {
1097 self.0
1098 .neighbors_directed(node_id.into(), dir.into())
1099 .map(|petgraph_index| petgraph_index.index() as NodeId)
1100 .collect()
1101 }
1102
1103 #[allow(dead_code)]
1104 pub fn node_ids(&self) -> Vec<NodeId> {
1105 self.0
1106 .node_indices()
1107 .map(|index| index.index() as NodeId)
1108 .collect()
1109 }
1110
1111 #[allow(dead_code)]
1112 pub fn edge_id_between(&self, source: NodeId, target: NodeId) -> Option<usize> {
1113 self.0
1114 .find_edge(source.into(), target.into())
1115 .map(|edge| edge.index())
1116 }
1117
1118 #[allow(dead_code)]
1119 pub fn edge(&self, edge_id: usize) -> Option<&Cnx> {
1120 self.0.edge_weight(EdgeIndex::new(edge_id))
1121 }
1122
1123 #[allow(dead_code)]
1124 pub fn edges(&self) -> impl Iterator<Item = &Cnx> {
1125 self.0
1126 .edge_indices()
1127 .filter_map(|edge| self.0.edge_weight(edge))
1128 }
1129
1130 #[allow(dead_code)]
1131 pub fn bfs_nodes(&self, start: NodeId) -> Vec<NodeId> {
1132 let mut visitor = Bfs::new(&self.0, start.into());
1133 let mut nodes = Vec::new();
1134 while let Some(node) = visitor.next(&self.0) {
1135 nodes.push(node.index() as NodeId);
1136 }
1137 nodes
1138 }
1139
1140 #[allow(dead_code)]
1141 pub fn incoming_neighbor_count(&self, node_id: NodeId) -> usize {
1142 self.0.neighbors_directed(node_id.into(), Incoming).count()
1143 }
1144
1145 #[allow(dead_code)]
1146 pub fn outgoing_neighbor_count(&self, node_id: NodeId) -> usize {
1147 self.0.neighbors_directed(node_id.into(), Outgoing).count()
1148 }
1149
1150 pub fn node_indices(&self) -> Vec<petgraph::stable_graph::NodeIndex> {
1151 self.0.node_indices().collect()
1152 }
1153
1154 pub fn add_node(&mut self, node: Node) -> CuResult<NodeId> {
1155 Ok(self.0.add_node(node).index() as NodeId)
1156 }
1157
1158 #[allow(dead_code)]
1159 pub fn connection_exists(&self, source: NodeId, target: NodeId) -> bool {
1160 self.0.find_edge(source.into(), target.into()).is_some()
1161 }
1162
1163 pub fn connect_ext(
1164 &mut self,
1165 source: NodeId,
1166 target: NodeId,
1167 msg_type: &str,
1168 missions: Option<Vec<String>>,
1169 src_channel: Option<String>,
1170 dst_channel: Option<String>,
1171 ) -> CuResult<()> {
1172 self.connect_ext_with_order(
1173 source,
1174 target,
1175 msg_type,
1176 missions,
1177 src_channel,
1178 dst_channel,
1179 usize::MAX,
1180 )
1181 }
1182
1183 #[allow(clippy::too_many_arguments)]
1184 pub fn connect_ext_with_order(
1185 &mut self,
1186 source: NodeId,
1187 target: NodeId,
1188 msg_type: &str,
1189 missions: Option<Vec<String>>,
1190 src_channel: Option<String>,
1191 dst_channel: Option<String>,
1192 order: usize,
1193 ) -> CuResult<()> {
1194 let (src_id, dst_id) = (
1195 self.0
1196 .node_weight(source.into())
1197 .ok_or("Source node not found")?
1198 .id
1199 .clone(),
1200 self.0
1201 .node_weight(target.into())
1202 .ok_or("Target node not found")?
1203 .id
1204 .clone(),
1205 );
1206
1207 let _ = self.0.add_edge(
1208 petgraph::stable_graph::NodeIndex::from(source),
1209 petgraph::stable_graph::NodeIndex::from(target),
1210 Cnx {
1211 src: src_id,
1212 dst: dst_id,
1213 msg: msg_type.to_string(),
1214 missions,
1215 src_channel,
1216 dst_channel,
1217 order,
1218 },
1219 );
1220 Ok(())
1221 }
1222 #[allow(dead_code)]
1226 pub fn get_node(&self, node_id: NodeId) -> Option<&Node> {
1227 self.0.node_weight(node_id.into())
1228 }
1229
1230 #[allow(dead_code)]
1231 pub fn get_node_weight(&self, index: NodeId) -> Option<&Node> {
1232 self.0.node_weight(index.into())
1233 }
1234
1235 #[allow(dead_code)]
1236 pub fn get_node_mut(&mut self, node_id: NodeId) -> Option<&mut Node> {
1237 self.0.node_weight_mut(node_id.into())
1238 }
1239
1240 pub fn get_node_id_by_name(&self, name: &str) -> Option<NodeId> {
1241 self.0
1242 .node_indices()
1243 .into_iter()
1244 .find(|idx| self.0[*idx].get_id() == name)
1245 .map(|i| i.index() as NodeId)
1246 }
1247
1248 #[allow(dead_code)]
1249 pub fn get_edge_weight(&self, index: usize) -> Option<Cnx> {
1250 self.0.edge_weight(EdgeIndex::new(index)).cloned()
1251 }
1252
1253 #[allow(dead_code)]
1254 pub fn get_node_output_msg_type(&self, node_id: &str) -> Option<String> {
1255 self.get_node_output_msg_types(node_id)
1256 .and_then(|mut msgs| msgs.drain(..1).next())
1257 }
1258
1259 #[allow(dead_code)]
1260 pub fn get_node_output_msg_types(&self, node_id: &str) -> Option<Vec<String>> {
1261 let node_id = self.get_node_id_by_name(node_id)?;
1262 let msgs = self.get_node_output_msg_types_by_id(node_id).ok()?;
1263 (!msgs.is_empty()).then_some(msgs)
1264 }
1265
1266 #[allow(dead_code)]
1267 pub fn get_node_output_msg_types_by_id(&self, node_id: NodeId) -> CuResult<Vec<String>> {
1268 let mut edge_ids = self.get_src_edges(node_id)?;
1269 edge_ids.sort();
1270
1271 let node = self
1272 .get_node(node_id)
1273 .ok_or_else(|| CuError::from(format!("Node id {node_id} not found")))?;
1274
1275 let mut msg_order: Vec<(usize, String)> = Vec::new();
1276 let mut record_msg = |msg: String, order: usize| {
1277 if let Some((existing_order, _)) = msg_order
1278 .iter_mut()
1279 .find(|(_, existing_msg)| *existing_msg == msg)
1280 {
1281 if order < *existing_order {
1282 *existing_order = order;
1283 }
1284 return;
1285 }
1286 msg_order.push((order, msg));
1287 };
1288
1289 for edge_id in edge_ids {
1290 let Some(edge) = self.edge(edge_id) else {
1291 continue;
1292 };
1293 let order = if edge.order == usize::MAX {
1294 edge_id
1295 } else {
1296 edge.order
1297 };
1298 record_msg(edge.msg.clone(), order);
1299 }
1300
1301 for (msg, order) in node.nc_outputs_with_order() {
1302 record_msg(msg.clone(), order);
1303 }
1304
1305 msg_order.sort_by(|(order_a, msg_a), (order_b, msg_b)| {
1306 order_a.cmp(order_b).then_with(|| msg_a.cmp(msg_b))
1307 });
1308 Ok(msg_order.into_iter().map(|(_, msg)| msg).collect())
1309 }
1310
1311 #[allow(dead_code)]
1319 pub fn get_node_output_ports_by_id(
1320 &self,
1321 node_id: NodeId,
1322 ) -> CuResult<Vec<(String, Option<String>)>> {
1323 let mut edge_ids = self.get_src_edges(node_id)?;
1324 edge_ids.sort();
1325
1326 let node = self
1327 .get_node(node_id)
1328 .ok_or_else(|| CuError::from(format!("Node id {node_id} not found")))?;
1329
1330 let mut port_order: Vec<(usize, String, Option<String>)> = Vec::new();
1331 let mut record_port = |msg: String, channel: Option<String>, order: usize| {
1332 if let Some((existing_order, _, _)) = port_order
1333 .iter_mut()
1334 .find(|(_, m, c)| *m == msg && *c == channel)
1335 {
1336 if order < *existing_order {
1337 *existing_order = order;
1338 }
1339 return;
1340 }
1341 port_order.push((order, msg, channel));
1342 };
1343
1344 for edge_id in edge_ids {
1345 let Some(edge) = self.edge(edge_id) else {
1346 continue;
1347 };
1348 let order = if edge.order == usize::MAX {
1349 edge_id
1350 } else {
1351 edge.order
1352 };
1353 record_port(edge.msg.clone(), edge.src_channel.clone(), order);
1354 }
1355
1356 for (msg, order) in node.nc_outputs_with_order() {
1357 record_port(msg.clone(), None, order);
1358 }
1359
1360 port_order.sort_by(|(order_a, msg_a, ch_a), (order_b, msg_b, ch_b)| {
1361 order_a
1362 .cmp(order_b)
1363 .then_with(|| msg_a.cmp(msg_b))
1364 .then_with(|| ch_a.cmp(ch_b))
1365 });
1366 Ok(port_order
1367 .into_iter()
1368 .map(|(_, msg, ch)| (msg, ch))
1369 .collect())
1370 }
1371
1372 #[allow(dead_code)]
1373 pub fn get_node_input_msg_type(&self, node_id: &str) -> Option<String> {
1374 self.get_node_input_msg_types(node_id)
1375 .and_then(|mut v| v.pop())
1376 }
1377
1378 pub fn get_node_input_msg_types(&self, node_id: &str) -> Option<Vec<String>> {
1379 self.0.node_indices().find_map(|node_index| {
1380 if let Some(node) = self.0.node_weight(node_index) {
1381 if node.id != node_id {
1382 return None;
1383 }
1384 let edges: Vec<_> = self
1385 .0
1386 .edges_directed(node_index, Incoming)
1387 .map(|edge| edge.id().index())
1388 .collect();
1389 if edges.is_empty() {
1390 return None;
1391 }
1392 let mut edges = edges;
1393 edges.sort();
1394 let msgs = edges
1395 .into_iter()
1396 .map(|edge_id| {
1397 let cnx = self
1398 .0
1399 .edge_weight(EdgeIndex::new(edge_id))
1400 .expect("Found an cnx id but could not retrieve it back");
1401 cnx.msg.clone()
1402 })
1403 .collect();
1404 return Some(msgs);
1405 }
1406 None
1407 })
1408 }
1409
1410 #[allow(dead_code)]
1411 pub fn get_connection_msg_type(&self, source: NodeId, target: NodeId) -> Option<&str> {
1412 self.0
1413 .find_edge(source.into(), target.into())
1414 .map(|edge_index| self.0[edge_index].msg.as_str())
1415 }
1416
1417 fn get_edges_by_direction(
1419 &self,
1420 node_id: NodeId,
1421 direction: petgraph::Direction,
1422 ) -> CuResult<Vec<usize>> {
1423 Ok(self
1424 .0
1425 .edges_directed(node_id.into(), direction)
1426 .map(|edge| edge.id().index())
1427 .collect())
1428 }
1429
1430 pub fn get_src_edges(&self, node_id: NodeId) -> CuResult<Vec<usize>> {
1431 self.get_edges_by_direction(node_id, Outgoing)
1432 }
1433
1434 pub fn get_dst_edges(&self, node_id: NodeId) -> CuResult<Vec<usize>> {
1436 self.get_edges_by_direction(node_id, Incoming)
1437 }
1438
1439 #[allow(dead_code)]
1440 pub fn node_count(&self) -> usize {
1441 self.0.node_count()
1442 }
1443
1444 #[allow(dead_code)]
1445 pub fn edge_count(&self) -> usize {
1446 self.0.edge_count()
1447 }
1448
1449 #[allow(dead_code)]
1452 pub fn connect(&mut self, source: NodeId, target: NodeId, msg_type: &str) -> CuResult<()> {
1453 self.connect_ext(source, target, msg_type, None, None, None)
1454 }
1455}
1456
1457fn validate_task_kind(
1458 node_id: &str,
1459 kind: TaskKind,
1460 has_inputs: bool,
1461 has_outputs: bool,
1462) -> CuResult<()> {
1463 match kind {
1464 TaskKind::Source if has_inputs => Err(CuError::from(format!(
1465 "Task '{node_id}' is declared as kind 'source' but has incoming connections. Sources map to CuSrcTask and cannot consume inputs. Use kind: task instead."
1466 ))),
1467 TaskKind::Regular | TaskKind::Stateless if !has_inputs => Err(CuError::from(format!(
1468 "Task '{node_id}' is declared as kind '{}' but has no incoming connections. Transform tasks need at least one input connection. Use kind: source if it is input-free.",
1469 kind.as_str()
1470 ))),
1471 TaskKind::Sink if has_outputs => Err(CuError::from(format!(
1472 "Task '{node_id}' is declared as kind 'sink' but has outgoing or NC outputs. Sinks map to CuSinkTask and cannot produce outputs. Use kind: task instead."
1473 ))),
1474 TaskKind::Sink if !has_inputs => Err(CuError::from(format!(
1475 "Task '{node_id}' is declared as kind 'sink' but has no incoming connections. Sinks need at least one input connection so Copper can determine their input message type."
1476 ))),
1477 _ => Ok(()),
1478 }
1479}
1480
1481#[allow(dead_code)]
1482pub fn infer_task_kind_for_id(graph: &CuGraph, node_id: NodeId) -> Option<TaskKind> {
1483 let node = graph.get_node(node_id)?;
1484 if node.get_flavor() != Flavor::Task {
1485 return None;
1486 }
1487
1488 let has_inputs = !graph.get_dst_edges(node_id).ok()?.is_empty();
1489 let has_outputs = !graph
1490 .get_node_output_msg_types_by_id(node_id)
1491 .ok()?
1492 .is_empty();
1493
1494 match (has_inputs, has_outputs) {
1495 (false, true) => Some(TaskKind::Source),
1496 (true, true) => Some(TaskKind::Regular),
1497 (true, false) => Some(TaskKind::Sink),
1498 (false, false) => None,
1499 }
1500}
1501
1502#[allow(dead_code)]
1503pub fn resolve_task_kind_for_id(graph: &CuGraph, node_id: NodeId) -> CuResult<TaskKind> {
1504 let node = graph
1505 .get_node(node_id)
1506 .ok_or_else(|| CuError::from(format!("Task node id {node_id} not found")))?;
1507 if node.get_flavor() != Flavor::Task {
1508 return Err(CuError::from(format!(
1509 "Node '{}' is not a task and does not have a task kind.",
1510 node.id
1511 )));
1512 }
1513
1514 let has_inputs = !graph.get_dst_edges(node_id)?.is_empty();
1515 let has_outputs = !graph.get_node_output_msg_types_by_id(node_id)?.is_empty();
1516
1517 if let Some(kind) = node.get_declared_task_kind() {
1518 validate_task_kind(node.id.as_str(), kind, has_inputs, has_outputs)?;
1519 return Ok(kind);
1520 }
1521
1522 let inferred = match (has_inputs, has_outputs) {
1523 (false, true) => TaskKind::Source,
1524 (true, true) => TaskKind::Regular,
1525 (true, false) => TaskKind::Sink,
1526 (false, false) => {
1527 return Err(CuError::from(format!(
1528 "Task '{}' has no declared inputs or outputs, so Copper cannot infer whether it is a source, task, or sink. Add `kind: source|task|sink`; source/task nodes also need an output declaration via a connection or `dst: \"{NC_ENDPOINT}\"`.",
1529 node.id
1530 )));
1531 }
1532 };
1533
1534 validate_task_kind(node.id.as_str(), inferred, has_inputs, has_outputs)?;
1535 Ok(inferred)
1536}
1537
1538impl core::ops::Index<NodeIndex> for CuGraph {
1539 type Output = Node;
1540
1541 fn index(&self, index: NodeIndex) -> &Self::Output {
1542 &self.0[index]
1543 }
1544}
1545
1546#[derive(Debug, Clone)]
1547pub enum ConfigGraphs {
1548 Simple(CuGraph),
1549 Missions(HashMap<String, CuGraph>),
1550}
1551
1552impl ConfigGraphs {
1553 #[allow(dead_code)]
1556 pub fn get_all_missions_graphs(&self) -> HashMap<String, CuGraph> {
1557 match self {
1558 Simple(graph) => HashMap::from([(DEFAULT_MISSION_ID.to_string(), graph.clone())]),
1559 Missions(graphs) => graphs.clone(),
1560 }
1561 }
1562
1563 #[allow(dead_code)]
1564 pub fn get_default_mission_graph(&self) -> CuResult<&CuGraph> {
1565 match self {
1566 Simple(graph) => Ok(graph),
1567 Missions(graphs) => {
1568 if graphs.len() == 1 {
1569 Ok(graphs.values().next().unwrap())
1570 } else {
1571 Err("Cannot get default mission graph from mission config".into())
1572 }
1573 }
1574 }
1575 }
1576
1577 #[allow(dead_code)]
1578 pub fn get_graph(&self, mission_id: Option<&str>) -> CuResult<&CuGraph> {
1579 match self {
1580 Simple(graph) => match mission_id {
1581 None | Some(DEFAULT_MISSION_ID) => Ok(graph),
1582 Some(_) => Err("Cannot get mission graph from simple config".into()),
1583 },
1584 Missions(graphs) => {
1585 let id = mission_id
1586 .ok_or_else(|| "Mission ID required for mission configs".to_string())?;
1587 graphs
1588 .get(id)
1589 .ok_or_else(|| format!("Mission {id} not found").into())
1590 }
1591 }
1592 }
1593
1594 #[allow(dead_code)]
1595 pub fn get_graph_mut(&mut self, mission_id: Option<&str>) -> CuResult<&mut CuGraph> {
1596 match self {
1597 Simple(graph) => match mission_id {
1598 None => Ok(graph),
1599 Some(_) => Err("Cannot get mission graph from simple config".into()),
1600 },
1601 Missions(graphs) => {
1602 let id = mission_id
1603 .ok_or_else(|| "Mission ID required for mission configs".to_string())?;
1604 graphs
1605 .get_mut(id)
1606 .ok_or_else(|| format!("Mission {id} not found").into())
1607 }
1608 }
1609 }
1610
1611 pub fn add_mission(&mut self, mission_id: &str) -> CuResult<&mut CuGraph> {
1612 match self {
1613 Simple(_) => Err("Cannot add mission to simple config".into()),
1614 Missions(graphs) => match graphs.entry(mission_id.to_string()) {
1615 hashbrown::hash_map::Entry::Occupied(_) => {
1616 Err(format!("Mission {mission_id} already exists").into())
1617 }
1618 hashbrown::hash_map::Entry::Vacant(entry) => Ok(entry.insert(CuGraph::default())),
1619 },
1620 }
1621 }
1622}
1623
1624#[derive(Debug, Clone)]
1630pub struct CuConfig {
1631 #[doc(hidden)]
1633 pub constants: Vec<ConstantConfig>,
1634 pub monitors: Vec<MonitorConfig>,
1636 pub logging: Option<LoggingConfig>,
1638 pub runtime: Option<RuntimeConfig>,
1640 pub resources: Vec<ResourceBundleConfig>,
1642 pub log_streaming: Option<LogStreamingConfig>,
1644 pub bridges: Vec<BridgeConfig>,
1646 pub graphs: ConfigGraphs,
1648}
1649
1650fn validate_reconstruction_inputs(graph: &CuGraph) -> CuResult<()> {
1653 for index in graph.0.node_indices() {
1654 let node = &graph.0[index];
1655 if node.streaming().replay != StreamReplay::Reconstruct {
1656 continue;
1657 }
1658 for edge in graph.0.edges_directed(index, Incoming) {
1659 let source = &graph.0[edge.source()];
1660 if !source.is_logging_enabled() {
1661 return Err(CuError::from(format!(
1662 "Task '{}' uses streaming.replay: reconstruct but input '{}' from '{}' has logging.enabled: false. Enable logging on '{}' or use streaming.replay: capture on '{}'.",
1663 node.id,
1664 edge.weight().msg,
1665 source.id,
1666 source.id,
1667 node.id
1668 )));
1669 }
1670 }
1671 }
1672 Ok(())
1673}
1674
1675impl CuConfig {
1676 pub fn validate_log_streaming_config(&self) -> CuResult<()> {
1678 match &self.graphs {
1679 Simple(graph) => validate_reconstruction_inputs(graph)?,
1680 Missions(graphs) => {
1681 for (mission, graph) in graphs {
1682 validate_reconstruction_inputs(graph)
1683 .map_err(|error| CuError::from(format!("Mission '{mission}': {error}")))?;
1684 }
1685 }
1686 }
1687
1688 let Some(streaming) = &self.log_streaming else {
1689 return Ok(());
1690 };
1691 if streaming.destinations.is_empty() {
1692 return Err(CuError::from(
1693 "log_streaming.destinations must contain at least one destination",
1694 ));
1695 }
1696
1697 let keyframe_interval = self
1698 .logging
1699 .as_ref()
1700 .and_then(|logging| logging.keyframe_interval)
1701 .unwrap_or(DEFAULT_KEYFRAME_INTERVAL);
1702 for (index, destination) in streaming.destinations.iter().enumerate() {
1703 if destination.id.trim().is_empty() {
1704 return Err(CuError::from(format!(
1705 "log_streaming destination at index {index} has an empty id"
1706 )));
1707 }
1708 if streaming.destinations[..index]
1709 .iter()
1710 .any(|other| other.id == destination.id)
1711 {
1712 return Err(CuError::from(format!(
1713 "Duplicate log_streaming destination id '{}'",
1714 destination.id
1715 )));
1716 }
1717 if destination.transport.type_.trim().is_empty() {
1718 return Err(CuError::from(format!(
1719 "log_streaming destination '{}' has an empty transport type",
1720 destination.id
1721 )));
1722 }
1723 let Some((bundle_id, resource_name)) = destination.transport.resource.split_once('.')
1724 else {
1725 return Err(CuError::from(format!(
1726 "log_streaming destination '{}' resource '{}' must use 'bundle.resource' syntax",
1727 destination.id, destination.transport.resource
1728 )));
1729 };
1730 if bundle_id.is_empty()
1731 || resource_name.is_empty()
1732 || resource_name.contains('.')
1733 || !self.resources.iter().any(|bundle| bundle.id == bundle_id)
1734 {
1735 return Err(CuError::from(format!(
1736 "log_streaming destination '{}' references invalid resource '{}'",
1737 destination.id, destination.transport.resource
1738 )));
1739 }
1740 if streaming.destinations[..index]
1741 .iter()
1742 .any(|other| other.transport.resource == destination.transport.resource)
1743 {
1744 return Err(CuError::from(format!(
1745 "log_streaming resource '{}' is bound by more than one destination",
1746 destination.transport.resource
1747 )));
1748 }
1749 if let Some(feedback) = &destination.feedback {
1750 let baseline = destination.fec.continuous.repair_every_source_symbols;
1751 if feedback.report_interval_ms == 0
1752 || feedback.timeout_ms <= feedback.report_interval_ms
1753 || feedback.adaptation.is_some_and(|bounds| {
1754 bounds.min_repair_every_source_symbols == 0
1755 || bounds.min_repair_every_source_symbols > baseline
1756 || baseline > bounds.max_repair_every_source_symbols
1757 })
1758 {
1759 return Err(CuError::from(
1760 "Invalid log_streaming feedback cadence or FEC bounds",
1761 ));
1762 }
1763 let resource = &feedback.transport.resource;
1764 let valid_resource = resource.split_once('.').is_some_and(|(bundle, slot)| {
1765 !slot.is_empty()
1766 && !slot.contains('.')
1767 && self.resources.iter().any(|r| r.id == bundle)
1768 });
1769 if feedback.transport.type_.trim().is_empty() || !valid_resource {
1770 return Err(CuError::from("Invalid log_streaming feedback resource"));
1771 }
1772 }
1773 let resources = core::iter::once(&destination.transport.resource)
1775 .chain(destination.feedback.iter().map(|f| &f.transport.resource));
1776 for resource in resources {
1777 let uses = streaming
1778 .destinations
1779 .iter()
1780 .map(|d| {
1781 usize::from(&d.transport.resource == resource)
1782 + usize::from(
1783 d.feedback
1784 .as_ref()
1785 .is_some_and(|f| &f.transport.resource == resource),
1786 )
1787 })
1788 .sum::<usize>();
1789 if uses > 1 {
1790 return Err(CuError::from(format!(
1791 "Log-stream resource '{resource}' has multiple owners"
1792 )));
1793 }
1794 }
1795 if destination.link.mtu_bytes <= 72 {
1796 return Err(CuError::from(format!(
1797 "log_streaming destination '{}' mtu_bytes must exceed the 72-byte packet header",
1798 destination.id
1799 )));
1800 }
1801 if destination.link.bitrate_bps == 0
1802 || destination.link.memory_budget_kib == 0
1803 || destination.link.max_latency_ms == 0
1804 || destination.link.burst_packets == 0
1805 {
1806 return Err(CuError::from(format!(
1807 "log_streaming destination '{}' link values must be nonzero",
1808 destination.id
1809 )));
1810 }
1811 if destination.fec.continuous.window_symbols == 0
1812 || destination.fec.continuous.repair_every_source_symbols == 0
1813 {
1814 return Err(CuError::from(format!(
1815 "log_streaming destination '{}' continuous FEC values must be nonzero",
1816 destination.id
1817 )));
1818 }
1819 if destination.fec.continuous.repair_density.threshold() > 15 {
1820 return Err(CuError::from(format!(
1821 "log_streaming destination '{}' repair density threshold must be in 0..=15",
1822 destination.id
1823 )));
1824 }
1825 if destination.fec.objects.max_object_bytes == 0
1826 || destination.fec.objects.repair_symbols_per_block == 0
1827 || destination.max_record_bytes == 0
1828 {
1829 return Err(CuError::from(format!(
1830 "log_streaming destination '{}' object and record bounds must be nonzero",
1831 destination.id
1832 )));
1833 }
1834 if destination.recovery_interval == 0
1835 || !destination
1836 .recovery_interval
1837 .is_multiple_of(keyframe_interval)
1838 {
1839 return Err(CuError::from(format!(
1840 "log_streaming destination '{}' recovery_interval must be a nonzero multiple of logging.keyframe_interval ({keyframe_interval})",
1841 destination.id
1842 )));
1843 }
1844 }
1845 Ok(())
1846 }
1847
1848 #[cfg(feature = "std")]
1854 fn ensure_default_background_pool(&mut self) {
1855 if !self.has_background_tasks() {
1856 return;
1857 }
1858
1859 const DEFAULT_BACKGROUND_THREADS: usize = 2;
1860
1861 let runtime = self.runtime.get_or_insert_with(RuntimeConfig::default);
1862 if !runtime
1863 .thread_pools
1864 .iter()
1865 .any(|pool| pool.id == DEFAULT_BACKGROUND_POOL)
1866 {
1867 runtime.thread_pools.push(ThreadPoolConfig {
1868 id: DEFAULT_BACKGROUND_POOL.to_string(),
1869 threads: DEFAULT_BACKGROUND_THREADS,
1870 affinity: None,
1871 policy: SchedulingPolicy::Fair,
1872 on_error: OnError::Warn,
1873 });
1874 }
1875 }
1876
1877 pub fn planner_config(&self) -> Option<&PlannerConfig> {
1879 self.runtime.as_ref()?.planner.as_ref()
1880 }
1881
1882 pub fn planner_kind(&self) -> PlannerKind {
1884 self.planner_config()
1885 .map_or(PlannerKind::Serial, PlannerConfig::kind)
1886 }
1887
1888 #[cfg(feature = "std")]
1889 fn has_background_tasks(&self) -> bool {
1890 match &self.graphs {
1891 ConfigGraphs::Simple(graph) => graph
1892 .get_all_nodes()
1893 .iter()
1894 .any(|(_, node)| node.is_background()),
1895 ConfigGraphs::Missions(graphs) => graphs.values().any(|graph| {
1896 graph
1897 .get_all_nodes()
1898 .iter()
1899 .any(|(_, node)| node.is_background())
1900 }),
1901 }
1902 }
1903}
1904
1905#[derive(Serialize, Deserialize, Default, Debug, Clone)]
1906pub struct MonitorConfig {
1907 #[serde(rename = "type")]
1908 type_: String,
1909 #[serde(skip_serializing_if = "Option::is_none")]
1910 config: Option<ComponentConfig>,
1911}
1912
1913impl MonitorConfig {
1914 #[allow(dead_code)]
1915 pub fn get_type(&self) -> &str {
1916 &self.type_
1917 }
1918
1919 #[allow(dead_code)]
1920 pub fn get_config(&self) -> Option<&ComponentConfig> {
1921 self.config.as_ref()
1922 }
1923}
1924
1925fn default_as_true() -> bool {
1926 true
1927}
1928
1929pub const DEFAULT_KEYFRAME_INTERVAL: u32 = 100;
1930
1931fn default_keyframe_interval() -> Option<u32> {
1932 Some(DEFAULT_KEYFRAME_INTERVAL)
1933}
1934
1935#[derive(Serialize, Deserialize, Debug, Clone)]
1936pub struct LoggingConfig {
1937 #[serde(default = "default_as_true", skip_serializing_if = "Clone::clone")]
1939 pub enable_task_logging: bool,
1940
1941 #[serde(default = "default_as_true", skip_serializing_if = "Clone::clone")]
1947 pub enable_keyframe_logging: bool,
1948
1949 #[serde(skip_serializing_if = "Option::is_none")]
1954 pub copperlist_count: Option<usize>,
1955
1956 #[serde(skip_serializing_if = "Option::is_none")]
1958 pub slab_size_mib: Option<u64>,
1959
1960 #[serde(skip_serializing_if = "Option::is_none")]
1962 pub section_size_mib: Option<u64>,
1963
1964 #[serde(
1966 default = "default_keyframe_interval",
1967 skip_serializing_if = "Option::is_none"
1968 )]
1969 pub keyframe_interval: Option<u32>,
1970
1971 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1973 pub codecs: Vec<LoggingCodecSpec>,
1974}
1975
1976impl Default for LoggingConfig {
1977 fn default() -> Self {
1978 Self {
1979 enable_task_logging: true,
1980 enable_keyframe_logging: true,
1981 copperlist_count: None,
1982 slab_size_mib: None,
1983 section_size_mib: None,
1984 keyframe_interval: default_keyframe_interval(),
1985 codecs: Vec::new(),
1986 }
1987 }
1988}
1989
1990#[derive(Serialize, Deserialize, Debug, Clone)]
1991pub struct LoggingCodecSpec {
1992 pub id: String,
1993 #[serde(rename = "type")]
1994 pub type_: String,
1995 #[serde(skip_serializing_if = "Option::is_none")]
1996 pub config: Option<ComponentConfig>,
1997}
1998
1999#[derive(Serialize, Deserialize, Default, Debug, Clone)]
2000#[serde(try_from = "RuntimeConfigInput")]
2001pub struct RuntimeConfig {
2002 #[serde(default, skip_serializing_if = "Option::is_none")]
2004 #[cfg(feature = "clock-sync")]
2005 pub clock: Option<ClockSyncConfig>,
2006 #[serde(skip_serializing_if = "Option::is_none")]
2012 pub rate_target_hz: Option<u64>,
2013
2014 #[serde(default, skip_serializing_if = "Vec::is_empty")]
2021 pub thread_pools: Vec<ThreadPoolConfig>,
2022
2023 #[serde(default, skip_serializing_if = "Option::is_none")]
2030 pub planner: Option<PlannerConfig>,
2031}
2032
2033#[derive(Deserialize)]
2037struct RuntimeConfigInput {
2038 clock: Option<ClockSyncConfig>,
2039 rate_target_hz: Option<u64>,
2040 #[serde(default)]
2041 thread_pools: Vec<ThreadPoolConfig>,
2042 planner: Option<PlannerConfig>,
2043}
2044impl TryFrom<RuntimeConfigInput> for RuntimeConfig {
2045 type Error = &'static str;
2046 fn try_from(input: RuntimeConfigInput) -> Result<Self, Self::Error> {
2047 #[cfg(not(feature = "clock-sync"))]
2048 if input.clock.is_some() {
2049 return Err("runtime.clock requires the clock-sync feature");
2050 }
2051 Ok(Self {
2052 #[cfg(feature = "clock-sync")]
2053 clock: input.clock,
2054 rate_target_hz: input.rate_target_hz,
2055 thread_pools: input.thread_pools,
2056 planner: input.planner,
2057 })
2058 }
2059}
2060
2061#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2063#[serde(deny_unknown_fields)]
2064pub struct ClockSyncConfig {
2065 pub parent: String,
2067 pub max_error_ns: u64,
2069 #[serde(default = "clock_acquisition_timeout")]
2071 pub acquisition_timeout_ns: u64,
2072 #[serde(default = "clock_sample_interval")]
2074 pub sample_interval_ns: u64,
2075 #[serde(default = "clock_max_age")]
2077 pub max_age_ns: u64,
2078 #[serde(default = "clock_drift_bound")]
2080 pub drift_bound_ppb: u32,
2081 #[serde(default = "clock_slew_bound")]
2083 pub max_slew_ppb: u32,
2084}
2085fn clock_acquisition_timeout() -> u64 {
2086 10_000_000_000
2087}
2088fn clock_sample_interval() -> u64 {
2089 1_000_000_000
2090}
2091fn clock_max_age() -> u64 {
2092 5_000_000_000
2093}
2094fn clock_drift_bound() -> u32 {
2095 100_000
2096}
2097fn clock_slew_bound() -> u32 {
2098 500_000
2099}
2100
2101impl ClockSyncConfig {
2102 pub fn validate(&self) -> CuResult<()> {
2104 if self
2105 .parent
2106 .split_once('.')
2107 .is_none_or(|(bundle, slot)| bundle.is_empty() || slot.is_empty() || slot.contains('.'))
2108 {
2109 return Err(CuError::from(
2110 "runtime.clock.parent must be a bundle.resource path",
2111 ));
2112 }
2113 if self.max_error_ns == 0
2114 || self.max_age_ns == 0
2115 || self.acquisition_timeout_ns == 0
2116 || self.sample_interval_ns == 0
2117 || self.sample_interval_ns >= self.acquisition_timeout_ns
2118 || u64::from(self.drift_bound_ppb) + u64::from(self.max_slew_ppb) >= 1_000_000_000
2119 {
2120 return Err(CuError::from(
2121 "runtime.clock has invalid acquisition, error, age or rate bounds",
2122 ));
2123 }
2124 Ok(())
2125 }
2126
2127 #[cfg(feature = "clock-sync")]
2129 pub fn maintenance(&self) -> crate::clock_sync::MaintenanceConfig {
2130 crate::clock_sync::MaintenanceConfig {
2131 acquisition_timeout: core::time::Duration::from_nanos(self.acquisition_timeout_ns),
2132 sample_interval: core::time::Duration::from_nanos(self.sample_interval_ns),
2133 sync: cu29_clock::sync::SyncConfig {
2134 max_error: cu29_clock::CuDuration(self.max_error_ns),
2135 max_age: cu29_clock::CuDuration(self.max_age_ns),
2136 drift_bound_ppb: self.drift_bound_ppb,
2137 max_slew_ppb: self.max_slew_ppb,
2138 },
2139 }
2140 }
2141}
2142
2143#[derive(Serialize, Deserialize, Debug, Clone)]
2159pub struct PlannerConfig {
2160 pub(crate) kind: PlannerKind,
2161 #[serde(skip_serializing_if = "Option::is_none")]
2162 pub(crate) config: Option<ComponentConfig>,
2163}
2164
2165impl PlannerConfig {
2166 #[allow(dead_code)]
2167 pub fn kind(&self) -> PlannerKind {
2168 self.kind
2169 }
2170
2171 #[allow(dead_code)]
2172 pub fn get_config(&self) -> Option<&ComponentConfig> {
2173 self.config.as_ref()
2174 }
2175
2176 pub fn max_in_flight(&self, copperlist_count: usize) -> CuResult<usize> {
2178 if self.kind != PlannerKind::Pipeline {
2179 return Ok(1);
2180 }
2181 let configured = match self.config.as_ref() {
2182 Some(config) => config
2183 .get_value::<usize>("max_in_flight")?
2184 .unwrap_or(copperlist_count),
2185 None => copperlist_count,
2186 };
2187 if configured == 0 {
2188 return Err(CuError::from(
2189 "Pipeline planner max_in_flight cannot be zero.",
2190 ));
2191 }
2192 if configured > copperlist_count {
2193 return Err(CuError::from(format!(
2194 "Pipeline planner max_in_flight ({configured}) exceeds logging.copperlist_count ({copperlist_count})."
2195 )));
2196 }
2197 Ok(configured)
2198 }
2199}
2200
2201#[derive(Serialize, Deserialize, Default, Debug, Clone, Copy, PartialEq, Eq)]
2203pub enum PlannerKind {
2204 #[default]
2206 Serial,
2207 TaskOrder,
2209 ExplicitSchedule,
2211 Pipeline,
2213}
2214
2215pub const MIN_RT_PRIORITY: u8 = 1;
2217pub const MAX_RT_PRIORITY: u8 = 99;
2219pub const MIN_NICE: i8 = -20;
2221pub const MAX_NICE: i8 = 19;
2223
2224#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
2230pub enum SchedulingPolicy {
2231 #[default]
2236 Fair,
2237 Nice(i8),
2244 Fifo { priority: u8 },
2252 RoundRobin { priority: u8 },
2259}
2260
2261#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
2264pub enum OnError {
2265 #[default]
2268 Warn,
2269 Strict,
2272}
2273
2274#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2276pub struct ThreadPoolConfig {
2277 pub id: String,
2280 pub threads: usize,
2282 #[serde(default, skip_serializing_if = "Option::is_none")]
2286 pub affinity: Option<Vec<usize>>,
2287 #[serde(default)]
2289 pub policy: SchedulingPolicy,
2290 #[serde(default)]
2292 pub on_error: OnError,
2293}
2294
2295fn validate_thread_pools<E>(runtime: &Option<RuntimeConfig>) -> Result<(), E>
2301where
2302 E: From<String>,
2303{
2304 let Some(runtime) = runtime else {
2305 return Ok(());
2306 };
2307
2308 let mut seen: Vec<&str> = Vec::new();
2309 for pool in &runtime.thread_pools {
2310 if pool.id.is_empty() {
2311 return Err(E::from("Thread pool id cannot be empty".to_string()));
2312 }
2313 if seen.contains(&pool.id.as_str()) {
2314 return Err(E::from(format!("Duplicate thread pool id '{}'", pool.id)));
2315 }
2316 seen.push(pool.id.as_str());
2317
2318 if pool.threads == 0 {
2319 return Err(E::from(format!(
2320 "Thread pool '{}' must have at least 1 thread",
2321 pool.id
2322 )));
2323 }
2324
2325 match pool.policy {
2326 SchedulingPolicy::Fifo { priority } | SchedulingPolicy::RoundRobin { priority } => {
2327 if !(MIN_RT_PRIORITY..=MAX_RT_PRIORITY).contains(&priority) {
2328 return Err(E::from(format!(
2329 "Thread pool '{}' real-time priority {priority} is out of range ({MIN_RT_PRIORITY}..={MAX_RT_PRIORITY})",
2330 pool.id
2331 )));
2332 }
2333 }
2334 SchedulingPolicy::Nice(nice) => {
2335 if !(MIN_NICE..=MAX_NICE).contains(&nice) {
2336 return Err(E::from(format!(
2337 "Thread pool '{}' niceness {nice} is out of range ({MIN_NICE}..={MAX_NICE})",
2338 pool.id
2339 )));
2340 }
2341 }
2342 SchedulingPolicy::Fair => {}
2343 }
2344
2345 if let Some(affinity) = &pool.affinity
2346 && affinity.is_empty()
2347 {
2348 return Err(E::from(format!(
2349 "Thread pool '{}' has an empty affinity list; omit `affinity` for no pinning",
2350 pool.id
2351 )));
2352 }
2353 }
2354
2355 Ok(())
2356}
2357
2358pub const MAX_RATE_TARGET_HZ: u64 = 1_000_000_000;
2363
2364#[derive(Serialize, Deserialize, Debug, Clone)]
2366pub struct MissionsConfig {
2367 pub id: String,
2368}
2369
2370#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2372pub enum ConfigPredicate {
2373 Feature(String),
2374 Not(Box<ConfigPredicate>),
2375 All(Vec<ConfigPredicate>),
2376 Any(Vec<ConfigPredicate>),
2377}
2378
2379#[cfg(feature = "std")]
2380impl ConfigPredicate {
2381 fn evaluate(&self, active_features: &[&str]) -> bool {
2382 match self {
2383 Self::Feature(feature) => active_features.contains(&feature.as_str()),
2384 Self::Not(predicate) => !predicate.evaluate(active_features),
2385 Self::All(predicates) => predicates
2386 .iter()
2387 .all(|predicate| predicate.evaluate(active_features)),
2388 Self::Any(predicates) => predicates
2389 .iter()
2390 .any(|predicate| predicate.evaluate(active_features)),
2391 }
2392 }
2393}
2394
2395#[derive(Serialize, Deserialize, Debug, Clone)]
2397pub struct IncludesConfig {
2398 pub path: String,
2399 #[serde(default)]
2400 #[serde(serialize_with = "serialize_ordered_map")]
2401 pub params: HashMap<String, Value>,
2402 #[serde(default)]
2403 pub missions: Option<Vec<String>>,
2404 #[serde(default)]
2405 pub when: Option<ConfigPredicate>,
2406}
2407
2408#[cfg(feature = "std")]
2410#[allow(dead_code)]
2411#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2412pub struct MultiCopperSubsystemConfig {
2413 pub id: String,
2414 pub config: String,
2415}
2416
2417#[cfg(feature = "std")]
2419#[allow(dead_code)]
2420#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2421pub struct MultiCopperInterconnectConfig {
2422 pub from: String,
2423 pub to: String,
2424 pub msg: String,
2425 #[serde(default)]
2426 pub when: Option<ConfigPredicate>,
2427}
2428
2429#[cfg(feature = "std")]
2431#[allow(dead_code)]
2432#[derive(Serialize, Deserialize, Debug, Clone)]
2433pub struct InstanceConfigSetOperation {
2434 pub path: String,
2435 pub value: ComponentConfig,
2436}
2437
2438#[cfg(feature = "std")]
2440#[allow(dead_code)]
2441#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2442pub struct MultiCopperEndpoint {
2443 pub subsystem_id: String,
2444 pub bridge_id: String,
2445 pub channel_id: String,
2446}
2447
2448#[cfg(feature = "std")]
2449impl Display for MultiCopperEndpoint {
2450 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2451 write!(
2452 f,
2453 "{}/{}/{}",
2454 self.subsystem_id, self.bridge_id, self.channel_id
2455 )
2456 }
2457}
2458
2459#[cfg(feature = "std")]
2461#[allow(dead_code)]
2462#[derive(Debug, Clone)]
2463pub struct MultiCopperSubsystem {
2464 pub id: String,
2465 pub subsystem_code: u16,
2466 pub config_path: String,
2467 pub config: CuConfig,
2468}
2469
2470#[cfg(feature = "std")]
2472#[allow(dead_code)]
2473#[derive(Debug, Clone, PartialEq, Eq)]
2474pub struct MultiCopperInterconnect {
2475 pub from: MultiCopperEndpoint,
2476 pub to: MultiCopperEndpoint,
2477 pub msg: String,
2478 pub bridge_type: String,
2479}
2480
2481#[cfg(feature = "std")]
2483#[allow(dead_code)]
2484#[derive(Debug, Clone)]
2485pub struct MultiCopperConfig {
2486 pub subsystems: Vec<MultiCopperSubsystem>,
2487 pub interconnects: Vec<MultiCopperInterconnect>,
2488 pub instance_overrides_root: Option<String>,
2489}
2490
2491#[cfg(feature = "std")]
2492impl MultiCopperConfig {
2493 #[allow(dead_code)]
2494 pub fn subsystem(&self, id: &str) -> Option<&MultiCopperSubsystem> {
2495 self.subsystems.iter().find(|subsystem| subsystem.id == id)
2496 }
2497
2498 #[allow(dead_code)]
2499 pub fn resolve_subsystem_config_for_instance(
2500 &self,
2501 subsystem_id: &str,
2502 instance_id: u32,
2503 ) -> CuResult<CuConfig> {
2504 let subsystem = self.subsystem(subsystem_id).ok_or_else(|| {
2505 CuError::from(format!(
2506 "Multi-Copper config does not define subsystem '{}'.",
2507 subsystem_id
2508 ))
2509 })?;
2510 let mut config = subsystem.config.clone();
2511
2512 let Some(root) = &self.instance_overrides_root else {
2513 return Ok(config);
2514 };
2515
2516 let override_path = std::path::Path::new(root)
2517 .join(instance_id.to_string())
2518 .join(format!("{subsystem_id}.ron"));
2519 if !override_path.exists() {
2520 return Ok(config);
2521 }
2522
2523 apply_instance_overrides_from_file(&mut config, &override_path)?;
2524 Ok(config)
2525 }
2526}
2527
2528#[cfg(feature = "std")]
2529#[allow(dead_code)]
2530#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2531struct MultiCopperConfigRepresentation {
2532 subsystems: Vec<MultiCopperSubsystemConfig>,
2533 interconnects: Vec<MultiCopperInterconnectConfig>,
2534 instance_overrides_root: Option<String>,
2535}
2536
2537#[cfg(feature = "std")]
2538#[derive(Serialize, Deserialize, Debug, Clone, Default)]
2539struct InstanceConfigOverridesRepresentation {
2540 #[serde(default)]
2541 set: Vec<InstanceConfigSetOperation>,
2542}
2543
2544#[cfg(feature = "std")]
2545#[allow(dead_code)]
2546#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2547enum MultiCopperChannelDirection {
2548 Rx,
2549 Tx,
2550}
2551
2552#[cfg(feature = "std")]
2553#[allow(dead_code)]
2554#[derive(Debug, Clone)]
2555struct MultiCopperChannelContract {
2556 bridge_type: String,
2557 direction: MultiCopperChannelDirection,
2558 msg: Option<String>,
2559}
2560
2561#[cfg(feature = "std")]
2562#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2563enum InstanceConfigTargetKind {
2564 Task,
2565 Resource,
2566 Bridge,
2567}
2568
2569#[derive(Serialize, Deserialize, Default)]
2571struct CuConfigRepresentation {
2572 constants: Option<Vec<ConstantConfig>>,
2573 tasks: Option<Vec<Node>>,
2574 resources: Option<Vec<ResourceBundleConfig>>,
2575 log_streaming: Option<LogStreamingConfig>,
2576 bridges: Option<Vec<BridgeConfig>>,
2577 cnx: Option<Vec<SerializedCnx>>,
2578 #[serde(
2579 default,
2580 alias = "monitor",
2581 deserialize_with = "deserialize_monitor_configs"
2582 )]
2583 monitors: Option<Vec<MonitorConfig>>,
2584 logging: Option<LoggingConfig>,
2585 runtime: Option<RuntimeConfig>,
2586 missions: Option<Vec<MissionsConfig>>,
2587 includes: Option<Vec<IncludesConfig>>,
2588}
2589
2590#[derive(Deserialize)]
2591#[serde(untagged)]
2592enum OneOrManyMonitorConfig {
2593 One(MonitorConfig),
2594 Many(Vec<MonitorConfig>),
2595}
2596
2597fn deserialize_monitor_configs<'de, D>(
2598 deserializer: D,
2599) -> Result<Option<Vec<MonitorConfig>>, D::Error>
2600where
2601 D: Deserializer<'de>,
2602{
2603 let parsed = Option::<OneOrManyMonitorConfig>::deserialize(deserializer)?;
2604 Ok(parsed.map(|value| match value {
2605 OneOrManyMonitorConfig::One(single) => vec![single],
2606 OneOrManyMonitorConfig::Many(many) => many,
2607 }))
2608}
2609
2610fn deserialize_config_representation<E>(
2612 representation: &CuConfigRepresentation,
2613) -> Result<CuConfig, E>
2614where
2615 E: From<String>,
2616{
2617 let mut cuconfig = CuConfig::default();
2618 let bridge_lookup = build_bridge_lookup(representation.bridges.as_ref());
2619
2620 if let Some(mission_configs) = &representation.missions {
2621 let mut missions = Missions(HashMap::new());
2623
2624 for mission_config in mission_configs {
2625 let mission_id = mission_config.id.as_str();
2626 let graph = missions
2627 .add_mission(mission_id)
2628 .map_err(|e| E::from(e.to_string()))?;
2629
2630 if let Some(tasks) = &representation.tasks {
2631 for task in tasks {
2632 if let Some(task_missions) = &task.missions {
2633 if task_missions.contains(&mission_id.to_owned()) {
2635 graph
2636 .add_node(task.clone())
2637 .map_err(|e| E::from(e.to_string()))?;
2638 }
2639 } else {
2640 graph
2642 .add_node(task.clone())
2643 .map_err(|e| E::from(e.to_string()))?;
2644 }
2645 }
2646 }
2647
2648 if let Some(bridges) = &representation.bridges {
2649 for bridge in bridges {
2650 if mission_applies(&bridge.missions, mission_id) {
2651 insert_bridge_node(graph, bridge).map_err(E::from)?;
2652 }
2653 }
2654 }
2655
2656 if let Some(cnx) = &representation.cnx {
2657 for (connection_order, c) in cnx.iter().enumerate() {
2658 if let Some(cnx_missions) = &c.missions {
2659 if cnx_missions.contains(&mission_id.to_owned()) {
2661 if c.dst == NC_ENDPOINT {
2662 register_nc_output::<E>(
2663 graph,
2664 &c.src,
2665 &c.msg,
2666 connection_order,
2667 &bridge_lookup,
2668 )?;
2669 continue;
2670 }
2671 let (src_name, src_channel) =
2672 parse_endpoint(&c.src, EndpointRole::Source, &bridge_lookup)
2673 .map_err(E::from)?;
2674 let (dst_name, dst_channel) =
2675 parse_endpoint(&c.dst, EndpointRole::Destination, &bridge_lookup)
2676 .map_err(E::from)?;
2677 let src =
2678 graph
2679 .get_node_id_by_name(src_name.as_str())
2680 .ok_or_else(|| {
2681 E::from(format!("Source node not found: {}", c.src))
2682 })?;
2683 let dst =
2684 graph
2685 .get_node_id_by_name(dst_name.as_str())
2686 .ok_or_else(|| {
2687 E::from(format!("Destination node not found: {}", c.dst))
2688 })?;
2689 graph
2690 .connect_ext_with_order(
2691 src,
2692 dst,
2693 &c.msg,
2694 Some(cnx_missions.clone()),
2695 src_channel,
2696 dst_channel,
2697 connection_order,
2698 )
2699 .map_err(|e| E::from(e.to_string()))?;
2700 }
2701 } else {
2702 if c.dst == NC_ENDPOINT {
2704 register_nc_output::<E>(
2705 graph,
2706 &c.src,
2707 &c.msg,
2708 connection_order,
2709 &bridge_lookup,
2710 )?;
2711 continue;
2712 }
2713 let (src_name, src_channel) =
2714 parse_endpoint(&c.src, EndpointRole::Source, &bridge_lookup)
2715 .map_err(E::from)?;
2716 let (dst_name, dst_channel) =
2717 parse_endpoint(&c.dst, EndpointRole::Destination, &bridge_lookup)
2718 .map_err(E::from)?;
2719 let src = graph
2720 .get_node_id_by_name(src_name.as_str())
2721 .ok_or_else(|| E::from(format!("Source node not found: {}", c.src)))?;
2722 let dst =
2723 graph
2724 .get_node_id_by_name(dst_name.as_str())
2725 .ok_or_else(|| {
2726 E::from(format!("Destination node not found: {}", c.dst))
2727 })?;
2728 graph
2729 .connect_ext_with_order(
2730 src,
2731 dst,
2732 &c.msg,
2733 None,
2734 src_channel,
2735 dst_channel,
2736 connection_order,
2737 )
2738 .map_err(|e| E::from(e.to_string()))?;
2739 }
2740 }
2741 }
2742 }
2743 cuconfig.graphs = missions;
2744 } else {
2745 let mut graph = CuGraph::default();
2747
2748 if let Some(tasks) = &representation.tasks {
2749 for task in tasks {
2750 graph
2751 .add_node(task.clone())
2752 .map_err(|e| E::from(e.to_string()))?;
2753 }
2754 }
2755
2756 if let Some(bridges) = &representation.bridges {
2757 for bridge in bridges {
2758 insert_bridge_node(&mut graph, bridge).map_err(E::from)?;
2759 }
2760 }
2761
2762 if let Some(cnx) = &representation.cnx {
2763 for (connection_order, c) in cnx.iter().enumerate() {
2764 if c.dst == NC_ENDPOINT {
2765 register_nc_output::<E>(
2766 &mut graph,
2767 &c.src,
2768 &c.msg,
2769 connection_order,
2770 &bridge_lookup,
2771 )?;
2772 continue;
2773 }
2774 let (src_name, src_channel) =
2775 parse_endpoint(&c.src, EndpointRole::Source, &bridge_lookup)
2776 .map_err(E::from)?;
2777 let (dst_name, dst_channel) =
2778 parse_endpoint(&c.dst, EndpointRole::Destination, &bridge_lookup)
2779 .map_err(E::from)?;
2780 let src = graph
2781 .get_node_id_by_name(src_name.as_str())
2782 .ok_or_else(|| E::from(format!("Source node not found: {}", c.src)))?;
2783 let dst = graph
2784 .get_node_id_by_name(dst_name.as_str())
2785 .ok_or_else(|| E::from(format!("Destination node not found: {}", c.dst)))?;
2786 graph
2787 .connect_ext_with_order(
2788 src,
2789 dst,
2790 &c.msg,
2791 None,
2792 src_channel,
2793 dst_channel,
2794 connection_order,
2795 )
2796 .map_err(|e| E::from(e.to_string()))?;
2797 }
2798 }
2799 cuconfig.graphs = Simple(graph);
2800 }
2801
2802 cuconfig.monitors = representation.monitors.clone().unwrap_or_default();
2803 cuconfig.constants = representation.constants.clone().unwrap_or_default();
2804 cuconfig.logging = representation.logging.clone();
2805 cuconfig.runtime = representation.runtime.clone();
2806 cuconfig.resources = representation.resources.clone().unwrap_or_default();
2807 cuconfig.log_streaming = representation.log_streaming.clone();
2808 cuconfig.bridges = representation.bridges.clone().unwrap_or_default();
2809
2810 validate_thread_pools::<E>(&cuconfig.runtime)?;
2811
2812 Ok(cuconfig)
2813}
2814
2815impl<'de> Deserialize<'de> for CuConfig {
2816 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
2818 where
2819 D: Deserializer<'de>,
2820 {
2821 let representation =
2822 CuConfigRepresentation::deserialize(deserializer).map_err(serde::de::Error::custom)?;
2823
2824 match deserialize_config_representation::<String>(&representation) {
2826 Ok(config) => Ok(config),
2827 Err(e) => Err(serde::de::Error::custom(e)),
2828 }
2829 }
2830}
2831
2832fn serialize_ordered_map<T: Serialize, S: Serializer>(
2833 map: &HashMap<String, T>,
2834 serializer: S,
2835) -> Result<S::Ok, S::Error> {
2836 map.iter().collect::<BTreeMap<_, _>>().serialize(serializer)
2837}
2838fn serialize_optional_ordered_map<T: Serialize, S: Serializer>(
2839 map: &Option<HashMap<String, T>>,
2840 serializer: S,
2841) -> Result<S::Ok, S::Error> {
2842 map.as_ref()
2843 .map(|map| map.iter().collect::<BTreeMap<_, _>>())
2844 .serialize(serializer)
2845}
2846
2847impl Serialize for CuConfig {
2848 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
2850 where
2851 S: Serializer,
2852 {
2853 let bridges = if self.bridges.is_empty() {
2854 None
2855 } else {
2856 Some(self.bridges.clone())
2857 };
2858 let resources = if self.resources.is_empty() {
2859 None
2860 } else {
2861 Some(self.resources.clone())
2862 };
2863 let monitors = (!self.monitors.is_empty()).then_some(self.monitors.clone());
2864 match &self.graphs {
2865 Simple(graph) => {
2866 let tasks: Vec<Node> = graph
2867 .0
2868 .node_indices()
2869 .map(|idx| graph.0[idx].clone())
2870 .filter(|node| node.get_flavor() == Flavor::Task)
2871 .collect();
2872
2873 let mut ordered_cnx: Vec<(usize, SerializedCnx)> = graph
2874 .0
2875 .edge_indices()
2876 .map(|edge_idx| {
2877 let edge = &graph.0[edge_idx];
2878 let order = if edge.order == usize::MAX {
2879 edge_idx.index()
2880 } else {
2881 edge.order
2882 };
2883 (order, SerializedCnx::from(edge))
2884 })
2885 .collect();
2886 for node_idx in graph.0.node_indices() {
2887 let node = &graph.0[node_idx];
2888 if node.get_flavor() != Flavor::Task {
2889 continue;
2890 }
2891 for (msg, order) in node.nc_outputs_with_order() {
2892 ordered_cnx.push((
2893 order,
2894 SerializedCnx {
2895 src: node.get_id(),
2896 dst: NC_ENDPOINT.to_string(),
2897 msg: msg.clone(),
2898 missions: None,
2899 },
2900 ));
2901 }
2902 }
2903 ordered_cnx.sort_by(|(order_a, cnx_a), (order_b, cnx_b)| {
2904 order_a
2905 .cmp(order_b)
2906 .then_with(|| cnx_a.src.cmp(&cnx_b.src))
2907 .then_with(|| cnx_a.dst.cmp(&cnx_b.dst))
2908 .then_with(|| cnx_a.msg.cmp(&cnx_b.msg))
2909 });
2910 let cnx: Vec<SerializedCnx> = ordered_cnx
2911 .into_iter()
2912 .map(|(_, serialized)| serialized)
2913 .collect();
2914
2915 CuConfigRepresentation {
2916 constants: (!self.constants.is_empty()).then_some(self.constants.clone()),
2917 tasks: Some(tasks),
2918 bridges: bridges.clone(),
2919 cnx: Some(cnx),
2920 monitors: monitors.clone(),
2921 logging: self.logging.clone(),
2922 runtime: self.runtime.clone(),
2923 resources: resources.clone(),
2924 log_streaming: self.log_streaming.clone(),
2925 missions: None,
2926 includes: None,
2927 }
2928 .serialize(serializer)
2929 }
2930 Missions(graphs) => {
2931 let graphs = graphs.iter().collect::<BTreeMap<_, _>>();
2932 let missions = graphs
2933 .keys()
2934 .map(|id| MissionsConfig { id: (*id).clone() })
2935 .collect();
2936
2937 let mut tasks = Vec::new();
2939 let mut ordered_cnx: Vec<(usize, SerializedCnx)> = Vec::new();
2940
2941 for (mission_id, graph) in graphs {
2942 for node_idx in graph.node_indices() {
2944 let node = &graph[node_idx];
2945 if node.get_flavor() == Flavor::Task
2946 && !tasks.iter().any(|n: &Node| n.id == node.id)
2947 {
2948 tasks.push(node.clone());
2949 }
2950 }
2951
2952 for edge_idx in graph.0.edge_indices() {
2954 let edge = &graph.0[edge_idx];
2955 let order = if edge.order == usize::MAX {
2956 edge_idx.index()
2957 } else {
2958 edge.order
2959 };
2960 let serialized = SerializedCnx::from(edge);
2961 if let Some((existing_order, existing_serialized)) =
2962 ordered_cnx.iter_mut().find(|(_, c)| {
2963 c.src == serialized.src
2964 && c.dst == serialized.dst
2965 && c.msg == serialized.msg
2966 })
2967 {
2968 if order < *existing_order {
2969 *existing_order = order;
2970 }
2971 merge_connection_missions(
2972 &mut existing_serialized.missions,
2973 &serialized.missions,
2974 );
2975 } else {
2976 ordered_cnx.push((order, serialized));
2977 }
2978 }
2979 for node_idx in graph.0.node_indices() {
2980 let node = &graph.0[node_idx];
2981 if node.get_flavor() != Flavor::Task {
2982 continue;
2983 }
2984 for (msg, order) in node.nc_outputs_with_order() {
2985 let serialized = SerializedCnx {
2986 src: node.get_id(),
2987 dst: NC_ENDPOINT.to_string(),
2988 msg: msg.clone(),
2989 missions: Some(vec![mission_id.clone()]),
2990 };
2991 if let Some((existing_order, existing_serialized)) =
2992 ordered_cnx.iter_mut().find(|(_, c)| {
2993 c.src == serialized.src
2994 && c.dst == serialized.dst
2995 && c.msg == serialized.msg
2996 })
2997 {
2998 if order < *existing_order {
2999 *existing_order = order;
3000 }
3001 merge_connection_missions(
3002 &mut existing_serialized.missions,
3003 &serialized.missions,
3004 );
3005 } else {
3006 ordered_cnx.push((order, serialized));
3007 }
3008 }
3009 }
3010 }
3011 ordered_cnx.sort_by(|(order_a, cnx_a), (order_b, cnx_b)| {
3012 order_a
3013 .cmp(order_b)
3014 .then_with(|| cnx_a.src.cmp(&cnx_b.src))
3015 .then_with(|| cnx_a.dst.cmp(&cnx_b.dst))
3016 .then_with(|| cnx_a.msg.cmp(&cnx_b.msg))
3017 });
3018 let cnx: Vec<SerializedCnx> = ordered_cnx
3019 .into_iter()
3020 .map(|(_, serialized)| serialized)
3021 .collect();
3022
3023 CuConfigRepresentation {
3024 constants: (!self.constants.is_empty()).then_some(self.constants.clone()),
3025 tasks: Some(tasks),
3026 resources: resources.clone(),
3027 log_streaming: self.log_streaming.clone(),
3028 bridges,
3029 cnx: Some(cnx),
3030 monitors,
3031 logging: self.logging.clone(),
3032 runtime: self.runtime.clone(),
3033 missions: Some(missions),
3034 includes: None,
3035 }
3036 .serialize(serializer)
3037 }
3038 }
3039 }
3040}
3041
3042impl Default for CuConfig {
3043 fn default() -> Self {
3044 CuConfig {
3045 constants: Vec::new(),
3046 graphs: Simple(CuGraph(StableDiGraph::new())),
3047 monitors: Vec::new(),
3048 logging: None,
3049 runtime: None,
3050 resources: Vec::new(),
3051 log_streaming: None,
3052 bridges: Vec::new(),
3053 }
3054 }
3055}
3056
3057impl CuConfig {
3060 #[allow(dead_code)]
3061 pub fn new_simple_type() -> Self {
3062 Self::default()
3063 }
3064
3065 #[allow(dead_code)]
3066 pub fn new_mission_type() -> Self {
3067 CuConfig {
3068 constants: Vec::new(),
3069 graphs: Missions(HashMap::new()),
3070 monitors: Vec::new(),
3071 logging: None,
3072 runtime: None,
3073 resources: Vec::new(),
3074 log_streaming: None,
3075 bridges: Vec::new(),
3076 }
3077 }
3078
3079 pub(crate) fn get_options() -> Options {
3080 Options::default()
3081 .with_default_extension(Extensions::IMPLICIT_SOME)
3082 .with_default_extension(Extensions::UNWRAP_NEWTYPES)
3083 .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
3084 }
3085
3086 #[allow(dead_code)]
3087 pub fn serialize_ron(&self) -> CuResult<String> {
3088 let ron = Self::get_options();
3089 let pretty = ron::ser::PrettyConfig::default();
3090 ron.to_string_pretty(&self, pretty)
3091 .map_err(|e| CuError::from(format!("Error serializing configuration: {e}")))
3092 }
3093
3094 #[allow(dead_code)]
3095 pub fn deserialize_ron(ron: &str) -> CuResult<Self> {
3096 let representation = Self::get_options().from_str(ron).map_err(|e| {
3097 CuError::from(format!(
3098 "Syntax Error in config: {} at position {}",
3099 e.code, e.span
3100 ))
3101 })?;
3102 Self::deserialize_impl(representation)
3103 .map_err(|e| CuError::from(format!("Error deserializing configuration: {e}")))
3104 }
3105
3106 fn deserialize_impl(representation: CuConfigRepresentation) -> Result<Self, String> {
3107 deserialize_config_representation(&representation)
3108 }
3109
3110 #[cfg(feature = "std")]
3112 #[allow(dead_code)]
3113 pub fn render(
3114 &self,
3115 output: &mut dyn std::io::Write,
3116 mission_id: Option<&str>,
3117 ) -> CuResult<()> {
3118 writeln!(output, "digraph G {{")
3119 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3120 writeln!(output, " graph [rankdir=LR, nodesep=0.8, ranksep=1.2];")
3121 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3122 writeln!(output, " node [shape=plain, fontname=\"Noto Sans\"];")
3123 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3124 writeln!(output, " edge [fontname=\"Noto Sans\"];")
3125 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3126
3127 let sections = match (&self.graphs, mission_id) {
3128 (Simple(graph), _) => vec![RenderSection { label: None, graph }],
3129 (Missions(graphs), Some(id)) => {
3130 let graph = graphs
3131 .get(id)
3132 .ok_or_else(|| CuError::from(format!("Mission {id} not found")))?;
3133 vec![RenderSection {
3134 label: Some(id.to_string()),
3135 graph,
3136 }]
3137 }
3138 (Missions(graphs), None) => {
3139 let mut missions: Vec<_> = graphs.iter().collect();
3140 missions.sort_by(|a, b| a.0.cmp(b.0));
3141 missions
3142 .into_iter()
3143 .map(|(label, graph)| RenderSection {
3144 label: Some(label.clone()),
3145 graph,
3146 })
3147 .collect()
3148 }
3149 };
3150
3151 for section in sections {
3152 self.render_section(output, section.graph, section.label.as_deref())?;
3153 }
3154
3155 writeln!(output, "}}")
3156 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3157 Ok(())
3158 }
3159
3160 #[allow(dead_code)]
3161 pub fn get_all_instances_configs(
3162 &self,
3163 mission_id: Option<&str>,
3164 ) -> Vec<Option<&ComponentConfig>> {
3165 let graph = self.graphs.get_graph(mission_id).unwrap();
3166 graph
3167 .get_all_nodes()
3168 .iter()
3169 .map(|(_, node)| node.get_instance_config())
3170 .collect()
3171 }
3172
3173 #[allow(dead_code)]
3174 pub fn get_graph(&self, mission_id: Option<&str>) -> CuResult<&CuGraph> {
3175 self.graphs.get_graph(mission_id)
3176 }
3177
3178 #[allow(dead_code)]
3179 pub fn get_graph_mut(&mut self, mission_id: Option<&str>) -> CuResult<&mut CuGraph> {
3180 self.graphs.get_graph_mut(mission_id)
3181 }
3182
3183 #[allow(dead_code)]
3184 pub fn get_monitor_config(&self) -> Option<&MonitorConfig> {
3185 self.monitors.first()
3186 }
3187
3188 #[allow(dead_code)]
3189 pub fn get_monitor_configs(&self) -> &[MonitorConfig] {
3190 &self.monitors
3191 }
3192
3193 #[allow(dead_code)]
3194 pub fn get_runtime_config(&self) -> Option<&RuntimeConfig> {
3195 self.runtime.as_ref()
3196 }
3197
3198 #[allow(dead_code)]
3199 pub fn find_task_node(&self, mission_id: Option<&str>, task_id: &str) -> Option<&Node> {
3200 self.get_graph(mission_id)
3201 .ok()?
3202 .get_all_nodes()
3203 .into_iter()
3204 .find_map(|(_, node)| {
3205 (node.get_flavor() == Flavor::Task && node.id == task_id).then_some(node)
3206 })
3207 }
3208
3209 #[allow(dead_code)]
3210 pub fn find_logging_codec_spec(&self, codec_id: &str) -> Option<&LoggingCodecSpec> {
3211 self.logging
3212 .as_ref()?
3213 .codecs
3214 .iter()
3215 .find(|spec| spec.id == codec_id)
3216 }
3217
3218 pub fn validate_constants(&self) -> CuResult<()> {
3220 fn validate_integer(
3221 id: &str,
3222 storage: ConstantStorage,
3223 number: ConstantNumber,
3224 ) -> CuResult<()> {
3225 let valid = match (storage, number) {
3226 (ConstantStorage::I8, ConstantNumber::Signed(value)) => i8::try_from(value).is_ok(),
3227 (ConstantStorage::I16, ConstantNumber::Signed(value)) => {
3228 i16::try_from(value).is_ok()
3229 }
3230 (ConstantStorage::I32, ConstantNumber::Signed(value)) => {
3231 i32::try_from(value).is_ok()
3232 }
3233 (ConstantStorage::I64, ConstantNumber::Signed(_)) => true,
3234 (ConstantStorage::Isize, ConstantNumber::Signed(value)) => {
3235 isize::try_from(value).is_ok()
3236 }
3237 (ConstantStorage::U8, ConstantNumber::Unsigned(value)) => {
3238 u8::try_from(value).is_ok()
3239 }
3240 (ConstantStorage::U16, ConstantNumber::Unsigned(value)) => {
3241 u16::try_from(value).is_ok()
3242 }
3243 (ConstantStorage::U32, ConstantNumber::Unsigned(value)) => {
3244 u32::try_from(value).is_ok()
3245 }
3246 (ConstantStorage::U64, ConstantNumber::Unsigned(_)) => true,
3247 (ConstantStorage::Usize, ConstantNumber::Unsigned(value)) => {
3248 usize::try_from(value).is_ok()
3249 }
3250 _ => false,
3251 };
3252 if valid {
3253 Ok(())
3254 } else {
3255 Err(CuError::from(format!(
3256 "Constant '{id}' value {number:?} cannot be represented as {}",
3257 storage.rust_type()
3258 )))
3259 }
3260 }
3261
3262 let mut ids = HashMap::new();
3263 for constant in &self.constants {
3264 if constant.id().is_empty() {
3265 return Err(CuError::from("Constant ids cannot be empty"));
3266 }
3267 if ids
3268 .insert((constant.module_path(), constant.id()), ())
3269 .is_some()
3270 {
3271 return Err(CuError::from(format!(
3272 "Duplicate constant '{}'. Constant ids must be unique within a module.",
3273 constant.qualified_id()
3274 )));
3275 }
3276
3277 match (
3278 constant.value.is_some(),
3279 constant.rust_type.as_deref(),
3280 constant.expression.as_deref(),
3281 ) {
3282 (true, None, None) => {}
3283 (true, _, _) => {
3284 return Err(CuError::from(format!(
3285 "Constant '{}' cannot combine numeric 'value' with 'type' or 'expression'",
3286 constant.id()
3287 )));
3288 }
3289 (false, Some(rust_type), Some(expression)) => {
3290 if constant.storage.is_some()
3291 || constant.quantity.is_some()
3292 || constant.unit.is_some()
3293 {
3294 return Err(CuError::from(format!(
3295 "Constant '{}' cannot combine 'type' and 'expression' with numeric 'storage', 'quantity', or 'unit'",
3296 constant.id()
3297 )));
3298 }
3299 if rust_type.trim().is_empty() {
3300 return Err(CuError::from(format!(
3301 "Constant '{}' type cannot be empty",
3302 constant.id()
3303 )));
3304 }
3305 if expression.trim().is_empty() {
3306 return Err(CuError::from(format!(
3307 "Constant '{}' expression cannot be empty",
3308 constant.id()
3309 )));
3310 }
3311 continue;
3312 }
3313 (false, Some(_), None) => {
3314 return Err(CuError::from(format!(
3315 "Constant '{}' declares 'type' without 'expression'",
3316 constant.id()
3317 )));
3318 }
3319 (false, None, Some(_)) => {
3320 return Err(CuError::from(format!(
3321 "Constant '{}' declares 'expression' without 'type'",
3322 constant.id()
3323 )));
3324 }
3325 (false, None, None) => {
3326 return Err(CuError::from(format!(
3327 "Constant '{}' must declare either numeric 'value' or both 'type' and 'expression'",
3328 constant.id()
3329 )));
3330 }
3331 }
3332
3333 if constant.quantity().is_none() && constant.explicit_unit().is_some() {
3334 return Err(CuError::from(format!(
3335 "Constant '{}' declares a unit without a quantity",
3336 constant.id()
3337 )));
3338 }
3339
3340 if constant.quantity().is_some() {
3341 if !constant.storage().supports_quantity() {
3342 return Err(CuError::from(format!(
3343 "Constant '{}' quantity '{}' requires storage f32 or f64, not {}",
3344 constant.id(),
3345 constant.quantity().map_or("", |quantity| quantity.name()),
3346 constant.storage().rust_type()
3347 )));
3348 }
3349 let normalized = match constant.storage() {
3350 ConstantStorage::F32 => constant.normalized_f32().map(|_| ()),
3351 ConstantStorage::F64 => constant.normalized_f64().map(|_| ()),
3352 _ => unreachable!("quantity storage was checked above"),
3353 };
3354 normalized.map_err(CuError::from)?;
3355 continue;
3356 }
3357
3358 let (_, numbers) = constant.numbers().map_err(CuError::from)?;
3359 for number in numbers {
3360 match constant.storage() {
3361 ConstantStorage::F32 => {
3362 if !(number.as_f64() as f32).is_finite() {
3363 return Err(CuError::from(format!(
3364 "Constant '{}' values must be finite",
3365 constant.id()
3366 )));
3367 }
3368 }
3369 ConstantStorage::F64 => {
3370 if !number.as_f64().is_finite() {
3371 return Err(CuError::from(format!(
3372 "Constant '{}' values must be finite",
3373 constant.id()
3374 )));
3375 }
3376 }
3377 storage => validate_integer(constant.id(), storage, number)?,
3378 }
3379 }
3380 }
3381 Ok(())
3382 }
3383
3384 pub fn validate_logging_config(&self) -> CuResult<()> {
3387 if let Some(logging) = &self.logging {
3388 return logging.validate();
3389 }
3390 Ok(())
3391 }
3392
3393 pub fn validate_runtime_config(&self) -> CuResult<()> {
3395 if let Some(runtime) = &self.runtime {
3396 runtime.validate()?;
3397 if let Some(planner) = &runtime.planner {
3398 let copperlist_count = self
3399 .logging
3400 .as_ref()
3401 .and_then(|logging| logging.copperlist_count)
3402 .unwrap_or(DEFAULT_COPPERLIST_COUNT);
3403 planner.max_in_flight(copperlist_count)?;
3404 }
3405 }
3406 Ok(())
3407 }
3408
3409 fn validate_background_input_configs(&self) -> CuResult<()> {
3410 let validate = |graph: &CuGraph| -> CuResult<()> {
3411 for (node_id, node) in graph.get_all_nodes() {
3412 if node.background_process_empty()
3413 && (!node.is_background()
3414 || resolve_task_kind_for_id(graph, node_id)? != TaskKind::Regular)
3415 {
3416 return Err(CuError::from(format!(
3417 "Task '{}': background_process_empty requires a regular background task",
3418 node.id
3419 )));
3420 }
3421 }
3422 Ok(())
3423 };
3424 match &self.graphs {
3425 Simple(graph) => validate(graph),
3426 Missions(graphs) => {
3427 for graph in graphs.values() {
3428 validate(graph)?;
3429 }
3430 Ok(())
3431 }
3432 }
3433 }
3434
3435 fn validate_stateless_configs(&self) -> CuResult<()> {
3436 match &self.graphs {
3437 Simple(graph) => validate_stateless_graph(graph),
3438 Missions(graphs) => {
3439 for graph in graphs.values() {
3440 validate_stateless_graph(graph)?;
3441 }
3442 Ok(())
3443 }
3444 }
3445 }
3446
3447 pub fn validate_anytime_configs(&self) -> CuResult<()> {
3468 let rate_target_hz = self.runtime.as_ref().and_then(|r| r.rate_target_hz);
3469 match &self.graphs {
3470 Simple(graph) => validate_anytime_graph(graph, rate_target_hz),
3471 Missions(graphs) => {
3472 for graph in graphs.values() {
3473 validate_anytime_graph(graph, rate_target_hz)?;
3474 }
3475 Ok(())
3476 }
3477 }
3478 }
3479}
3480
3481fn validate_anytime_graph(graph: &CuGraph, rate_target_hz: Option<u64>) -> CuResult<()> {
3485 for (node_id, node) in graph.get_all_nodes() {
3486 let Some(anytime) = node.anytime() else {
3487 continue;
3488 };
3489 anytime.validate(&node.id)?;
3490
3491 let kind = resolve_task_kind_for_id(graph, node_id)?;
3492 if kind != TaskKind::Regular {
3493 return Err(CuError::from(format!(
3494 "Task '{}' is declared with an anytime: policy but resolves to kind '{}'. Anytime refinement needs both an input and an output, so it is only supported on regular tasks.",
3495 node.id,
3496 kind.as_str()
3497 )));
3498 }
3499
3500 let input_count = graph.get_dst_edges(node_id)?.len();
3505 if input_count != 1 {
3506 return Err(CuError::from(format!(
3507 "Task '{}' is an anytime task and must have exactly one input connection (found {input_count}): the runner anchors the job on the input's Tov.",
3508 node.id
3509 )));
3510 }
3511 let output_count = graph.get_node_output_msg_types_by_id(node_id)?.len();
3512 if output_count > 1 {
3513 return Err(CuError::from(format!(
3514 "Task '{}' is an anytime task and must have exactly one output message type (found {output_count}): base() and every refine() write the same output slot.",
3515 node.id
3516 )));
3517 }
3518
3519 if node.is_background() {
3522 continue;
3523 }
3524
3525 if anytime.max_refines.is_none() {
3529 return Err(CuError::from(format!(
3530 "Task '{}' is a foreground anytime task and needs anytime.max_refines: the execution plan is static, so the refine step count must be known at compile time. time_budget_ms/max_age_ms remain early-stop conditions within those quanta.",
3531 node.id
3532 )));
3533 }
3534
3535 let Some(rate_target_hz) = rate_target_hz else {
3536 continue;
3537 };
3538 let window_ms = match (anytime.time_budget_ms, anytime.max_age_ms) {
3539 (Some(budget), Some(age)) => budget.min(age),
3540 (Some(budget), None) => budget,
3541 (None, Some(age)) => age,
3542 (None, None) => {
3543 return Err(CuError::from(format!(
3544 "Task '{}' is a foreground anytime task in a rate-limited config and needs a time bound: set anytime.time_budget_ms or anytime.max_age_ms. max_refines alone gives the runtime no time quantity to check against the {rate_target_hz} Hz loop period, and one slow quantum would silently overrun it.",
3545 node.id
3546 )));
3547 }
3548 };
3549 let period_ms = 1_000.0 / rate_target_hz as f64;
3550 if window_ms >= period_ms {
3551 return Err(CuError::from(format!(
3552 "Task '{}': the worst-case anytime window ({window_ms} ms) does not fit within the {rate_target_hz} Hz loop period ({period_ms} ms) with headroom for the rest of the copperlist. Tighten the time bound, lower runtime.rate_target_hz, or run the task with background: true.",
3553 node.id
3554 )));
3555 }
3556 }
3557 Ok(())
3558}
3559
3560fn validate_stateless_graph(graph: &CuGraph) -> CuResult<()> {
3561 for (node_id, node) in graph.get_all_nodes() {
3562 if node.get_declared_task_kind() != Some(TaskKind::Stateless) {
3563 continue;
3564 }
3565
3566 resolve_task_kind_for_id(graph, node_id)?;
3567 if node.is_background() {
3568 return Err(CuError::from(format!(
3569 "Task '{}' is declared as kind 'stateless_task' and cannot be backgrounded.",
3570 node.id
3571 )));
3572 }
3573 if node.is_anytime() {
3574 return Err(CuError::from(format!(
3575 "Task '{}' is declared as kind 'stateless_task' and cannot use an anytime policy.",
3576 node.id
3577 )));
3578 }
3579 }
3580 Ok(())
3581}
3582
3583#[cfg(feature = "std")]
3584#[derive(Default)]
3585#[doc(hidden)]
3586pub struct PortLookup {
3587 pub inputs: HashMap<String, String>,
3588 pub outputs: HashMap<String, String>,
3589 pub default_input: Option<String>,
3590 pub default_output: Option<String>,
3591}
3592
3593#[cfg(feature = "std")]
3594#[derive(Clone)]
3595#[doc(hidden)]
3596pub struct RenderNode {
3597 pub id: String,
3598 pub type_name: String,
3599 pub flavor: Flavor,
3600 pub inputs: Vec<String>,
3601 pub outputs: Vec<String>,
3602}
3603
3604#[cfg(feature = "std")]
3605#[derive(Clone)]
3606#[doc(hidden)]
3607pub struct RenderConnection {
3608 pub src: String,
3609 pub src_port: Option<String>,
3610 #[allow(dead_code)]
3611 pub src_channel: Option<String>,
3612 pub dst: String,
3613 pub dst_port: Option<String>,
3614 #[allow(dead_code)]
3615 pub dst_channel: Option<String>,
3616 pub msg: String,
3617}
3618
3619#[cfg(feature = "std")]
3620#[doc(hidden)]
3621pub struct RenderTopology {
3622 pub nodes: Vec<RenderNode>,
3623 pub connections: Vec<RenderConnection>,
3624}
3625
3626#[cfg(feature = "std")]
3627impl RenderTopology {
3628 pub fn sort_connections(&mut self) {
3629 self.connections.sort_by(|a, b| {
3630 a.src
3631 .cmp(&b.src)
3632 .then(a.dst.cmp(&b.dst))
3633 .then(a.msg.cmp(&b.msg))
3634 });
3635 }
3636}
3637
3638#[cfg(feature = "std")]
3639#[allow(dead_code)]
3640struct RenderSection<'a> {
3641 label: Option<String>,
3642 graph: &'a CuGraph,
3643}
3644
3645#[cfg(feature = "std")]
3646impl CuConfig {
3647 #[allow(dead_code)]
3648 fn render_section(
3649 &self,
3650 output: &mut dyn std::io::Write,
3651 graph: &CuGraph,
3652 label: Option<&str>,
3653 ) -> CuResult<()> {
3654 use std::fmt::Write as FmtWrite;
3655
3656 let mut topology = build_render_topology(graph, &self.bridges);
3657 topology.nodes.sort_by(|a, b| a.id.cmp(&b.id));
3658 topology.sort_connections();
3659
3660 let cluster_id = label.map(|lbl| format!("cluster_{}", sanitize_identifier(lbl)));
3661 if let Some(ref cluster_id) = cluster_id {
3662 writeln!(output, " subgraph \"{cluster_id}\" {{")
3663 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3664 writeln!(
3665 output,
3666 " label=<<B>Mission: {}</B>>;",
3667 encode_text(label.unwrap())
3668 )
3669 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3670 writeln!(
3671 output,
3672 " labelloc=t; labeljust=l; color=\"#bbbbbb\"; style=\"rounded\"; margin=20;"
3673 )
3674 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3675 }
3676 let indent = if cluster_id.is_some() {
3677 " "
3678 } else {
3679 " "
3680 };
3681 let node_prefix = label
3682 .map(|lbl| format!("{}__", sanitize_identifier(lbl)))
3683 .unwrap_or_default();
3684
3685 let mut port_lookup: HashMap<String, PortLookup> = HashMap::new();
3686 let mut id_lookup: HashMap<String, String> = HashMap::new();
3687
3688 for node in &topology.nodes {
3689 let node_idx = graph
3690 .get_node_id_by_name(node.id.as_str())
3691 .ok_or_else(|| CuError::from(format!("Node '{}' missing from graph", node.id)))?;
3692 let node_weight = graph
3693 .get_node(node_idx)
3694 .ok_or_else(|| CuError::from(format!("Node '{}' missing weight", node.id)))?;
3695
3696 let fillcolor = match node.flavor {
3697 Flavor::Bridge => "#faedcd",
3698 Flavor::Task => match resolve_task_kind_for_id(graph, node_idx)? {
3699 TaskKind::Source => "#ddefc7",
3700 TaskKind::Sink => "#cce0ff",
3701 TaskKind::Regular => "#f2f2f2",
3702 TaskKind::Stateless => "#cba6f7",
3703 },
3704 };
3705
3706 let port_base = format!("{}{}", node_prefix, sanitize_identifier(&node.id));
3707 let (inputs_table, input_map, default_input) =
3708 build_port_table("Inputs", &node.inputs, &port_base, "in");
3709 let (outputs_table, output_map, default_output) =
3710 build_port_table("Outputs", &node.outputs, &port_base, "out");
3711 let config_html = node_weight.config.as_ref().and_then(build_config_table);
3712
3713 let mut label_html = String::new();
3714 write!(
3715 label_html,
3716 "<TABLE BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\" CELLPADDING=\"6\" COLOR=\"gray\" BGCOLOR=\"white\">"
3717 )
3718 .unwrap();
3719 write!(
3720 label_html,
3721 "<TR><TD COLSPAN=\"2\" ALIGN=\"LEFT\" BGCOLOR=\"{fillcolor}\"><FONT POINT-SIZE=\"12\"><B>{}</B></FONT><BR/><FONT COLOR=\"dimgray\">[{}]</FONT></TD></TR>",
3722 encode_text(&node.id),
3723 encode_text(&node.type_name)
3724 )
3725 .unwrap();
3726 write!(
3727 label_html,
3728 "<TR><TD ALIGN=\"LEFT\" VALIGN=\"TOP\">{inputs_table}</TD><TD ALIGN=\"LEFT\" VALIGN=\"TOP\">{outputs_table}</TD></TR>"
3729 )
3730 .unwrap();
3731
3732 if let Some(config_html) = config_html {
3733 write!(
3734 label_html,
3735 "<TR><TD COLSPAN=\"2\" ALIGN=\"LEFT\">{config_html}</TD></TR>"
3736 )
3737 .unwrap();
3738 }
3739
3740 label_html.push_str("</TABLE>");
3741
3742 let identifier_raw = if node_prefix.is_empty() {
3743 node.id.clone()
3744 } else {
3745 format!("{node_prefix}{}", node.id)
3746 };
3747 let identifier = escape_dot_id(&identifier_raw);
3748 writeln!(output, "{indent}\"{identifier}\" [label=<{label_html}>];")
3749 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3750
3751 id_lookup.insert(node.id.clone(), identifier);
3752 port_lookup.insert(
3753 node.id.clone(),
3754 PortLookup {
3755 inputs: input_map,
3756 outputs: output_map,
3757 default_input,
3758 default_output,
3759 },
3760 );
3761 }
3762
3763 for cnx in &topology.connections {
3764 let src_id = id_lookup
3765 .get(&cnx.src)
3766 .ok_or_else(|| CuError::from(format!("Unknown node '{}'", cnx.src)))?;
3767 let dst_id = id_lookup
3768 .get(&cnx.dst)
3769 .ok_or_else(|| CuError::from(format!("Unknown node '{}'", cnx.dst)))?;
3770 let src_suffix = port_lookup
3771 .get(&cnx.src)
3772 .and_then(|lookup| lookup.resolve_output(cnx.src_port.as_deref()))
3773 .map(|port| format!(":\"{port}\":e"))
3774 .unwrap_or_default();
3775 let dst_suffix = port_lookup
3776 .get(&cnx.dst)
3777 .and_then(|lookup| lookup.resolve_input(cnx.dst_port.as_deref()))
3778 .map(|port| format!(":\"{port}\":w"))
3779 .unwrap_or_default();
3780 let msg = encode_text(&cnx.msg);
3781 writeln!(
3782 output,
3783 "{indent}\"{src_id}\"{src_suffix} -> \"{dst_id}\"{dst_suffix} [label=< <B><FONT COLOR=\"gray\">{msg}</FONT></B> >];"
3784 )
3785 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3786 }
3787
3788 if cluster_id.is_some() {
3789 writeln!(output, " }}")
3790 .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3791 }
3792
3793 Ok(())
3794 }
3795}
3796
3797#[cfg(feature = "std")]
3798#[doc(hidden)]
3799pub fn build_render_topology(graph: &CuGraph, bridges: &[BridgeConfig]) -> RenderTopology {
3800 let mut bridge_lookup = HashMap::new();
3801 for bridge in bridges {
3802 bridge_lookup.insert(bridge.id.as_str(), bridge);
3803 }
3804
3805 let mut nodes: Vec<RenderNode> = Vec::new();
3806 let mut node_lookup: HashMap<String, usize> = HashMap::new();
3807 for (node_idx, node) in graph.get_all_nodes() {
3808 let node_id = node.get_id();
3809 let mut inputs = Vec::new();
3810 let mut outputs = Vec::new();
3811 if node.get_flavor() == Flavor::Bridge
3812 && let Some(bridge) = bridge_lookup.get(node_id.as_str())
3813 {
3814 for channel in &bridge.channels {
3815 match channel {
3816 BridgeChannelConfigRepresentation::Rx { id, .. } => outputs.push(id.clone()),
3818 BridgeChannelConfigRepresentation::Tx { id, .. } => inputs.push(id.clone()),
3820 }
3821 }
3822 } else if node.get_flavor() == Flavor::Task {
3823 for (idx, msg) in graph
3824 .get_node_output_msg_types_by_id(node_idx)
3825 .unwrap_or_default()
3826 .into_iter()
3827 .enumerate()
3828 {
3829 outputs.push(format!("out{idx}: {msg}"));
3830 }
3831 }
3832
3833 node_lookup.insert(node_id.clone(), nodes.len());
3834 nodes.push(RenderNode {
3835 id: node_id,
3836 type_name: node.get_type().to_string(),
3837 flavor: node.get_flavor(),
3838 inputs,
3839 outputs,
3840 });
3841 }
3842
3843 let mut output_port_lookup: Vec<HashMap<String, String>> = vec![HashMap::new(); nodes.len()];
3844 for (node_idx, node) in graph.get_all_nodes() {
3845 let Some(&idx) = node_lookup.get(&node.get_id()) else {
3846 continue;
3847 };
3848 if node.get_flavor() != Flavor::Task {
3849 continue;
3850 }
3851 for (port_idx, msg) in graph
3852 .get_node_output_msg_types_by_id(node_idx)
3853 .unwrap_or_default()
3854 .into_iter()
3855 .enumerate()
3856 {
3857 output_port_lookup[idx].insert(msg.clone(), format!("out{port_idx}: {msg}"));
3858 }
3859 }
3860
3861 let mut auto_input_counts = vec![0usize; nodes.len()];
3862 for edge in graph.0.edge_references() {
3863 let cnx = edge.weight();
3864 if let Some(&idx) = node_lookup.get(&cnx.dst)
3865 && nodes[idx].flavor == Flavor::Task
3866 && cnx.dst_channel.is_none()
3867 {
3868 auto_input_counts[idx] += 1;
3869 }
3870 }
3871
3872 let mut next_auto_input = vec![0usize; nodes.len()];
3873 let mut connections = Vec::new();
3874 for edge in graph.0.edge_references() {
3875 let cnx = edge.weight();
3876 let mut src_port = cnx.src_channel.clone();
3877 let mut dst_port = cnx.dst_channel.clone();
3878
3879 if let Some(&idx) = node_lookup.get(&cnx.src) {
3880 let node = &mut nodes[idx];
3881 if node.flavor == Flavor::Task && src_port.is_none() {
3882 src_port = output_port_lookup[idx].get(&cnx.msg).cloned();
3883 }
3884 }
3885 if let Some(&idx) = node_lookup.get(&cnx.dst) {
3886 let node = &mut nodes[idx];
3887 if node.flavor == Flavor::Task && dst_port.is_none() {
3888 let count = auto_input_counts[idx];
3889 let next = if count <= 1 {
3890 "in".to_string()
3891 } else {
3892 let next = format!("in.{}", next_auto_input[idx]);
3893 next_auto_input[idx] += 1;
3894 next
3895 };
3896 node.inputs.push(next.clone());
3897 dst_port = Some(next);
3898 }
3899 }
3900
3901 connections.push(RenderConnection {
3902 src: cnx.src.clone(),
3903 src_port,
3904 src_channel: cnx.src_channel.clone(),
3905 dst: cnx.dst.clone(),
3906 dst_port,
3907 dst_channel: cnx.dst_channel.clone(),
3908 msg: cnx.msg.clone(),
3909 });
3910 }
3911
3912 RenderTopology { nodes, connections }
3913}
3914
3915#[cfg(feature = "std")]
3916impl PortLookup {
3917 pub fn resolve_input(&self, name: Option<&str>) -> Option<&str> {
3918 if let Some(name) = name
3919 && let Some(port) = self.inputs.get(name)
3920 {
3921 return Some(port.as_str());
3922 }
3923 self.default_input.as_deref()
3924 }
3925
3926 pub fn resolve_output(&self, name: Option<&str>) -> Option<&str> {
3927 if let Some(name) = name
3928 && let Some(port) = self.outputs.get(name)
3929 {
3930 return Some(port.as_str());
3931 }
3932 self.default_output.as_deref()
3933 }
3934}
3935
3936#[cfg(feature = "std")]
3937#[allow(dead_code)]
3938fn build_port_table(
3939 title: &str,
3940 names: &[String],
3941 base_id: &str,
3942 prefix: &str,
3943) -> (String, HashMap<String, String>, Option<String>) {
3944 use std::fmt::Write as FmtWrite;
3945
3946 let mut html = String::new();
3947 write!(
3948 html,
3949 "<TABLE BORDER=\"0\" CELLBORDER=\"0\" CELLSPACING=\"0\" CELLPADDING=\"1\">"
3950 )
3951 .unwrap();
3952 write!(
3953 html,
3954 "<TR><TD ALIGN=\"LEFT\"><FONT COLOR=\"dimgray\">{}</FONT></TD></TR>",
3955 encode_text(title)
3956 )
3957 .unwrap();
3958
3959 let mut lookup = HashMap::new();
3960 let mut default_port = None;
3961
3962 if names.is_empty() {
3963 html.push_str("<TR><TD ALIGN=\"LEFT\"><FONT COLOR=\"lightgray\">—</FONT></TD></TR>");
3964 } else {
3965 for (idx, name) in names.iter().enumerate() {
3966 let port_id = format!("{base_id}_{prefix}_{idx}");
3967 write!(
3968 html,
3969 "<TR><TD PORT=\"{port_id}\" ALIGN=\"LEFT\">{}</TD></TR>",
3970 encode_text(name)
3971 )
3972 .unwrap();
3973 lookup.insert(name.clone(), port_id.clone());
3974 if idx == 0 {
3975 default_port = Some(port_id);
3976 }
3977 }
3978 }
3979
3980 html.push_str("</TABLE>");
3981 (html, lookup, default_port)
3982}
3983
3984#[cfg(feature = "std")]
3985#[allow(dead_code)]
3986fn build_config_table(config: &ComponentConfig) -> Option<String> {
3987 use std::fmt::Write as FmtWrite;
3988
3989 if config.0.is_empty() {
3990 return None;
3991 }
3992
3993 let mut entries: Vec<_> = config.0.iter().collect();
3994 entries.sort_by(|a, b| a.0.cmp(b.0));
3995
3996 let mut html = String::new();
3997 html.push_str("<TABLE BORDER=\"0\" CELLBORDER=\"0\" CELLSPACING=\"0\" CELLPADDING=\"1\">");
3998 for (key, value) in entries {
3999 let value_txt = format!("{value}");
4000 write!(
4001 html,
4002 "<TR><TD ALIGN=\"LEFT\"><FONT COLOR=\"dimgray\">{}</FONT> = {}</TD></TR>",
4003 encode_text(key),
4004 encode_text(&value_txt)
4005 )
4006 .unwrap();
4007 }
4008 html.push_str("</TABLE>");
4009 Some(html)
4010}
4011
4012#[cfg(feature = "std")]
4013#[allow(dead_code)]
4014fn sanitize_identifier(value: &str) -> String {
4015 value
4016 .chars()
4017 .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
4018 .collect()
4019}
4020
4021#[cfg(feature = "std")]
4022#[allow(dead_code)]
4023fn escape_dot_id(value: &str) -> String {
4024 let mut escaped = String::with_capacity(value.len());
4025 for ch in value.chars() {
4026 match ch {
4027 '"' => escaped.push_str("\\\""),
4028 '\\' => escaped.push_str("\\\\"),
4029 _ => escaped.push(ch),
4030 }
4031 }
4032 escaped
4033}
4034
4035impl LoggingConfig {
4036 pub fn validate(&self) -> CuResult<()> {
4038 if let Some(copperlist_count) = self.copperlist_count
4039 && copperlist_count == 0
4040 {
4041 return Err(CuError::from(
4042 "CopperList count cannot be zero. Set logging.copperlist_count to at least 1.",
4043 ));
4044 }
4045
4046 if let Some(section_size_mib) = self.section_size_mib
4047 && let Some(slab_size_mib) = self.slab_size_mib
4048 && section_size_mib > slab_size_mib
4049 {
4050 return Err(CuError::from(format!(
4051 "Section size ({section_size_mib} MiB) cannot be larger than slab size ({slab_size_mib} MiB). Adjust the parameters accordingly."
4052 )));
4053 }
4054
4055 let mut codec_ids = HashMap::new();
4056 for codec in &self.codecs {
4057 if codec_ids.insert(codec.id.as_str(), ()).is_some() {
4058 return Err(CuError::from(format!(
4059 "Duplicate logging codec id '{}'. Codec ids must be unique.",
4060 codec.id
4061 )));
4062 }
4063 }
4064
4065 Ok(())
4066 }
4067}
4068
4069impl RuntimeConfig {
4070 pub fn clock_sync_config(&self) -> Option<&ClockSyncConfig> {
4072 #[cfg(feature = "clock-sync")]
4073 {
4074 self.clock.as_ref()
4075 }
4076 #[cfg(not(feature = "clock-sync"))]
4077 {
4078 None
4079 }
4080 }
4081
4082 pub fn validate(&self) -> CuResult<()> {
4084 if let Some(clock) = self.clock_sync_config() {
4085 clock.validate()?;
4086 }
4087 if let Some(rate_target_hz) = self.rate_target_hz {
4088 if rate_target_hz == 0 {
4089 return Err(CuError::from(
4090 "Runtime rate target cannot be zero. Set runtime.rate_target_hz to at least 1.",
4091 ));
4092 }
4093
4094 if rate_target_hz > MAX_RATE_TARGET_HZ {
4095 return Err(CuError::from(format!(
4096 "Runtime rate target ({rate_target_hz} Hz) exceeds the supported maximum of {MAX_RATE_TARGET_HZ} Hz."
4097 )));
4098 }
4099 }
4100
4101 Ok(())
4102 }
4103}
4104
4105#[allow(dead_code)] fn substitute_parameters(content: &str, params: &HashMap<String, Value>) -> String {
4107 let mut result = content.to_string();
4108
4109 for (key, value) in params {
4110 let pattern = format!("{{{{{key}}}}}");
4111 result = result.replace(&pattern, &value.to_string());
4112 }
4113
4114 result
4115}
4116
4117#[cfg(feature = "std")]
4119fn process_includes(
4120 file_path: &str,
4121 base_representation: CuConfigRepresentation,
4122 processed_files: &mut Vec<String>,
4123 active_features: &[&str],
4124) -> CuResult<CuConfigRepresentation> {
4125 processed_files.push(file_path.to_string());
4127
4128 let mut result = base_representation;
4129
4130 if let Some(includes) = result.includes.take() {
4131 for include in includes {
4132 if include
4133 .when
4134 .as_ref()
4135 .is_some_and(|predicate| !predicate.evaluate(active_features))
4136 {
4137 continue;
4138 }
4139
4140 let include_path = if include.path.starts_with('/') {
4141 include.path.clone()
4142 } else {
4143 let current_dir = std::path::Path::new(file_path).parent();
4144
4145 match current_dir.map(|path| path.to_string_lossy().to_string()) {
4146 Some(current_dir) if !current_dir.is_empty() => {
4147 format!("{}/{}", current_dir, include.path)
4148 }
4149 _ => include.path,
4150 }
4151 };
4152
4153 let include_content = read_to_string(&include_path).map_err(|e| {
4154 CuError::from(format!("Failed to read include file: {include_path}"))
4155 .add_cause(e.to_string().as_str())
4156 })?;
4157
4158 let processed_content = substitute_parameters(&include_content, &include.params);
4159
4160 let mut included_representation: CuConfigRepresentation = match Options::default()
4161 .with_default_extension(Extensions::IMPLICIT_SOME)
4162 .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4163 .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4164 .from_str(&processed_content)
4165 {
4166 Ok(rep) => rep,
4167 Err(e) => {
4168 return Err(CuError::from(format!(
4169 "Failed to parse include file: {} - Error: {} at position {}",
4170 include_path, e.code, e.span
4171 )));
4172 }
4173 };
4174
4175 included_representation = process_includes(
4176 &include_path,
4177 included_representation,
4178 processed_files,
4179 active_features,
4180 )?;
4181
4182 if let Some(included_constants) = included_representation.constants {
4183 if result.constants.is_none() {
4184 result.constants = Some(included_constants);
4185 } else {
4186 let mut constants = result.constants.take().unwrap();
4187 for included_constant in included_constants {
4188 if !constants.iter().any(|constant| {
4189 constant.id == included_constant.id
4190 && constant.module_path() == included_constant.module_path()
4191 }) {
4192 constants.push(included_constant);
4193 }
4194 }
4195 result.constants = Some(constants);
4196 }
4197 }
4198
4199 if let Some(included_tasks) = included_representation.tasks {
4200 if result.tasks.is_none() {
4201 result.tasks = Some(included_tasks);
4202 } else {
4203 let mut tasks = result.tasks.take().unwrap();
4204 for included_task in included_tasks {
4205 if !tasks.iter().any(|t| t.id == included_task.id) {
4206 tasks.push(included_task);
4207 }
4208 }
4209 result.tasks = Some(tasks);
4210 }
4211 }
4212
4213 if let Some(included_bridges) = included_representation.bridges {
4214 if result.bridges.is_none() {
4215 result.bridges = Some(included_bridges);
4216 } else {
4217 let mut bridges = result.bridges.take().unwrap();
4218 for included_bridge in included_bridges {
4219 if !bridges.iter().any(|b| b.id == included_bridge.id) {
4220 bridges.push(included_bridge);
4221 }
4222 }
4223 result.bridges = Some(bridges);
4224 }
4225 }
4226
4227 if let Some(included_resources) = included_representation.resources {
4228 if result.resources.is_none() {
4229 result.resources = Some(included_resources);
4230 } else {
4231 let mut resources = result.resources.take().unwrap();
4232 for included_resource in included_resources {
4233 if !resources.iter().any(|r| r.id == included_resource.id) {
4234 resources.push(included_resource);
4235 }
4236 }
4237 result.resources = Some(resources);
4238 }
4239 }
4240
4241 if let Some(included_cnx) = included_representation.cnx {
4242 if result.cnx.is_none() {
4243 result.cnx = Some(included_cnx);
4244 } else {
4245 let mut cnx = result.cnx.take().unwrap();
4246 for included_c in included_cnx {
4247 if let Some(existing_cnx) = cnx.iter_mut().find(|c| {
4248 c.src == included_c.src
4249 && c.dst == included_c.dst
4250 && c.msg == included_c.msg
4251 }) {
4252 merge_connection_missions(
4253 &mut existing_cnx.missions,
4254 &included_c.missions,
4255 );
4256 } else {
4257 cnx.push(included_c);
4258 }
4259 }
4260 result.cnx = Some(cnx);
4261 }
4262 }
4263
4264 if let Some(included_monitors) = included_representation.monitors {
4265 if result.monitors.is_none() {
4266 result.monitors = Some(included_monitors);
4267 } else {
4268 let mut monitors = result.monitors.take().unwrap();
4269 for included_monitor in included_monitors {
4270 if !monitors.iter().any(|m| m.type_ == included_monitor.type_) {
4271 monitors.push(included_monitor);
4272 }
4273 }
4274 result.monitors = Some(monitors);
4275 }
4276 }
4277
4278 if result.logging.is_none() {
4279 result.logging = included_representation.logging;
4280 }
4281
4282 if result.runtime.is_none() {
4283 result.runtime = included_representation.runtime;
4284 }
4285
4286 if result.log_streaming.is_none() {
4287 result.log_streaming = included_representation.log_streaming;
4288 }
4289
4290 if let Some(included_missions) = included_representation.missions {
4291 if result.missions.is_none() {
4292 result.missions = Some(included_missions);
4293 } else {
4294 let mut missions = result.missions.take().unwrap();
4295 for included_mission in included_missions {
4296 if !missions.iter().any(|m| m.id == included_mission.id) {
4297 missions.push(included_mission);
4298 }
4299 }
4300 result.missions = Some(missions);
4301 }
4302 }
4303 }
4304 }
4305
4306 Ok(result)
4307}
4308
4309#[cfg(feature = "std")]
4310fn parse_instance_config_overrides_string(
4311 content: &str,
4312) -> CuResult<InstanceConfigOverridesRepresentation> {
4313 Options::default()
4314 .with_default_extension(Extensions::IMPLICIT_SOME)
4315 .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4316 .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4317 .from_str(content)
4318 .map_err(|e| {
4319 CuError::from(format!(
4320 "Failed to parse instance override file: Error: {} at position {}",
4321 e.code, e.span
4322 ))
4323 })
4324}
4325
4326#[cfg(feature = "std")]
4327fn merge_component_config(target: &mut Option<ComponentConfig>, value: &ComponentConfig) {
4328 if let Some(existing) = target {
4329 existing.merge_from(value);
4330 } else {
4331 *target = Some(value.clone());
4332 }
4333}
4334
4335#[cfg(feature = "std")]
4336fn apply_task_config_override_to_graph(
4337 graph: &mut CuGraph,
4338 task_id: &str,
4339 value: &ComponentConfig,
4340) -> usize {
4341 let mut matches = 0usize;
4342 let node_indices: Vec<_> = graph.0.node_indices().collect();
4343 for node_index in node_indices {
4344 let node = &mut graph.0[node_index];
4345 if node.get_flavor() == Flavor::Task && node.id == task_id {
4346 merge_component_config(&mut node.config, value);
4347 matches += 1;
4348 }
4349 }
4350 matches
4351}
4352
4353#[cfg(feature = "std")]
4354fn apply_bridge_node_config_override_to_graph(
4355 graph: &mut CuGraph,
4356 bridge_id: &str,
4357 value: &ComponentConfig,
4358) {
4359 let node_indices: Vec<_> = graph.0.node_indices().collect();
4360 for node_index in node_indices {
4361 let node = &mut graph.0[node_index];
4362 if node.get_flavor() == Flavor::Bridge && node.id == bridge_id {
4363 merge_component_config(&mut node.config, value);
4364 }
4365 }
4366}
4367
4368#[cfg(feature = "std")]
4369fn parse_instance_override_target(path: &str) -> CuResult<(InstanceConfigTargetKind, String)> {
4370 let mut parts = path.split('/');
4371 let scope = parts.next().unwrap_or_default();
4372 let id = parts.next().unwrap_or_default();
4373 let leaf = parts.next().unwrap_or_default();
4374
4375 if scope.is_empty() || id.is_empty() || leaf.is_empty() || parts.next().is_some() {
4376 return Err(CuError::from(format!(
4377 "Invalid instance override path '{}'. Expected 'tasks/<id>/config', 'resources/<id>/config', or 'bridges/<id>/config'.",
4378 path
4379 )));
4380 }
4381
4382 if leaf != "config" {
4383 return Err(CuError::from(format!(
4384 "Invalid instance override path '{}'. Only the '/config' leaf is supported.",
4385 path
4386 )));
4387 }
4388
4389 let kind = match scope {
4390 "tasks" => InstanceConfigTargetKind::Task,
4391 "resources" => InstanceConfigTargetKind::Resource,
4392 "bridges" => InstanceConfigTargetKind::Bridge,
4393 _ => {
4394 return Err(CuError::from(format!(
4395 "Invalid instance override path '{}'. Supported roots are 'tasks', 'resources', and 'bridges'.",
4396 path
4397 )));
4398 }
4399 };
4400
4401 Ok((kind, id.to_string()))
4402}
4403
4404#[cfg(feature = "std")]
4405fn apply_instance_config_set_operation(
4406 config: &mut CuConfig,
4407 operation: &InstanceConfigSetOperation,
4408) -> CuResult<()> {
4409 let (target_kind, target_id) = parse_instance_override_target(&operation.path)?;
4410
4411 match target_kind {
4412 InstanceConfigTargetKind::Task => {
4413 let matches = match &mut config.graphs {
4414 ConfigGraphs::Simple(graph) => {
4415 apply_task_config_override_to_graph(graph, &target_id, &operation.value)
4416 }
4417 ConfigGraphs::Missions(graphs) => graphs
4418 .values_mut()
4419 .map(|graph| {
4420 apply_task_config_override_to_graph(graph, &target_id, &operation.value)
4421 })
4422 .sum(),
4423 };
4424
4425 if matches == 0 {
4426 return Err(CuError::from(format!(
4427 "Instance override path '{}' targets unknown task '{}'.",
4428 operation.path, target_id
4429 )));
4430 }
4431 }
4432 InstanceConfigTargetKind::Resource => {
4433 let mut matches = 0usize;
4434 for resource in &mut config.resources {
4435 if resource.id == target_id {
4436 merge_component_config(&mut resource.config, &operation.value);
4437 matches += 1;
4438 }
4439 }
4440 if matches == 0 {
4441 return Err(CuError::from(format!(
4442 "Instance override path '{}' targets unknown resource '{}'.",
4443 operation.path, target_id
4444 )));
4445 }
4446 }
4447 InstanceConfigTargetKind::Bridge => {
4448 let mut matches = 0usize;
4449 for bridge in &mut config.bridges {
4450 if bridge.id == target_id {
4451 merge_component_config(&mut bridge.config, &operation.value);
4452 matches += 1;
4453 }
4454 }
4455 if matches == 0 {
4456 return Err(CuError::from(format!(
4457 "Instance override path '{}' targets unknown bridge '{}'.",
4458 operation.path, target_id
4459 )));
4460 }
4461
4462 match &mut config.graphs {
4463 ConfigGraphs::Simple(graph) => {
4464 apply_bridge_node_config_override_to_graph(graph, &target_id, &operation.value);
4465 }
4466 ConfigGraphs::Missions(graphs) => {
4467 for graph in graphs.values_mut() {
4468 apply_bridge_node_config_override_to_graph(
4469 graph,
4470 &target_id,
4471 &operation.value,
4472 );
4473 }
4474 }
4475 }
4476 }
4477 }
4478
4479 Ok(())
4480}
4481
4482#[cfg(feature = "std")]
4483fn apply_instance_overrides(
4484 config: &mut CuConfig,
4485 overrides: &InstanceConfigOverridesRepresentation,
4486) -> CuResult<()> {
4487 for operation in &overrides.set {
4488 apply_instance_config_set_operation(config, operation)?;
4489 }
4490 Ok(())
4491}
4492
4493#[cfg(feature = "std")]
4494fn apply_instance_overrides_from_file(
4495 config: &mut CuConfig,
4496 override_path: &std::path::Path,
4497) -> CuResult<()> {
4498 let override_content = read_to_string(override_path).map_err(|e| {
4499 CuError::from(format!(
4500 "Failed to read instance override file '{}'",
4501 override_path.display()
4502 ))
4503 .add_cause(e.to_string().as_str())
4504 })?;
4505 let overrides = parse_instance_config_overrides_string(&override_content).map_err(|e| {
4506 CuError::from(format!(
4507 "Failed to parse instance override file '{}': {e}",
4508 override_path.display()
4509 ))
4510 })?;
4511 apply_instance_overrides(config, &overrides)
4512}
4513
4514#[cfg(feature = "std")]
4515#[allow(dead_code)]
4516fn parse_multi_config_string(content: &str) -> CuResult<MultiCopperConfigRepresentation> {
4517 Options::default()
4518 .with_default_extension(Extensions::IMPLICIT_SOME)
4519 .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4520 .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4521 .from_str(content)
4522 .map_err(|e| {
4523 CuError::from(format!(
4524 "Failed to parse multi-Copper configuration: Error: {} at position {}",
4525 e.code, e.span
4526 ))
4527 })
4528}
4529
4530#[cfg(feature = "std")]
4531#[allow(dead_code)]
4532fn resolve_relative_config_path(base_path: Option<&str>, referenced_path: &str) -> String {
4533 if referenced_path.starts_with('/') || base_path.is_none() {
4534 return referenced_path.to_string();
4535 }
4536
4537 let current_dir = std::path::Path::new(base_path.expect("checked above"))
4538 .parent()
4539 .unwrap_or_else(|| std::path::Path::new(""))
4540 .to_path_buf();
4541 current_dir
4542 .join(referenced_path)
4543 .to_string_lossy()
4544 .to_string()
4545}
4546
4547#[cfg(feature = "std")]
4548#[allow(dead_code)]
4549fn parse_multi_endpoint(endpoint: &str) -> CuResult<MultiCopperEndpoint> {
4550 let mut parts = endpoint.split('/');
4551 let subsystem_id = parts.next().unwrap_or_default();
4552 let bridge_id = parts.next().unwrap_or_default();
4553 let channel_id = parts.next().unwrap_or_default();
4554
4555 if subsystem_id.is_empty()
4556 || bridge_id.is_empty()
4557 || channel_id.is_empty()
4558 || parts.next().is_some()
4559 {
4560 return Err(CuError::from(format!(
4561 "Invalid multi-Copper endpoint '{endpoint}'. Expected 'subsystem/bridge/channel'."
4562 )));
4563 }
4564
4565 Ok(MultiCopperEndpoint {
4566 subsystem_id: subsystem_id.to_string(),
4567 bridge_id: bridge_id.to_string(),
4568 channel_id: channel_id.to_string(),
4569 })
4570}
4571
4572#[cfg(feature = "std")]
4573#[allow(dead_code)]
4574fn multi_channel_key(bridge_id: &str, channel_id: &str) -> String {
4575 format!("{bridge_id}/{channel_id}")
4576}
4577
4578#[cfg(feature = "std")]
4579#[allow(dead_code)]
4580fn register_multi_channel_msg(
4581 contracts: &mut HashMap<String, MultiCopperChannelContract>,
4582 bridge_id: &str,
4583 channel_id: &str,
4584 expected_direction: MultiCopperChannelDirection,
4585 msg: &str,
4586) -> CuResult<()> {
4587 let key = multi_channel_key(bridge_id, channel_id);
4588 let contract = contracts.get_mut(&key).ok_or_else(|| {
4589 CuError::from(format!(
4590 "Bridge channel '{bridge_id}/{channel_id}' is referenced by the graph but not declared in the bridge config."
4591 ))
4592 })?;
4593
4594 if contract.direction != expected_direction {
4595 let expected = match expected_direction {
4596 MultiCopperChannelDirection::Rx => "Rx",
4597 MultiCopperChannelDirection::Tx => "Tx",
4598 };
4599 return Err(CuError::from(format!(
4600 "Bridge channel '{bridge_id}/{channel_id}' is used as {expected} in the graph but declared with the opposite direction."
4601 )));
4602 }
4603
4604 match &contract.msg {
4605 Some(existing) if existing != msg => Err(CuError::from(format!(
4606 "Bridge channel '{bridge_id}/{channel_id}' carries inconsistent message types '{existing}' and '{msg}'."
4607 ))),
4608 Some(_) => Ok(()),
4609 None => {
4610 contract.msg = Some(msg.to_string());
4611 Ok(())
4612 }
4613 }
4614}
4615
4616#[cfg(feature = "std")]
4617#[allow(dead_code)]
4618fn build_multi_bridge_channel_contracts(
4619 config: &CuConfig,
4620) -> CuResult<HashMap<String, MultiCopperChannelContract>> {
4621 let graph = config
4622 .graphs
4623 .get_graph(Some(DEFAULT_MISSION_ID))
4624 .map_err(|e| {
4625 CuError::from(format!(
4626 "Multi-Copper subsystem configs with missions must define a '{DEFAULT_MISSION_ID}' mission: {e}"
4627 ))
4628 })?;
4629
4630 let mut contracts = HashMap::new();
4631 for bridge in &config.bridges {
4632 for channel in &bridge.channels {
4633 let (channel_id, direction) = match channel {
4634 BridgeChannelConfigRepresentation::Rx { id, .. } => {
4635 (id.as_str(), MultiCopperChannelDirection::Rx)
4636 }
4637 BridgeChannelConfigRepresentation::Tx { id, .. } => {
4638 (id.as_str(), MultiCopperChannelDirection::Tx)
4639 }
4640 };
4641
4642 let key = multi_channel_key(&bridge.id, channel_id);
4643 if contracts.contains_key(&key) {
4644 return Err(CuError::from(format!(
4645 "Duplicate bridge channel declaration for '{key}'."
4646 )));
4647 }
4648
4649 contracts.insert(
4650 key,
4651 MultiCopperChannelContract {
4652 bridge_type: bridge.type_.clone(),
4653 direction,
4654 msg: None,
4655 },
4656 );
4657 }
4658 }
4659
4660 for edge in graph.edges() {
4661 if let Some(channel_id) = &edge.src_channel {
4662 register_multi_channel_msg(
4663 &mut contracts,
4664 &edge.src,
4665 channel_id,
4666 MultiCopperChannelDirection::Rx,
4667 &edge.msg,
4668 )?;
4669 }
4670 if let Some(channel_id) = &edge.dst_channel {
4671 register_multi_channel_msg(
4672 &mut contracts,
4673 &edge.dst,
4674 channel_id,
4675 MultiCopperChannelDirection::Tx,
4676 &edge.msg,
4677 )?;
4678 }
4679 }
4680
4681 Ok(contracts)
4682}
4683
4684#[cfg(feature = "std")]
4685#[allow(dead_code)]
4686fn validate_multi_config_representation(
4687 representation: MultiCopperConfigRepresentation,
4688 file_path: Option<&str>,
4689 active_features: &[&str],
4690) -> CuResult<MultiCopperConfig> {
4691 if representation
4692 .instance_overrides_root
4693 .as_ref()
4694 .is_some_and(|root| root.trim().is_empty())
4695 {
4696 return Err(CuError::from(
4697 "Multi-Copper instance_overrides_root must not be empty.",
4698 ));
4699 }
4700
4701 if representation.subsystems.is_empty() {
4702 return Err(CuError::from(
4703 "Multi-Copper config must declare at least one subsystem.",
4704 ));
4705 }
4706 if representation.subsystems.len() > usize::from(u16::MAX) + 1 {
4707 return Err(CuError::from(
4708 "Multi-Copper config supports at most 65536 distinct subsystem ids.",
4709 ));
4710 }
4711
4712 let mut seen_subsystems = std::collections::HashSet::new();
4713 for subsystem in &representation.subsystems {
4714 if subsystem.id.trim().is_empty() {
4715 return Err(CuError::from(
4716 "Multi-Copper subsystem ids must not be empty.",
4717 ));
4718 }
4719 if !seen_subsystems.insert(subsystem.id.clone()) {
4720 return Err(CuError::from(format!(
4721 "Duplicate multi-Copper subsystem id '{}'.",
4722 subsystem.id
4723 )));
4724 }
4725 }
4726
4727 let mut sorted_ids: Vec<_> = representation
4728 .subsystems
4729 .iter()
4730 .map(|subsystem| subsystem.id.clone())
4731 .collect();
4732 sorted_ids.sort();
4733 let subsystem_code_map: HashMap<_, _> = sorted_ids
4734 .into_iter()
4735 .enumerate()
4736 .map(|(idx, id)| {
4737 (
4738 id,
4739 u16::try_from(idx).expect("subsystem count was validated against u16 range"),
4740 )
4741 })
4742 .collect();
4743
4744 let mut subsystem_contracts: HashMap<String, HashMap<String, MultiCopperChannelContract>> =
4745 HashMap::new();
4746 let mut subsystems = Vec::with_capacity(representation.subsystems.len());
4747
4748 for subsystem in representation.subsystems {
4749 let resolved_config_path = resolve_relative_config_path(file_path, &subsystem.config);
4750 let config = read_configuration_with_features(&resolved_config_path, active_features)
4751 .map_err(|e| {
4752 CuError::from(format!(
4753 "Failed to read subsystem '{}' from '{}': {e}",
4754 subsystem.id, resolved_config_path
4755 ))
4756 })?;
4757 let contracts = build_multi_bridge_channel_contracts(&config).map_err(|e| {
4758 CuError::from(format!(
4759 "Invalid subsystem '{}' for multi-Copper validation: {e}",
4760 subsystem.id
4761 ))
4762 })?;
4763 subsystem_contracts.insert(subsystem.id.clone(), contracts);
4764 subsystems.push(MultiCopperSubsystem {
4765 subsystem_code: *subsystem_code_map
4766 .get(&subsystem.id)
4767 .expect("subsystem code map must contain every subsystem"),
4768 id: subsystem.id,
4769 config_path: resolved_config_path,
4770 config,
4771 });
4772 }
4773
4774 let mut interconnects = Vec::with_capacity(representation.interconnects.len());
4775 for interconnect in representation.interconnects {
4776 if interconnect
4777 .when
4778 .as_ref()
4779 .is_some_and(|predicate| !predicate.evaluate(active_features))
4780 {
4781 continue;
4782 }
4783
4784 let from = parse_multi_endpoint(&interconnect.from).map_err(|e| {
4785 CuError::from(format!(
4786 "Invalid multi-Copper interconnect source '{}': {e}",
4787 interconnect.from
4788 ))
4789 })?;
4790 let to = parse_multi_endpoint(&interconnect.to).map_err(|e| {
4791 CuError::from(format!(
4792 "Invalid multi-Copper interconnect destination '{}': {e}",
4793 interconnect.to
4794 ))
4795 })?;
4796
4797 let from_contracts = subsystem_contracts.get(&from.subsystem_id).ok_or_else(|| {
4798 CuError::from(format!(
4799 "Interconnect source '{}' references unknown subsystem '{}'.",
4800 from, from.subsystem_id
4801 ))
4802 })?;
4803 let to_contracts = subsystem_contracts.get(&to.subsystem_id).ok_or_else(|| {
4804 CuError::from(format!(
4805 "Interconnect destination '{}' references unknown subsystem '{}'.",
4806 to, to.subsystem_id
4807 ))
4808 })?;
4809
4810 let from_contract = from_contracts
4811 .get(&multi_channel_key(&from.bridge_id, &from.channel_id))
4812 .ok_or_else(|| {
4813 CuError::from(format!(
4814 "Interconnect source '{}' references unknown bridge channel.",
4815 from
4816 ))
4817 })?;
4818 let to_contract = to_contracts
4819 .get(&multi_channel_key(&to.bridge_id, &to.channel_id))
4820 .ok_or_else(|| {
4821 CuError::from(format!(
4822 "Interconnect destination '{}' references unknown bridge channel.",
4823 to
4824 ))
4825 })?;
4826
4827 if from_contract.direction != MultiCopperChannelDirection::Tx {
4828 return Err(CuError::from(format!(
4829 "Interconnect source '{}' must reference a Tx bridge channel.",
4830 from
4831 )));
4832 }
4833 if to_contract.direction != MultiCopperChannelDirection::Rx {
4834 return Err(CuError::from(format!(
4835 "Interconnect destination '{}' must reference an Rx bridge channel.",
4836 to
4837 )));
4838 }
4839
4840 if from_contract.bridge_type != to_contract.bridge_type {
4841 return Err(CuError::from(format!(
4842 "Interconnect '{}' -> '{}' mixes incompatible bridge types '{}' and '{}'.",
4843 from, to, from_contract.bridge_type, to_contract.bridge_type
4844 )));
4845 }
4846
4847 let from_msg = from_contract.msg.as_ref().ok_or_else(|| {
4848 CuError::from(format!(
4849 "Interconnect source '{}' is not wired inside subsystem '{}', so its message type cannot be inferred.",
4850 from, from.subsystem_id
4851 ))
4852 })?;
4853 let to_msg = to_contract.msg.as_ref().ok_or_else(|| {
4854 CuError::from(format!(
4855 "Interconnect destination '{}' is not wired inside subsystem '{}', so its message type cannot be inferred.",
4856 to, to.subsystem_id
4857 ))
4858 })?;
4859
4860 if from_msg != to_msg {
4861 return Err(CuError::from(format!(
4862 "Interconnect '{}' -> '{}' connects incompatible message types '{}' and '{}'.",
4863 from, to, from_msg, to_msg
4864 )));
4865 }
4866 if interconnect.msg != *from_msg {
4867 return Err(CuError::from(format!(
4868 "Interconnect '{}' -> '{}' declares message type '{}' but subsystem graphs require '{}'.",
4869 from, to, interconnect.msg, from_msg
4870 )));
4871 }
4872
4873 interconnects.push(MultiCopperInterconnect {
4874 from,
4875 to,
4876 msg: interconnect.msg,
4877 bridge_type: from_contract.bridge_type.clone(),
4878 });
4879 }
4880
4881 let instance_overrides_root = representation
4882 .instance_overrides_root
4883 .as_ref()
4884 .map(|root| resolve_relative_config_path(file_path, root));
4885
4886 Ok(MultiCopperConfig {
4887 subsystems,
4888 interconnects,
4889 instance_overrides_root,
4890 })
4891}
4892
4893#[cfg(feature = "std")]
4895pub fn read_configuration(config_filename: &str) -> CuResult<CuConfig> {
4896 read_configuration_with_features(config_filename, &[])
4897}
4898
4899#[cfg(feature = "std")]
4901pub fn read_configuration_with_features(
4902 config_filename: &str,
4903 active_features: &[&str],
4904) -> CuResult<CuConfig> {
4905 let config_content = read_configuration_content(config_filename)?;
4906 read_configuration_str_with_features(config_content, Some(config_filename), active_features)
4907}
4908
4909#[cfg(feature = "std")]
4910fn read_configuration_content(config_filename: &str) -> CuResult<String> {
4911 read_to_string(config_filename).map_err(|e| {
4912 CuError::from(format!(
4913 "Failed to read configuration file: {:?}",
4914 config_filename
4915 ))
4916 .add_cause(e.to_string().as_str())
4917 })
4918}
4919
4920fn parse_config_string(content: &str) -> CuResult<CuConfigRepresentation> {
4924 Options::default()
4925 .with_default_extension(Extensions::IMPLICIT_SOME)
4926 .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4927 .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4928 .from_str(content)
4929 .map_err(|e| {
4930 CuError::from(format!(
4931 "Failed to parse configuration: Error: {} at position {}",
4932 e.code, e.span
4933 ))
4934 })
4935}
4936
4937fn config_representation_to_config(representation: CuConfigRepresentation) -> CuResult<CuConfig> {
4940 #[allow(unused_mut)]
4941 let mut cuconfig = CuConfig::deserialize_impl(representation)
4942 .map_err(|e| CuError::from(format!("Error deserializing configuration: {e}")))?;
4943
4944 #[cfg(feature = "std")]
4945 cuconfig.ensure_default_background_pool();
4946
4947 cuconfig.validate_logging_config()?;
4948 cuconfig.validate_runtime_config()?;
4949 cuconfig.validate_stateless_configs()?;
4950 cuconfig.validate_background_input_configs()?;
4951 cuconfig.validate_anytime_configs()?;
4952 cuconfig.validate_constants()?;
4953 cuconfig.validate_log_streaming_config()?;
4954
4955 Ok(cuconfig)
4956}
4957
4958#[allow(unused_variables)]
4959fn resolve_configuration_representation(
4960 config_content: &str,
4961 file_path: Option<&str>,
4962 active_features: &[&str],
4963) -> CuResult<CuConfigRepresentation> {
4964 let representation = parse_config_string(config_content)?;
4966
4967 #[cfg(feature = "std")]
4970 let representation = if let Some(path) = file_path {
4971 process_includes(path, representation, &mut Vec::new(), active_features)?
4972 } else {
4973 representation
4974 };
4975
4976 Ok(representation)
4977}
4978
4979#[cfg(feature = "std")]
4984#[doc(hidden)]
4985#[allow(dead_code)]
4986pub fn read_configuration_with_resolved_ron(config_filename: &str) -> CuResult<(CuConfig, String)> {
4987 read_configuration_with_resolved_ron_and_features(config_filename, &[])
4988}
4989
4990#[cfg(feature = "std")]
4992#[doc(hidden)]
4993pub fn read_configuration_with_resolved_ron_and_features(
4994 config_filename: &str,
4995 active_features: &[&str],
4996) -> CuResult<(CuConfig, String)> {
4997 let config_content = read_configuration_content(config_filename)?;
4998 let representation = resolve_configuration_representation(
4999 &config_content,
5000 Some(config_filename),
5001 active_features,
5002 )?;
5003 let resolved_ron = CuConfig::get_options()
5004 .to_string_pretty(&representation, ron::ser::PrettyConfig::default())
5005 .map_err(|e| CuError::from(format!("Error serializing configuration: {e}")))?;
5006 let config = config_representation_to_config(representation)?;
5007 Ok((config, resolved_ron))
5008}
5009
5010#[allow(dead_code)]
5011pub fn read_configuration_str(
5012 config_content: String,
5013 file_path: Option<&str>,
5014) -> CuResult<CuConfig> {
5015 read_configuration_str_with_features(config_content, file_path, &[])
5016}
5017
5018pub fn read_configuration_str_with_features(
5020 config_content: String,
5021 file_path: Option<&str>,
5022 active_features: &[&str],
5023) -> CuResult<CuConfig> {
5024 let representation =
5025 resolve_configuration_representation(&config_content, file_path, active_features)?;
5026
5027 config_representation_to_config(representation)
5029}
5030
5031#[cfg(feature = "std")]
5033#[allow(dead_code)]
5034pub fn read_multi_configuration(config_filename: &str) -> CuResult<MultiCopperConfig> {
5035 read_multi_configuration_with_features(config_filename, &[])
5036}
5037
5038#[cfg(feature = "std")]
5040#[allow(dead_code)]
5041pub fn read_multi_configuration_with_features(
5042 config_filename: &str,
5043 active_features: &[&str],
5044) -> CuResult<MultiCopperConfig> {
5045 let config_content = read_to_string(config_filename).map_err(|e| {
5046 CuError::from(format!(
5047 "Failed to read multi-Copper configuration file: {:?}",
5048 config_filename
5049 ))
5050 .add_cause(e.to_string().as_str())
5051 })?;
5052 read_multi_configuration_str_with_features(
5053 config_content,
5054 Some(config_filename),
5055 active_features,
5056 )
5057}
5058
5059#[cfg(feature = "std")]
5061#[allow(dead_code)]
5062pub fn read_multi_configuration_str(
5063 config_content: String,
5064 file_path: Option<&str>,
5065) -> CuResult<MultiCopperConfig> {
5066 read_multi_configuration_str_with_features(config_content, file_path, &[])
5067}
5068
5069#[cfg(feature = "std")]
5071#[allow(dead_code)]
5072pub fn read_multi_configuration_str_with_features(
5073 config_content: String,
5074 file_path: Option<&str>,
5075 active_features: &[&str],
5076) -> CuResult<MultiCopperConfig> {
5077 let representation = parse_multi_config_string(&config_content)?;
5078 validate_multi_config_representation(representation, file_path, active_features)
5079}
5080
5081#[cfg(test)]
5082mod tests;