Issue #24 - Use ELK layout to position SVG tasks + links

examples/convert_ewoks_to_elk_graph.py

@@ -3,9 +3,9 @@
 from ewokscore import load_graph
 from ewokscore.tests.examples.graphs import get_graph
 
-from ewoksdraw import build_svg_task_group
-from ewoksdraw.layout.elk_converter import ElkGraph
+from ewoksdraw.layout.elk_converter import ElkGraphBeforeLayout
 from ewoksdraw.layout.elk_converter import convert_ewoks_to_elk_graph
+from ewoksdraw.layout.ewoks_task_group_builder import build_svg_task_group
 from ewoksdraw.svg.svg_task_group import SvgTaskGroup
 from ewoksdraw.svg.svg_task_group import TaskInputPositions
 from ewoksdraw.svg.svg_task_group import TaskOutputPositions
@@ -18,7 +18,7 @@
 task_sizes: TaskSizes = svg_task_group.extract_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(
+elk_graph: ElkGraphBeforeLayout = convert_ewoks_to_elk_graph(
     ewoks_graph, task_sizes, task_input_positions, task_output_positions
 )
 

src/ewoksdraw/__init__.py

@@ -1,35 +1,3 @@
-from pathlib import Path
+from .bindings import graph_to_svg
 
-from ewokscore.graph import TaskGraph
-from ewokscore.graph.inputs import _get_all_node_inputs
-from ewokscore.graph.inputs import _get_all_task_output_names
-
-from .svg.svg_canvas import SvgCanvas
-from .svg.svg_task import SvgTask
-from .svg.svg_task_group import SvgTaskGroup
-
-GAP = 10.0
-DEFAULT_HEIGHT = 500
-
-
-def build_svg_task_group(graph: TaskGraph) -> SvgTaskGroup:
-    svg_tasks = {}
-    for node_id, node_attrs in graph.graph.nodes.items():
-        node_inputs = _get_all_node_inputs(node_id, node_attrs)
-        node_outputs = _get_all_task_output_names(
-            node_attrs["task_type"], node_attrs["task_identifier"]
-        )
-        svg_tasks[node_id] = SvgTask(
-            task_name=node_id,
-            input_names=[n.name for n in node_inputs],
-            output_names=node_outputs,
-        )
-    return SvgTaskGroup(svg_tasks, horizontal_gap=GAP, group_id=str(graph.graph_id))
-
-
-def graph_to_svg(graph: TaskGraph, output_path: str | Path) -> None:
-    task_group = build_svg_task_group(graph)
-    canvas = SvgCanvas(width=task_group.width, height=task_group.height + 2 * GAP)
-    canvas.add_background()
-    canvas.add_element(task_group)
-    canvas.draw(output_path)
+__all__ = ["graph_to_svg"]

src/ewoksdraw/bindings.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+from ewokscore.graph import TaskGraph
+from pyelk import ELK
+
+from .layout.elk_converter import ElkGraph
+from .layout.elk_converter import ElkGraphBeforeLayout
+from .layout.elk_converter import convert_ewoks_to_elk_graph
+from .layout.elk_converter import extract_task_positions_from_elk_graph
+from .layout.elk_link_group_builder import build_svg_link_group
+from .layout.ewoks_task_group_builder import build_svg_task_group
+from .svg.svg_canvas import SvgCanvas
+
+__all__ = ["graph_to_svg"]
+
+
+def graph_to_svg(graph: TaskGraph, output_path: str | Path) -> None:
+    task_group = build_svg_task_group(graph)
+    elk_graph_before_layout: ElkGraphBeforeLayout = convert_ewoks_to_elk_graph(
+        graph,
+        task_group.extract_task_sizes(),
+        task_group.extract_input_positions(),
+        task_group.extract_output_positions(),
+    )
+    elk_graph: ElkGraph = ELK().layout(elk_graph_before_layout)
+
+    task_positions = extract_task_positions_from_elk_graph(elk_graph)
+    task_group.set_task_positions(task_positions)
+
+    canvas = SvgCanvas(width=elk_graph["width"], height=elk_graph["height"])
+    canvas.add_background()
+    canvas.add_element(
+        build_svg_link_group(elk_graph, group_id=f"{graph.graph_id}-links")
+    )
+    canvas.add_element(task_group)
+    canvas.draw(output_path)

src/ewoksdraw/config/constants.py

@@ -10,6 +10,9 @@
 IO_ANCHOR_TEXT_MARGIN = 10
 IO_TOP_MARGIN = 5
 IO_INTER_IO_MARGIN = 3
+LINK_TURN_RADIUS = 5
+TASK_GROUP_HORIZONTAL_GAP = 10.0
+
 
 ELK_LAYOUT_OPTIONS = {
     "org.eclipse.elk.algorithm": "layered",

src/ewoksdraw/geometry/cubic_bezier_path.py

@@ -1,13 +1,19 @@
 import sys
 from dataclasses import dataclass
 from typing import Sequence
+from typing import TypedDict
 
 if sys.version_info < (3, 11):
     from typing_extensions import Self
 else:
     from typing import Self
 
-Point = tuple[float, float]
+
+class Point(TypedDict):
+    x: float
+    y: float
+
+
 Vector = tuple[float, float]
 
 
@@ -69,17 +75,17 @@ def from_points(cls, points: Sequence[Point], radius: float) -> Self:
 
 def _straight_segment(start: Point, end: Point) -> CubicBezierSegment:
     """Create a cubic Bezier segment that renders as a straight line."""
-    start_x, start_y = start
-    end_x, end_y = end
-
     # Control points are set to 1/2; 2/3 arbitrarly so they are not combine with
     # start and end points.
     return CubicBezierSegment(
-        control1=(start_x + (end_x - start_x) / 3, start_y + (end_y - start_y) / 3),
-        control2=(
-            start_x + 2 * (end_x - start_x) / 3,
-            start_y + 2 * (end_y - start_y) / 3,
-        ),
+        control1={
+            "x": start["x"] + (end["x"] - start["x"]) / 3,
+            "y": start["y"] + (end["y"] - start["y"]) / 3,
+        },
+        control2={
+            "x": start["x"] + 2 * (end["x"] - start["x"]) / 3,
+            "y": start["y"] + 2 * (end["y"] - start["y"]) / 3,
+        },
         end=end,
     )
 
@@ -89,15 +95,18 @@ def _direction(start: Point, end: Point) -> Vector:
     Return the horizontal or vertical direction from start to end.
     Example : (1, 0) right; (-1, 0) left ...
     """
-    if start[0] == end[0]:
-        return (0, 1 if end[1] > start[1] else -1)
-    return (1 if end[0] > start[0] else -1, 0)
+    if start["x"] == end["x"]:
+        return (0, 1 if end["y"] > start["y"] else -1)
+    return (1 if end["x"] > start["x"] else -1, 0)
 
 
 def _l1_distance(start: Point, end: Point) -> float:
-    return abs(end[0] - start[0]) + abs(end[1] - start[1])
+    return abs(end["x"] - start["x"]) + abs(end["y"] - start["y"])
 
 
 def _move(point: Point, direction: Vector, distance: float) -> Point:
     """Move a point along a direction by a distance."""
-    return (point[0] + direction[0] * distance, point[1] + direction[1] * distance)
+    return {
+        "x": point["x"] + direction[0] * distance,
+        "y": point["y"] + direction[1] * distance,
+    }

src/ewoksdraw/layout/elk_converter.py

@@ -6,9 +6,12 @@
 
 from ewoksdraw.config.constants import ELK_LAYOUT_OPTIONS
 
+from ..geometry.cubic_bezier_path import Point
 from ..svg.svg_task import TaskIOPosition
+from ..svg.svg_task import TaskPosition
 from ..svg.svg_task_group import TaskInputPositions
 from ..svg.svg_task_group import TaskOutputPositions
+from ..svg.svg_task_group import TaskPositions
 from ..svg.svg_task_group import TaskSizes
 
 
@@ -21,33 +24,86 @@ class ElkPort(TypedDict):
     layoutOptions: dict[str, Any]
 
 
-class ElkChild(TypedDict):
+class ElkChildBeforeLayout(TypedDict):
+    """An ELK child before layout."""
+
     id: str
     width: float
     height: float
     layoutOptions: dict[str, Any]
     ports: list[ElkPort]
 
 
-class ElkEdge(TypedDict):
+class ElkChild(ElkChildBeforeLayout):
+    """An ELK child with coordinates computed by ELK."""
+
+    x: float
+    y: float
+
+
+class ElkPoint(Point):
+    """A point in an ELK layout."""
+
+
+class ElkSection(TypedDict):
+    id: str
+    startPoint: ElkPoint
+    bendPoints: list[ElkPoint]
+    endPoint: ElkPoint
+    routing: str
+
+
+class ElkEdgeBeforeLayout(TypedDict):
+    """An ELK edge before layout."""
+
     id: str
     sources: list[str]
     targets: list[str]
 
 
-class ElkGraph(TypedDict):
+class ElkEdge(ElkEdgeBeforeLayout):
+    """An ELK edge with routing sections computed by ELK."""
+
+    sections: list[ElkSection]
+
+
+class ElkGraphBase(TypedDict):
     id: str
     layoutOptions: dict[str, Any]
+
+
+class ElkGraphBeforeLayout(ElkGraphBase):
+    """An ELK graph before layout."""
+
+    children: list[ElkChildBeforeLayout]
+    edges: list[ElkEdgeBeforeLayout]
+
+
+class ElkGraph(ElkGraphBase):
+    """An ELK graph with coordinates and routing computed by ELK."""
+
+    width: float
+    height: float
     children: list[ElkChild]
     edges: list[ElkEdge]
 
 
+def extract_task_positions_from_elk_graph(
+    elk_graph: ElkGraph,
+) -> TaskPositions:
+    """Extract SVG task positions from a laid-out ELK graph."""
+    return {
+        child["id"]: TaskPosition(name=child["id"], x=child["x"], y=child["y"])
+        for child in elk_graph["children"]
+    }
+
+
 def convert_ewoks_to_elk_graph(
     ewoks_graph: TaskGraph,
     task_sizes: TaskSizes,
     task_input_positions: TaskInputPositions,
     task_output_positions: TaskOutputPositions,
-) -> ElkGraph:
+) -> ElkGraphBeforeLayout:
     """Convert an Ewoks task graph into an ELK layout graph.
 
     :param ewoks_graph: the task graph to convert, e.g. from ``ewokscore.load_graph``.
@@ -76,7 +132,7 @@ def convert_ewoks_to_elk_graph(
             f"{sorted(node_ids)}"
         )
 
-    children: list[ElkChild] = []
+    children: list[ElkChildBeforeLayout] = []
     used_ids: set[str] = set()
     for task_id in ewoks_graph.graph.nodes:
         ports = _convert_io_positions_to_elk_ports(
@@ -101,7 +157,7 @@ def convert_ewoks_to_elk_graph(
     root_id = _available_elk_id("__ewoksdraw_root__", used_ids)
     used_ids.add(root_id)
 
-    edges: list[ElkEdge] = []
+    edges: list[ElkEdgeBeforeLayout] = []
     for source, target, link_attrs in ewoks_graph.graph.edges(data=True):
         if link_attrs.get("map_all_data", False):
             warnings.warn(

src/ewoksdraw/layout/elk_link_group_builder.py

@@ -0,0 +1,56 @@
+from ..config.constants import LINK_TURN_RADIUS
+from ..geometry.cubic_bezier_path import CubicBezierPath
+from ..geometry.cubic_bezier_path import Point
+from ..svg.svg_group import SvgGroup
+from ..svg.svg_link_cubic_bezier import SvgLinkCubicBezier
+from .elk_converter import ElkGraph
+from .elk_converter import ElkSection
+
+
+def build_svg_link_group(
+    elk_graph: ElkGraph, group_id: str | None = None
+) -> SvgGroup[SvgLinkCubicBezier]:
+    """Build an SVG link group from the routed edges of an ELK graph."""
+    link_group: SvgGroup[SvgLinkCubicBezier] = SvgGroup(group_id=group_id)
+    svg_links: list[SvgLinkCubicBezier] = []
+
+    for edge in elk_graph["edges"]:
+        for section in edge["sections"]:
+            points = _section_points(section)
+            cubic_bezier_path = CubicBezierPath.from_points(
+                points=points,
+                radius=LINK_TURN_RADIUS,
+            )
+            svg_link = SvgLinkCubicBezier(cubic_bezier_path)
+            svg_links.append(svg_link)
+
+    link_group.add_elements(svg_links)
+    return link_group
+
+
+def _section_points(section: ElkSection) -> list[Point]:
+    """Convert an ELK edge section into an ordered list of points.
+
+    :param section: an ELK edge section containing start, bend and end points.
+        For example::
+
+            {
+                "startPoint": {"x": 10.0, "y": 20.0},
+                "bendPoints": [{"x": 30.0, "y": 20.0}],
+                "endPoint": {"x": 30.0, "y": 40.0},
+            }
+
+    :return: the points ordered from start to end.
+        For example::
+
+            [
+                {"x": 10.0, "y": 20.0},
+                {"x": 30.0, "y": 20.0},
+                {"x": 30.0, "y": 40.0},
+            ]
+    """
+    return [
+        section["startPoint"],
+        *section["bendPoints"],
+        section["endPoint"],
+    ]

src/ewoksdraw/layout/ewoks_task_group_builder.py

@@ -0,0 +1,27 @@
+from ewokscore.graph import TaskGraph
+from ewokscore.graph.inputs import _get_all_node_inputs
+from ewokscore.graph.inputs import _get_all_task_output_names
+
+from ..config.constants import TASK_GROUP_HORIZONTAL_GAP
+from ..svg.svg_task import SvgTask
+from ..svg.svg_task_group import SvgTaskGroup
+
+
+def build_svg_task_group(graph: TaskGraph) -> SvgTaskGroup:
+    """Build an SVG task group from an Ewoks task graph."""
+    svg_tasks = {}
+    for node_id, node_attrs in graph.graph.nodes.items():
+        node_inputs = _get_all_node_inputs(node_id, node_attrs)
+        node_outputs = _get_all_task_output_names(
+            node_attrs["task_type"], node_attrs["task_identifier"]
+        )
+        svg_tasks[node_id] = SvgTask(
+            task_name=node_id,
+            input_names=[node_input.name for node_input in node_inputs],
+            output_names=node_outputs,
+        )
+    return SvgTaskGroup(
+        svg_tasks,
+        horizontal_gap=TASK_GROUP_HORIZONTAL_GAP,
+        group_id=str(graph.graph_id),
+    )