feat(calc,routing): config, orthogonal routing, and mock flow solver

Add observable AppConfig (grid/snap/arc-on-crossing/animation/theme),
pure-geometry orthogonal router with grid-snapped bends and perpendicular
crossing detection for arc hops, and a mock flow-distribution solver that
pushes supply to demand along shortest paths and records signed per-edge
flow and direction. Covered by routing and flow-solver tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

pipeline_editor/calc/flow_solver.py

@@ -0,0 +1,160 @@
+"""Mock flow-distribution 'calculation'.
+
+This is a deliberately simplified stand-in for a real hydraulic/electrical
+solver. It distributes flow from source nodes to sink nodes across the network
+so that each edge receives a plausible signed flow value and direction, which
+the animation layer then visualizes.
+
+Approach (mock, not physically rigorous):
+  1. Identify sources (supply/feed) and sinks (demand) from node properties.
+  2. Build an undirected graph of nodes connected by edges.
+  3. For each (source, sink) pair with a path, push the sink's demand (scaled by
+     available supply) along the shortest path, accumulating signed flow per edge.
+  4. Edge direction is the sign of the accumulated flow along source->target.
+
+The result is stored on each :class:`EdgeModel` as ``flow`` (magnitude) and
+``flow_direction`` (+1/-1/0).
+"""
+from __future__ import annotations
+
+from collections import deque
+from typing import Optional
+
+
+class FlowResult:
+    """Per-edge flow outcome plus network totals."""
+
+    def __init__(self) -> None:
+        self.edge_flow: dict[str, float] = {}       # signed, along source->target
+        self.total_supply: float = 0.0
+        self.total_demand: float = 0.0
+        self.unbalanced: float = 0.0                # supply - served demand
+
+    def direction(self, edge_id: str) -> int:
+        f = self.edge_flow.get(edge_id, 0.0)
+        if f > 1e-9:
+            return 1
+        if f < -1e-9:
+            return -1
+        return 0
+
+
+def _node_supply(node) -> float:
+    return float(node.properties.value("supply", 0.0) or 0.0)
+
+
+def _node_demand(node) -> float:
+    return float(node.properties.value("demand", 0.0) or 0.0)
+
+
+def solve(document) -> FlowResult:
+    """Distribute flow across the document's edges. Returns a :class:`FlowResult`."""
+    result = FlowResult()
+    nodes = list(document.nodes())
+    edges = list(document.edges())
+    if not edges:
+        return result
+
+    # adjacency: node_id -> list of (neighbor_id, edge_id, sign_for_source_to_target)
+    adj: dict[str, list[tuple[str, str, int]]] = {n.node_id: [] for n in nodes}
+    for e in edges:
+        if e.source_node in adj and e.target_node in adj:
+            adj[e.source_node].append((e.target_node, e.edge_id, +1))
+            adj[e.target_node].append((e.source_node, e.edge_id, -1))
+        result.edge_flow[e.edge_id] = 0.0
+
+    sources = [(n.node_id, _node_supply(n)) for n in nodes if _node_supply(n) > 0]
+    sinks = [(n.node_id, _node_demand(n)) for n in nodes if _node_demand(n) > 0]
+    result.total_supply = sum(s for _, s in sources)
+    result.total_demand = sum(d for _, d in sinks)
+
+    if not sources or not sinks:
+        # No defined supply/demand: assign a nominal unit flow so the diagram
+        # still animates. Direction follows edge source->target.
+        for e in edges:
+            result.edge_flow[e.edge_id] = 1.0
+        return result
+
+    # Scale served demand so we never exceed available supply.
+    scale = 1.0
+    if result.total_demand > result.total_supply and result.total_demand > 0:
+        scale = result.total_supply / result.total_demand
+    result.unbalanced = result.total_supply - result.total_demand * scale
+
+    remaining_supply = {sid: s for sid, s in sources}
+    for sink_id, demand in sinks:
+        served = demand * scale
+        # distribute this sink's served demand across reachable sources
+        while served > 1e-9:
+            src = _nearest_source_with_supply(sink_id, remaining_supply, adj)
+            if src is None:
+                break
+            path = _shortest_path(src, sink_id, adj)
+            if not path:
+                remaining_supply[src] = 0.0
+                continue
+            push = min(served, remaining_supply[src])
+            _accumulate_along_path(path, adj, push, result)
+            remaining_supply[src] -= push
+            served -= push
+
+    return result
+
+
+def _nearest_source_with_supply(sink_id, remaining_supply, adj) -> Optional[str]:
+    """BFS from sink to the closest source that still has supply."""
+    seen = {sink_id}
+    q = deque([sink_id])
+    while q:
+        cur = q.popleft()
+        if cur in remaining_supply and remaining_supply[cur] > 1e-9:
+            return cur
+        for nb, _eid, _sign in adj.get(cur, []):
+            if nb not in seen:
+                seen.add(nb)
+                q.append(nb)
+    return None
+
+
+def _shortest_path(start, goal, adj) -> list[tuple[str, str, int]]:
+    """BFS path as a list of (from_node, edge_id, sign) hops from start to goal."""
+    if start == goal:
+        return []
+    prev: dict[str, tuple[str, str, int]] = {}
+    seen = {start}
+    q = deque([start])
+    while q:
+        cur = q.popleft()
+        for nb, eid, sign in adj.get(cur, []):
+            if nb not in seen:
+                seen.add(nb)
+                prev[nb] = (cur, eid, sign)
+                if nb == goal:
+                    q.clear()
+                    break
+                q.append(nb)
+    if goal not in prev:
+        return []
+    hops: list[tuple[str, str, int]] = []
+    node = goal
+    while node != start:
+        frm, eid, sign = prev[node]
+        hops.append((frm, eid, sign))
+        node = frm
+    hops.reverse()
+    return hops
+
+
+def _accumulate_along_path(path, adj, amount, result) -> None:
+    for _frm, eid, sign in path:
+        result.edge_flow[eid] += sign * amount
+
+
+def apply_to_document(document) -> FlowResult:
+    """Solve and write ``flow`` / ``flow_direction`` back onto each edge."""
+    result = solve(document)
+    for e in document.edges():
+        f = result.edge_flow.get(e.edge_id, 0.0)
+        e.flow = abs(f)
+        e.flow_direction = result.direction(e.edge_id)
+    return result

pipeline_editor/config.py

@@ -0,0 +1,63 @@
+"""Application configuration (grid, snapping, routing, animation, theming)."""
+from __future__ import annotations
+
+from PyQt6.QtCore import QObject, pyqtSignal
+
+
+class AppConfig(QObject):
+    """Live, observable editor configuration.
+
+    Emitting :attr:`changed` lets the canvas and items refresh when the user
+    tweaks a setting in the configuration dialog.
+    """
+
+    changed = pyqtSignal()
+
+    def __init__(self, parent: QObject | None = None) -> None:
+        super().__init__(parent)
+        self.grid_size = 20
+        self.show_grid = True
+        self.snap_to_grid = True
+        # Draw a small arc "hop" where an edge crosses another edge.
+        self.arc_on_crossing = True
+        self.arc_radius = 5.0
+        # Routing stub length from a port before turning.
+        self.route_stub = 20
+        # Flow animation.
+        self.animation_speed = 1.0  # multiplier
+        self.color_scheme = "default"
+        # Canvas theme.
+        self.background = "#fafafa"
+        self.grid_color = "#e0e0e0"
+
+    def snap(self, value: float) -> float:
+        if not self.snap_to_grid or self.grid_size <= 0:
+            return value
+        g = self.grid_size
+        return round(value / g) * g
+
+    def snap_point(self, x: float, y: float) -> tuple[float, float]:
+        return self.snap(x), self.snap(y)
+
+    def emit_changed(self) -> None:
+        self.changed.emit()
+
+    def to_dict(self) -> dict:
+        return {
+            "grid_size": self.grid_size,
+            "show_grid": self.show_grid,
+            "snap_to_grid": self.snap_to_grid,
+            "arc_on_crossing": self.arc_on_crossing,
+            "arc_radius": self.arc_radius,
+            "route_stub": self.route_stub,
+            "animation_speed": self.animation_speed,
+            "color_scheme": self.color_scheme,
+            "background": self.background,
+            "grid_color": self.grid_color,
+        }
+
+    def update_from(self, data: dict) -> None:
+        for k, v in data.items():
+            if hasattr(self, k):
+                setattr(self, k, v)
+        self.changed.emit()

pipeline_editor/view/routing.py

@@ -0,0 +1,125 @@
+"""Pure-geometry orthogonal routing and crossing detection.
+
+Kept free of Qt imports so it can be unit-tested independently. Points are
+``(x, y)`` float tuples; segments are ``(p0, p1)`` axis-aligned pairs.
+"""
+from __future__ import annotations
+
+from typing import Iterable
+
+Point = tuple[float, float]
+Segment = tuple[Point, Point]
+
+
+def _snap(v: float, grid: int) -> float:
+    if grid <= 0:
+        return v
+    return round(v / grid) * grid
+
+
+def orthogonal_route(
+    start: Point,
+    start_normal: Point,
+    end: Point,
+    end_normal: Point,
+    *,
+    grid: int = 20,
+    stub: int | None = None,
+) -> list[Point]:
+    """Compute an axis-aligned poly-line from ``start`` to ``end``.
+
+    The path leaves ``start`` along ``start_normal`` and arrives at ``end``
+    against ``end_normal``, turning through a mid-line. Endpoints are exact;
+    intermediate bends are snapped to ``grid``.
+    """
+    if stub is None:
+        stub = grid
+    sx, sy = start
+    ex, ey = end
+    snx, sny = start_normal
+    enx, eny = end_normal
+
+    s1 = (sx + snx * stub, sy + sny * stub)
+    e1 = (ex + enx * stub, ey + eny * stub)
+
+    pts: list[Point] = [(sx, sy), s1]
+    if snx != 0:  # start exits horizontally -> pivot on a vertical mid-line
+        midx = _snap((s1[0] + e1[0]) / 2.0, grid)
+        pts += [(midx, s1[1]), (midx, e1[1])]
+    else:  # start exits vertically -> pivot on a horizontal mid-line
+        midy = _snap((s1[1] + e1[1]) / 2.0, grid)
+        pts += [(s1[0], midy), (e1[0], midy)]
+    pts += [e1, (ex, ey)]
+
+    return _cleanup(pts)
+
+
+def _cleanup(pts: list[Point], eps: float = 1e-6) -> list[Point]:
+    """Drop duplicate and collinear intermediate points."""
+    out: list[Point] = []
+    for p in pts:
+        if out and abs(p[0] - out[-1][0]) < eps and abs(p[1] - out[-1][1]) < eps:
+            continue
+        out.append(p)
+    # remove collinear middles
+    cleaned: list[Point] = []
+    for i, p in enumerate(out):
+        if 0 < i < len(out) - 1:
+            a, b = out[i - 1], out[i + 1]
+            # collinear if all three share an x or all share a y
+            if (abs(a[0] - p[0]) < eps and abs(p[0] - b[0]) < eps) or \
+               (abs(a[1] - p[1]) < eps and abs(p[1] - b[1]) < eps):
+                continue
+        cleaned.append(p)
+    return cleaned
+
+
+def polyline_segments(points: Iterable[Point]) -> list[Segment]:
+    pts = list(points)
+    return [(pts[i], pts[i + 1]) for i in range(len(pts) - 1)]
+
+
+def _is_horizontal(seg: Segment, eps: float = 1e-6) -> bool:
+    return abs(seg[0][1] - seg[1][1]) < eps
+
+
+def _is_vertical(seg: Segment, eps: float = 1e-6) -> bool:
+    return abs(seg[0][0] - seg[1][0]) < eps
+
+
+def segment_crossings(
+    path: list[Segment],
+    others: list[Segment],
+    *,
+    eps: float = 1e-6,
+) -> list[Point]:
+    """Return true perpendicular crossing points between ``path`` and ``others``.
+
+    Only counts a horizontal segment crossing a vertical one (or vice versa)
+    strictly in the interior of both — shared endpoints and overlaps are ignored.
+    Used to render arc "hops" where pipes cross.
+    """
+    crossings: list[Point] = []
+    for a in path:
+        for b in others:
+            pt = _perpendicular_crossing(a, b, eps)
+            if pt is not None:
+                crossings.append(pt)
+    return crossings
+
+
+def _perpendicular_crossing(a: Segment, b: Segment, eps: float) -> Point | None:
+    if _is_horizontal(a) and _is_vertical(b):
+        h, v = a, b
+    elif _is_vertical(a) and _is_horizontal(b):
+        v, h = a, b
+    else:
+        return None
+    hy = h[0][1]
+    vx = v[0][0]
+    hx0, hx1 = sorted((h[0][0], h[1][0]))
+    vy0, vy1 = sorted((v[0][1], v[1][1]))
+    # strictly interior on both segments (avoid endpoints / T-joins)
+    if hx0 + eps < vx < hx1 - eps and vy0 + eps < hy < vy1 - eps:
+        return (vx, hy)
+    return None

tests/test_flow_solver.py

@@ -0,0 +1,74 @@
+"""Tests for the mock flow solver."""
+import pytest
+
+from pipeline_editor.calc import flow_solver
+from pipeline_editor.model.document import DiagramDocument
+from pipeline_editor.model.edge import EdgeModel
+from pipeline_editor.model.node_library import default_library
+
+
+@pytest.fixture
+def lib():
+    return default_library()
+
+
+def _add(doc, lib, key, x=0, y=0):
+    return doc.add_node(lib.get(key).instantiate(doc.next_id("n"), x, y))
+
+
+def test_linear_network_flow_direction(qapp, lib):
+    doc = DiagramDocument()
+    src = _add(doc, lib, "source")          # supply 120
+    pump = _add(doc, lib, "pump", 200)
+    cons = _add(doc, lib, "consumer", 400)  # demand 40
+    e1 = doc.add_edge(EdgeModel(doc.next_id("e"), src.node_id, "out", pump.node_id, "in"))
+    e2 = doc.add_edge(EdgeModel(doc.next_id("e"), pump.node_id, "out", cons.node_id, "in"))
+
+    result = flow_solver.apply_to_document(doc)
+    # 40 units flow from source to consumer along both edges, in source->target dir
+    assert result.edge_flow[e1.edge_id] == pytest.approx(40)
+    assert result.edge_flow[e2.edge_id] == pytest.approx(40)
+    assert e1.flow_direction == 1 and e2.flow_direction == 1
+    assert e1.flow == pytest.approx(40)
+
+
+def test_reversed_edge_gives_negative_direction(qapp, lib):
+    doc = DiagramDocument()
+    src = _add(doc, lib, "source")
+    cons = _add(doc, lib, "consumer", 400)
+    # edge authored consumer(source-of-edge) -> source(target-of-edge)
+    e = doc.add_edge(EdgeModel(doc.next_id("e"), cons.node_id, "in", src.node_id, "out"))
+    flow_solver.apply_to_document(doc)
+    # physical flow is source->consumer, i.e. against edge authoring direction
+    assert e.flow_direction == -1
+    assert e.flow == pytest.approx(40)
+
+
+def test_demand_capped_by_supply(qapp, lib):
+    doc = DiagramDocument()
+    src = _add(doc, lib, "source")     # supply 120
+    c1 = _add(doc, lib, "consumer", 200)
+    c2 = _add(doc, lib, "consumer", 400)
+    c1.properties.set_value("demand", 100)
+    c2.properties.set_value("demand", 100)  # total demand 200 > supply 120
+    doc.add_edge(EdgeModel(doc.next_id("e"), src.node_id, "out", c1.node_id, "in"))
+    doc.add_edge(EdgeModel(doc.next_id("e"), src.node_id, "out", c2.node_id, "in"))
+    result = flow_solver.apply_to_document(doc)
+    served = sum(abs(f) for f in result.edge_flow.values())
+    assert served == pytest.approx(120)  # cannot exceed supply
+
+
+def test_no_sources_uses_nominal_flow(qapp, lib):
+    doc = DiagramDocument()
+    a = _add(doc, lib, "junction")
+    b = _add(doc, lib, "junction", 200)
+    e = doc.add_edge(EdgeModel(doc.next_id("e"), a.node_id, "e", b.node_id, "w"))
+    flow_solver.apply_to_document(doc)
+    assert e.flow == pytest.approx(1.0)
+    assert e.flow_direction == 1
+
+
+def test_empty_document(qapp):
+    doc = DiagramDocument()
+    result = flow_solver.solve(doc)
+    assert result.edge_flow == {}

tests/test_routing.py

@@ -0,0 +1,61 @@
+"""Tests for orthogonal routing and crossing detection (pure geometry)."""
+import math
+
+from pipeline_editor.view.routing import (
+    orthogonal_route, polyline_segments, segment_crossings,
+)
+
+
+def _is_orthogonal(points):
+    for (x0, y0), (x1, y1) in zip(points, points[1:]):
+        if not (math.isclose(x0, x1) or math.isclose(y0, y1)):
+            return False
+    return True
+
+
+def test_route_is_orthogonal_horizontal_ports():
+    # source outlet facing right -> pump inlet facing left
+    pts = orthogonal_route((0, 0), (1, 0), (200, 80), (-1, 0), grid=20)
+    assert pts[0] == (0, 0)
+    assert pts[-1] == (200, 80)
+    assert _is_orthogonal(pts)
+    assert len(pts) >= 2
+
+
+def test_route_is_orthogonal_vertical_ports():
+    pts = orthogonal_route((50, 0), (0, -1), (130, 200), (0, 1), grid=20)
+    assert pts[0] == (50, 0) and pts[-1] == (130, 200)
+    assert _is_orthogonal(pts)
+
+
+def test_route_straight_line_collapses():
+    # aligned ports facing each other -> essentially a straight run
+    pts = orthogonal_route((0, 40), (1, 0), (200, 40), (-1, 0), grid=20)
+    assert _is_orthogonal(pts)
+    assert all(math.isclose(p[1], 40) for p in pts)
+
+
+def test_bends_snapped_to_grid():
+    pts = orthogonal_route((0, 0), (1, 0), (207, 83), (-1, 0), grid=20)
+    for x, y in pts[1:-1]:
+        assert math.isclose(x % 20, 0) or math.isclose(x % 20, 20)
+
+
+def test_perpendicular_crossing_detected():
+    horiz = polyline_segments([(0, 50), (100, 50)])
+    vert = polyline_segments([(50, 0), (50, 100)])
+    cx = segment_crossings(horiz, vert)
+    assert cx == [(50, 50)]
+
+
+def test_shared_endpoint_is_not_a_crossing():
+    # a T-junction where the vertical touches the horizontal endpoint
+    horiz = polyline_segments([(0, 50), (50, 50)])
+    vert = polyline_segments([(50, 50), (50, 100)])
+    assert segment_crossings(horiz, vert) == []
+
+
+def test_parallel_segments_no_crossing():
+    a = polyline_segments([(0, 50), (100, 50)])
+    b = polyline_segments([(0, 70), (100, 70)])
+    assert segment_crossings(a, b) == []