hw resolution

Author: networmix

examples/small_test_clos.yml

@@ -17,47 +17,18 @@ build:
     intra_metro_link:
       capacity: 12_800
       cost: 1
-      attrs:
-        link_type: intra_metro
-      match:
-        conditions:
-          - attr: role
-            operator: "=="
-            value: core
-          - attr: role
-            operator: "=="
-            value: leaf
+      attrs: { link_type: intra_metro }
+      role_pairs: ["core|leaf"]
     inter_metro_link:
       capacity: 3_200
       cost: 1
-      attrs:
-        link_type: inter_metro_corridor
-      match:
-        conditions:
-          - attr: role
-            operator: "=="
-            value: core
-          - attr: role
-            operator: "=="
-            value: leaf
-        logic: "or"
+      attrs: { link_type: inter_metro_corridor }
+      role_pairs: ["core|leaf"]
     dc_to_pop_link:
       capacity: 25_600
       cost: 1
-      attrs:
-        link_type: dc_to_pop
-      match:
-        conditions:
-          - attr: role
-            operator: "=="
-            value: core
-          - attr: role
-            operator: "=="
-            value: dc
-          - attr: role
-            operator: "=="
-            value: leaf
-        logic: "or"
+      attrs: { link_type: dc_to_pop }
+      role_pairs: ["dc|core", "dc|leaf"]
   build_overrides:
     - metros: [new-york-jersey-city-newark, columbus, washington-arlington]
       site_blueprint: Clos_16_8
@@ -79,13 +50,17 @@ build:
     respect_min_base_capacity: true
 
 components:
-  assignments:
-    core:
-      hw_component: CoreRouter
-    leaf:
-      hw_component: LeafRouter
-    spine:
-      hw_component: SpineRouter
+  hw_component:
+    core: CoreRouter
+    leaf: LeafRouter
+    spine: SpineRouter
+    dc: {}
+  optics:
+    core|core: 800G-ZR+
+    leaf|leaf: 800G-ZR+
+    core|leaf: 800G-ZR+
+    leaf|spine: 800G-DR4
+    spine|leaf: 1600G-2xDR4
 
 failure_policies:
   assignments:

tests/test_components_scenario_builder.py

@@ -4,9 +4,6 @@
 
 from topogen.blueprints_lib import get_builtin_blueprints
 from topogen.config import (
-    ComponentAssignment,
-    ComponentAssignments,
-    ComponentsConfig,
     TopologyConfig,
 )
 from topogen.scenario_builder import (
@@ -19,19 +16,21 @@ class TestComponentsScenarioBuilder:
     """Test component-related scenario builder functionality."""
 
     def create_test_config(self) -> TopologyConfig:
-        """Create a test configuration with component assignments."""
+        """Create a test configuration with streamlined role mappings."""
         config = TopologyConfig()
-
-        # Set up component assignments
-        config.components = ComponentsConfig(
-            assignments=ComponentAssignments(
-                spine=ComponentAssignment(hw_component="CoreRouter", optics="800G-ZR+"),
-                leaf=ComponentAssignment(hw_component="CoreRouter", optics="800G-ZR+"),
-                core=ComponentAssignment(hw_component="CoreRouter", optics="800G-ZR+"),
-                dc=ComponentAssignment(hw_component="CoreRouter", optics="800G-ZR+"),
-            ),
-        )
-
+        # Streamlined mapping: role -> platform, and role-pair -> optic (source end)
+        config.components.hw_component = {
+            "spine": "CoreRouter",
+            "leaf": "CoreRouter",
+            "core": "CoreRouter",
+            "dc": "CoreRouter",
+        }
+        config.components.optics = {
+            "core-core": "800G-ZR+",
+            # Add a couple to exercise inclusion; actual adjacency tests are elsewhere
+            "leaf-spine": "800G-DR4",
+            "spine-leaf": "1600G-2xDR4",
+        }
         return config
 
     def test_build_components_section_basic(self):
@@ -92,8 +91,8 @@ def test_build_components_section_missing_component_warning(self):
         """Test warning when referenced component is not found."""
         config = self.create_test_config()
 
-        # Reference a non-existent component
-        config.components.assignments.spine.hw_component = "NonExistentChassis"
+        # Reference a non-existent component via streamlined mapping
+        config.components.hw_component["spine"] = "NonExistentChassis"
 
         used_blueprints = {"Clos_64_256"}
 

tests/test_site_graph_sot.py

@@ -35,10 +35,12 @@ def test_sites_section_includes_per_site_blueprint_and_components():
     cfg.build.build_defaults.dc_region_blueprint = "DCRegion"
 
     # Provide explicit role->component assignments to be serialized per site
-    cfg.components.assignments.core.hw_component = "CoreRouter"
-    cfg.components.assignments.leaf.hw_component = "LeafRouter"
-    cfg.components.assignments.spine.hw_component = "SpineRouter"
-    cfg.components.assignments.dc.hw_component = "CoreRouter"
+    cfg.components.hw_component = {
+        "core": "CoreRouter",
+        "leaf": "LeafRouter",
+        "spine": "SpineRouter",
+        "dc": "CoreRouter",
+    }
 
     yaml_str = build_scenario(graph, cfg)
     data = yaml.safe_load(yaml_str)

topogen/cli.py

@@ -116,14 +116,11 @@ def build_command(args: argparse.Namespace) -> None:
         sys.exit(3)  # Validation failure
     except ValueError as e:
         logger.error(f"Validation error: {e}")
-        print(f"❌ Validation error: {e}")
         print("💡 Check input data quality and configuration parameters")
         sys.exit(3)  # Validation failure
     except Exception as e:
         logger.error(f"Pipeline failed: {e}")
-        print(f"❌ ERROR: {e}")
         print("💡 Use -v for detailed error information")
-
         sys.exit(1)  # Runtime error
 
 

topogen/config.py

@@ -130,6 +130,11 @@ class LinkParams:
     cost: int
     attrs: dict[str, Any] = field(default_factory=dict)
     match: dict[str, Any] = field(default_factory=dict)
+    # Unordered role-pairs allowed for this link type. Examples:
+    #   ["core|core", "core|leaf"] or [["core", "core"], ["core", "leaf"]]
+    # The pipeline will auto-render a symmetric match from the union of roles.
+    role_pairs: list[Any] = field(default_factory=list)
+    # endpoint_roles removed: explicit direction should not be configured.
 
 
 @dataclass
@@ -280,6 +285,11 @@ class ComponentsConfig:
     """
 
     assignments: ComponentAssignments = field(default_factory=ComponentAssignments)
+    # New streamlined mappings (preferred over assignments when provided)
+    # hw_component: role -> platform component name
+    # optics: "srcRole-dstRole" -> optic component name (applies to source end)
+    hw_component: dict[str, str] = field(default_factory=dict)
+    optics: dict[str, str] = field(default_factory=dict)
 
 
 @dataclass
@@ -653,6 +663,7 @@ def _from_dict(cls, config_dict: dict[str, Any]) -> TopologyConfig:
                 **intra_metro_link_dict.get("attrs", {}),
             },
             match=intra_metro_link_dict.get("match", {}),
+            role_pairs=intra_metro_link_dict.get("role_pairs", []) or [],
         )
         inter_metro_link = LinkParams(
             capacity=inter_metro_link_dict.get("capacity", 100),
@@ -662,6 +673,7 @@ def _from_dict(cls, config_dict: dict[str, Any]) -> TopologyConfig:
                 **inter_metro_link_dict.get("attrs", {}),
             },
             match=inter_metro_link_dict.get("match", {}),
+            role_pairs=inter_metro_link_dict.get("role_pairs", []) or [],
         )
         dc_to_pop_link = LinkParams(
             capacity=dc_to_pop_link_dict.get("capacity", 400),
@@ -671,6 +683,7 @@ def _from_dict(cls, config_dict: dict[str, Any]) -> TopologyConfig:
                 **dc_to_pop_link_dict.get("attrs", {}),
             },
             match=dc_to_pop_link_dict.get("match", {}),
+            role_pairs=dc_to_pop_link_dict.get("role_pairs", []) or [],
         )
 
         # Create BuildDefaults with explicit parameters
@@ -783,7 +796,7 @@ def _from_dict(cls, config_dict: dict[str, Any]) -> TopologyConfig:
             tm_sizing=tm_sizing,
         )
 
-        # Handle optional components configuration (assignments only)
+        # Handle optional components configuration (streamlined mappings)
         components_dict = config_dict.get("components", {})
         if not isinstance(components_dict, dict):
             raise ValueError("'components' configuration section must be a dictionary")
@@ -806,7 +819,21 @@ def _from_dict(cls, config_dict: dict[str, Any]) -> TopologyConfig:
             dc=dc_assignment,
         )
 
-        components = ComponentsConfig(assignments=assignments)
+        # New streamlined mappings
+        hw_component_map = components_dict.get("hw_component", {}) or {}
+        optics_map = components_dict.get("optics", {}) or {}
+        if hw_component_map is not None and not isinstance(hw_component_map, dict):
+            raise ValueError(
+                "'components.hw_component' must be a mapping when provided"
+            )
+        if optics_map is not None and not isinstance(optics_map, dict):
+            raise ValueError("'components.optics' must be a mapping when provided")
+
+        components = ComponentsConfig(
+            assignments=assignments,
+            hw_component=hw_component_map if isinstance(hw_component_map, dict) else {},
+            optics=optics_map if isinstance(optics_map, dict) else {},
+        )
 
         # Handle optional failure_policies configuration (assignments only)
         failure_policies_dict = config_dict.get("failure_policies", {})