Save port positions in the elk graph

CHANGELOG.md

@@ -8,5 +8,3 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
-
--  `graph_to_svg`: Generates SVG mock-ups from Ewoks workflows.

examples/convert_ewoks_to_elk_graph.py

@@ -7,16 +7,24 @@
 from ewoksdraw.layout.elk_converter import ElkGraph
 from ewoksdraw.layout.elk_converter import convert_ewoks_to_elk_graph
 from ewoksdraw.svg.svg_task_group import SvgTaskGroup
+from ewoksdraw.svg.svg_task_group import TaskInputPositions
+from ewoksdraw.svg.svg_task_group import TaskOutputPositions
 from ewoksdraw.svg.svg_task_group import TaskSizes
 
 graph_description, _ = get_graph("acyclic1")
 ewoks_graph = load_graph(graph_description)
 
 svg_task_group: SvgTaskGroup = build_svg_task_group(ewoks_graph)
 task_sizes: TaskSizes = svg_task_group.extract_task_sizes()
-elk_graph: ElkGraph = convert_ewoks_to_elk_graph(ewoks_graph, task_sizes)
+task_input_positions: TaskInputPositions = svg_task_group.extract_input_positions()
+task_output_positions: TaskOutputPositions = svg_task_group.extract_output_positions()
+elk_graph: ElkGraph = convert_ewoks_to_elk_graph(
+    ewoks_graph, task_sizes, task_input_positions, task_output_positions
+)
 
 pprint(dict(ewoks_graph.graph.nodes(data=True)))
 pprint(list(ewoks_graph.graph.edges(data=True)))
 pprint(task_sizes)
+pprint(task_input_positions)
+pprint(task_output_positions)
 pprint(elk_graph)

src/ewoksdraw/layout/elk_converter.py

@@ -1,17 +1,32 @@
+import warnings
 from typing import Any
 from typing import TypedDict
 
 from ewokscore.graph import TaskGraph
 
 from ewoksdraw.config.constants import ELK_LAYOUT_OPTIONS
 
+from ..svg.svg_task import TaskIOPosition
+from ..svg.svg_task_group import TaskInputPositions
+from ..svg.svg_task_group import TaskOutputPositions
 from ..svg.svg_task_group import TaskSizes
 
 
+class ElkPort(TypedDict):
+    id: str
+    x: float
+    y: float
+    width: float
+    height: float
+    layoutOptions: dict[str, Any]
+
+
 class ElkChild(TypedDict):
     id: str
     width: float
     height: float
+    layoutOptions: dict[str, Any]
+    ports: list[ElkPort]
 
 
 class ElkEdge(TypedDict):
@@ -28,13 +43,17 @@ class ElkGraph(TypedDict):
 
 
 def convert_ewoks_to_elk_graph(
-    ewoks_graph: TaskGraph, task_sizes: TaskSizes
+    ewoks_graph: TaskGraph,
+    task_sizes: TaskSizes,
+    task_input_positions: TaskInputPositions,
+    task_output_positions: TaskOutputPositions,
 ) -> ElkGraph:
     """Convert an Ewoks task graph into an ELK layout graph.
 
     :param ewoks_graph: the task graph to convert, e.g. from ``ewokscore.load_graph``.
-    :param task_sizes: ``(width, height)`` per task in ``ewoks_graph``, and no
-        other task id, e.g. ``{"task1": (39.56, 55.0), ...}``.
+    :param task_sizes: width and height of each task.
+    :param task_input_positions: input positions of each task.
+    :param task_output_positions: output positions of each task.
     """
     node_ids = set(ewoks_graph.graph.nodes)
     if node_ids != task_sizes.keys():
@@ -43,30 +62,135 @@ def convert_ewoks_to_elk_graph(
             f"{sorted(node_ids)}"
         )
 
-    children: list[ElkChild] = [
-        {
-            "id": task_id,
-            "width": task_sizes[task_id].width,
-            "height": task_sizes[task_id].height,
-        }
-        for task_id in ewoks_graph.graph.nodes
-    ]
+    if node_ids != task_input_positions.keys():
+        raise ValueError(
+            "task_input_positions "
+            f"{sorted(task_input_positions)} do not match ewoks_graph task ids "
+            f"{sorted(node_ids)}"
+        )
+
+    if node_ids != task_output_positions.keys():
+        raise ValueError(
+            "task_output_positions "
+            f"{sorted(task_output_positions)} do not match ewoks_graph task ids "
+            f"{sorted(node_ids)}"
+        )
+
+    children: list[ElkChild] = []
+    used_ids: set[str] = set()
+    for task_id in ewoks_graph.graph.nodes:
+        ports = _convert_io_positions_to_elk_ports(
+            task_id,
+            task_input_positions[task_id],
+            task_output_positions[task_id],
+        )
+        children.append(
+            {
+                "id": task_id,
+                "width": task_sizes[task_id].width,
+                "height": task_sizes[task_id].height,
+                "layoutOptions": {
+                    "org.eclipse.elk.portConstraints": "FIXED_POS",
+                },
+                "ports": ports,
+            }
+        )
+        used_ids.add(task_id)
+        used_ids.update(port["id"] for port in ports)
+
+    root_id = _available_elk_id("__ewoksdraw_root__", used_ids)
+    used_ids.add(root_id)
 
     edges: list[ElkEdge] = []
     for source, target, link_attrs in ewoks_graph.graph.edges(data=True):
-        n_mappings = len(link_attrs.get("data_mapping", [])) or 1
-        for index in range(n_mappings):
+        if link_attrs.get("map_all_data", False):
+            warnings.warn(
+                f"Ewoks link {source!r} -> {target!r} uses 'map_all_data', which "
+                "is not yet supported.",
+                UserWarning,
+                stacklevel=2,
+            )
+
+        for mapping in link_attrs.get("data_mapping", []):
+            source_output = mapping.get("source_output")
+            if source_output is None:
+                warnings.warn(
+                    f"Data mapping on Ewoks link {source!r} -> {target!r} has no "
+                    "'source_output', which is not yet supported.",
+                    UserWarning,
+                    stacklevel=2,
+                )
+                continue
+
+            edge_id = _available_elk_id(
+                f"edge_{len(edges)}_{source}_{target}", used_ids
+            )
+
             edges.append(
                 {
-                    "id": f"edge_{source}_{target}_{index}",
-                    "sources": [source],
-                    "targets": [target],
+                    "id": edge_id,
+                    "sources": [f"{source}.output.{source_output}"],
+                    "targets": [f"{target}.input.{mapping['target_input']}"],
                 }
             )
+            used_ids.add(edge_id)
 
     return {
-        "id": "root",
+        "id": root_id,
         "layoutOptions": ELK_LAYOUT_OPTIONS,
         "children": children,
         "edges": edges,
     }
+
+
+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):
+        ports.append(
+            {
+                "id": f"{id_prefix}.{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.index": index,
+                    "org.eclipse.elk.port.borderOffset": 0,
+                },
+            }
+        )
+
+    return ports
+
+
+def _available_elk_id(preferred_id: str, used_ids: set[str]) -> str:
+    element_id = preferred_id
+    index = 1
+    while element_id in used_ids:
+        element_id = f"{preferred_id}_{index}"
+        index += 1
+    return element_id

src/ewoksdraw/svg/svg_group.py

@@ -1,6 +1,7 @@
 import re
 from typing import Generic
 from typing import Iterable
+from typing import NamedTuple
 from typing import Protocol
 from typing import TypeVar
 from xml.etree.ElementTree import Element
@@ -14,6 +15,11 @@ def xml_element(self) -> Element: ...
 SvgElementType = TypeVar("SvgElementType", bound=SvgElementLike)
 
 
+class Translation(NamedTuple):
+    x: float
+    y: float
+
+
 class SvgGroup(Generic[SvgElementType]):
     """
     Represents a group of SVG elements.
@@ -27,6 +33,12 @@ def __init__(self, group_id: str | None = None):
         self.elements: list[SvgElementType] = []
         self._group_id = group_id
         self._transform = ""
+        self._translation = Translation(x=0.0, y=0.0)
+
+    @property
+    def translation(self) -> Translation:
+        """Returns the current translation as an ``(x, y)`` tuple."""
+        return self._translation
 
     def add_elements(self, elements: Iterable[SvgElementType]) -> None:
         """
@@ -49,6 +61,8 @@ def translate(self, x: float = 0, y: float = 0) -> None:
         else:
             self._transform = new_transform
         self._transform = self._transform.strip()
+        current_x, current_y = self._translation
+        self._translation = Translation(x=current_x + x, y=current_y + y)
 
     def set_translation(self, x: float = 0, y: float = 0) -> None:
         """
@@ -66,6 +80,7 @@ def set_translation(self, x: float = 0, y: float = 0) -> None:
             self._transform = f"{cleaned_transform} {new_translate}".strip()
         else:
             self._transform = new_translate
+        self._translation = Translation(x=x, y=y)
 
     @property
     def xml_element(self) -> Element:

src/ewoksdraw/svg/svg_task.py

@@ -14,6 +14,12 @@ class TaskSize(NamedTuple):
     height: float
 
 
+class TaskIOPosition(NamedTuple):
+    name: str
+    x: float
+    y: float
+
+
 class SvgTask(SvgGroup):
     """
     Represents a task as an SVG group containing title, input/output groups, box, and
@@ -35,6 +41,7 @@ def __init__(
     ):
         super().__init__(group_id=task_name)
 
+        self._task_name = task_name
         self._interspace_title_input = IO_TOP_MARGIN
         self._interspace_input_output = IO_INTER_IO_MARGIN
         self._title = SvgTaskTitle(text=task_name, x=0, y=0)
@@ -131,3 +138,20 @@ def height(self) -> float:
             + self._interspace_input_output
             + self._interspace_title_input
         )
+
+    def get_input_positions(self) -> list[TaskIOPosition]:
+        """Return the task-relative positions of the task inputs."""
+        return self._get_group_io_positions(self._inputs)
+
+    def get_output_positions(self) -> list[TaskIOPosition]:
+        """Return the task-relative positions of the task outputs."""
+        return self._get_group_io_positions(self._outputs)
+
+    @staticmethod
+    def _get_group_io_positions(group: SvgTaskIOGroup) -> list[TaskIOPosition]:
+        positions: list[TaskIOPosition] = []
+        for io in group.elements:
+            x = group.translation.x + io.translation.x
+            y = group.translation.y + io.translation.y
+            positions.append(TaskIOPosition(name=io.name, x=x, y=y))
+        return positions

src/ewoksdraw/svg/svg_task_group.py

@@ -1,8 +1,11 @@
 from .svg_group import SvgGroup
 from .svg_task import SvgTask
+from .svg_task import TaskIOPosition
 from .svg_task import TaskSize
 
 TaskSizes = dict[str, TaskSize]
+TaskInputPositions = dict[str, list[TaskIOPosition]]
+TaskOutputPositions = dict[str, list[TaskIOPosition]]
 
 
 class SvgTaskGroup(SvgGroup[SvgTask]):
@@ -53,3 +56,15 @@ def extract_task_sizes(self) -> TaskSizes:
             task_id: TaskSize(width=svg_task.width, height=svg_task.height)
             for task_id, svg_task in self._svg_tasks.items()
         }
+
+    def extract_input_positions(self) -> TaskInputPositions:
+        return {
+            task_id: svg_task.get_input_positions()
+            for task_id, svg_task in self._svg_tasks.items()
+        }
+
+    def extract_output_positions(self) -> TaskOutputPositions:
+        return {
+            task_id: svg_task.get_output_positions()
+            for task_id, svg_task in self._svg_tasks.items()
+        }

src/ewoksdraw/svg/svg_task_io.py

@@ -36,6 +36,15 @@ def set_font_size(self, font_size: float) -> None:
         """
         self.txt.set_font_size(font_size)
 
+    @property
+    def name(self) -> str:
+        """Return the original, untruncated task input or output name."""
+        return self._io_txt
+
+    @property
+    def io_type(self) -> Literal["input", "output"]:
+        return self._io_type
+
     @property
     def font_size(self) -> float:
         """