feat: add port name shortening functionality to connection resolver and emitter

Signed-off-by: Taekjin LEE <taekjin.lee@tier4.jp>

Author: technolojin

tools/system-config-generator/pipeline/connection_resolver.py

@@ -7,6 +7,7 @@
 
 from .launch_parser import RemapEntry  # noqa: F401 — used by callers via star-import
 from .namespace_tree import NamespaceNode
+from .port_utils import shorten_port_names
 
 # Topics that carry no domain-level information
 _INFRA_EXACT = {
@@ -103,6 +104,25 @@ def resolve_connections(top_nodes: list[NamespaceNode]) -> list[TopicConnection]
                     ref = PortRef(component=ns_node.name, port=canonical, node_path=node.full_path)
                     subscribers[topic].append(ref)
 
+    # Build per-component shortening maps from cross-component ports only.
+    # Internal-only ports are excluded so they can't cause false collisions.
+    comp_pub_raw: dict[str, set[str]] = defaultdict(set)
+    comp_sub_raw: dict[str, set[str]] = defaultdict(set)
+    for topic, pubs in publishers.items():
+        subs_for_topic = subscribers.get(topic, [])
+        for pub in pubs:
+            for sub in subs_for_topic:
+                if pub.component != sub.component:
+                    comp_pub_raw[pub.component].add(pub.port)
+                    comp_sub_raw[sub.component].add(sub.port)
+
+    comp_pub_short: dict[str, dict[str, str]] = {
+        c: shorten_port_names(list(ports)) for c, ports in comp_pub_raw.items()
+    }
+    comp_sub_short: dict[str, dict[str, str]] = {
+        c: shorten_port_names(list(ports)) for c, ports in comp_sub_raw.items()
+    }
+
     connections: list[TopicConnection] = []
     seen: set[tuple] = set()
 
@@ -112,11 +132,19 @@ def resolve_connections(top_nodes: list[NamespaceNode]) -> list[TopicConnection]
             for sub in subs:
                 if pub.component == sub.component:
                     continue  # internal – handled in module.yaml
-                key = (pub.component, pub.port, sub.component, sub.port)
+                pub_port = comp_pub_short.get(pub.component, {}).get(pub.port, pub.port)
+                sub_port = comp_sub_short.get(sub.component, {}).get(sub.port, sub.port)
+                key = (pub.component, pub_port, sub.component, sub_port)
                 if key in seen:
                     continue
                 seen.add(key)
-                connections.append(TopicConnection(publisher=pub, subscriber=sub, topic=topic))
+                connections.append(
+                    TopicConnection(
+                        publisher=PortRef(pub.component, pub_port, pub.node_path),
+                        subscriber=PortRef(sub.component, sub_port, sub.node_path),
+                        topic=topic,
+                    )
+                )
 
     return connections
 

tools/system-config-generator/pipeline/emitter/module.py

@@ -12,6 +12,7 @@
 )
 from ..launch_parser import NodeRecord
 from ..namespace_tree import NamespaceNode
+from ..port_utils import shorten_port_names
 
 DESIGN_FORMAT = "0.3.1"
 
@@ -162,6 +163,14 @@ def _extract_ns_module_interface(
                     )
                 )
 
+    if external_pub:
+        pub_short = shorten_port_names([p for p, _ in external_pub])
+        external_pub = [(pub_short[p], t) for p, t in external_pub]
+
+    if external_sub:
+        sub_short = shorten_port_names([p for p, _ in external_sub])
+        external_sub = [(sub_short[p], t) for p, t in external_sub]
+
     return ModuleInterface(
         publishers=external_pub,
         subscribers=external_sub,

tools/system-config-generator/pipeline/port_utils.py

@@ -0,0 +1,27 @@
+"""Port name utilities shared across emitters and connection resolver."""
+
+from __future__ import annotations
+
+
+def shorten_port_names(names: list[str]) -> dict[str, str]:
+    """Return {original: shortest_unique_suffix} split by '/' segments.
+
+    Each name is shortened to the minimum number of trailing slash-separated
+    segments that uniquely identifies it within the list.
+    """
+    segments_map = {name: name.split("/") for name in names}
+    result: dict[str, str] = {}
+    for name in names:
+        segs = segments_map[name]
+        for k in range(1, len(segs) + 1):
+            suffix = "/".join(segs[-k:])
+            if not any(
+                suffix == "/".join(other_segs[-min(k, len(other_segs)) :])
+                for other, other_segs in segments_map.items()
+                if other != name
+            ):
+                result[name] = suffix
+                break
+        else:
+            result[name] = name
+    return result