Refactor edge names

src/ewoksdraw/layout/elk_converter.py

@@ -19,6 +19,8 @@
 from ..svg.svg_task_group import TaskOutputPositions
 from ..svg.svg_task_group import TaskPositions
 from ..svg.svg_task_group import TaskSizes
+from ..utils import get_edge_source_id
+from ..utils import get_edge_target_id
 
 
 class ElkPort(TypedDict):
@@ -161,8 +163,8 @@ def convert_ewoks_to_elk_graph(
         edges.append(
             {
                 "id": edge_id,
-                "sources": [f"{mapping.source}.output.{mapping.source_output}"],
-                "targets": [f"{mapping.target}.input.{mapping.target_input}"],
+                "sources": [get_edge_source_id(mapping.source, mapping.source_output)],
+                "targets": [get_edge_target_id(mapping.target, mapping.target_input)],
             }
         )
         used_ids.add(edge_id)
@@ -179,37 +181,35 @@ def _convert_io_positions_to_elk_ports(
     task_id: str,
     input_positions: list[TaskIOPosition],
     output_positions: list[TaskIOPosition],
-) -> list[ElkPort]:
-    ports = _convert_positions_to_elk_ports(
-        input_positions,
-        id_prefix=f"{task_id}.input",
-        elk_port_side="WEST",
-    )
-    ports.extend(
-        _convert_positions_to_elk_ports(
-            output_positions,
-            id_prefix=f"{task_id}.output",
-            elk_port_side="EAST",
-        )
-    )
-    return ports
-
-
-def _convert_positions_to_elk_ports(
-    positions: list[TaskIOPosition], id_prefix: str, elk_port_side: str
 ) -> list[ElkPort]:
     ports: list[ElkPort] = []
 
-    for index, position in enumerate(positions):
+    for index, position in enumerate(input_positions):
+        ports.append(
+            {
+                "id": get_edge_target_id(task_id, position.name),
+                "x": position.x,
+                "y": position.y,
+                "width": 0,
+                "height": 0,
+                "layoutOptions": {
+                    "org.eclipse.elk.port.side": "WEST",
+                    "org.eclipse.elk.port.index": index,
+                    "org.eclipse.elk.port.borderOffset": 0,
+                },
+            }
+        )
+
+    for index, position in enumerate(output_positions):
         ports.append(
             {
-                "id": f"{id_prefix}.{position.name}",
+                "id": get_edge_source_id(task_id, position.name),
                 "x": position.x,
                 "y": position.y,
                 "width": 0,
                 "height": 0,
                 "layoutOptions": {
-                    "org.eclipse.elk.port.side": elk_port_side,
+                    "org.eclipse.elk.port.side": "EAST",
                     "org.eclipse.elk.port.index": index,
                     "org.eclipse.elk.port.borderOffset": 0,
                 },

src/ewoksdraw/layout/elk_link_group_builder.py

@@ -3,6 +3,8 @@
 from ..geometry.cubic_bezier_path import Point
 from ..svg.svg_group import SvgGroup
 from ..svg.svg_link_cubic_bezier import SvgLinkCubicBezier
+from ..utils import get_task_id_from_source_id
+from ..utils import get_task_id_from_target_id
 from .elk_converter import ElkEdge
 from .elk_converter import ElkGraph
 from .elk_converter import ElkSection
@@ -37,9 +39,8 @@ def _edge_has_import_error(
 ) -> bool:
     if task_import_errors is None:
         return False
-    # FIXME:
-    source_task_id = edge["sources"][0].split(".output.")[0]
-    target_task_id = edge["targets"][0].split(".input.")[0]
+    source_task_id = get_task_id_from_source_id(edge["sources"][0])
+    target_task_id = get_task_id_from_target_id(edge["targets"][0])
     return task_import_errors.get(source_task_id, False) or task_import_errors.get(
         target_task_id, False
     )

src/ewoksdraw/tests/test_utils.py

@@ -0,0 +1,20 @@
+from ..utils import get_edge_source_id
+from ..utils import get_edge_target_id
+from ..utils import get_task_id_from_source_id
+from ..utils import get_task_id_from_target_id
+
+
+def test_edge_source_task_id_roundtrip():
+    task_id = "pipeline.output.normalize"
+    output_name = "value"
+
+    source_id = get_edge_source_id(task_id, output_name)
+    assert get_task_id_from_source_id(source_id) == task_id
+
+
+def test_edge_target_task_id_roundtrip():
+    task_id = "pipeline.input.normalize"
+    output_name = "value"
+
+    target_id = get_edge_target_id(task_id, output_name)
+    assert get_task_id_from_target_id(target_id) == task_id

src/ewoksdraw/utils.py

@@ -51,3 +51,19 @@ def get_edge_sources_and_targets(
         targets[mapping.target].append(mapping.target_input)
 
     return sources, targets
+
+
+def get_edge_source_id(task_id: str, name: str) -> str:
+    return f"{task_id}.output.{name}"
+
+
+def get_task_id_from_source_id(source_id: str) -> str:
+    return source_id.rsplit(".output.", 1)[0]
+
+
+def get_edge_target_id(task_id: str, name: str) -> str:
+    return f"{task_id}.input.{name}"
+
+
+def get_task_id_from_target_id(target_id: str) -> str:
+    return target_id.rsplit(".input.", 1)[0]