feat(co4e): tách 6 widget UI khỏi ui/co4e_tab.py sang presentation/co4e/*
Lane N3 (Co4E Studio) — dùng bộ workflow refactor-god-file, mỗi bước có characterization test trước khi tách, hậu kiểm ranh giới tầng sau mỗi bước: - skills_list_panel.py / agent_list_panel.py — 2 khu vực sidebar - co4e_canvas_widget.py + canvas_items.py + canvas_interaction_mixin.py — Co4ECanvas tách 3 file (vượt 400 dòng nếu đứng một mình) - node_property_panel.py + node_property_actions_mixin.py + step_config_section.py — StepConfigPanel, cùng lý do - co4e_run_control_widget.py — RunsPagePanel (trang Flow Status) - co4e_chat_view.py — ChatPanel + _ChatInput + helper autocomplete - palette_list.py — _PaletteList dời khỏi ui/co4e_tab.py, hết import ngược presentation -> ui (agent/skills panel giờ import top-level) ui/co4e_tab.py giảm 2089 -> 1878 dòng, chỉ còn phần wiring + business logic (Co4ERunManager/AgentWorker chưa đổi — nằm ngoài phạm vi này, xem docstring presentation/co4e/co4e_tab.py). ui/co4e_canvas.py và ui/co4e_config_panel.py còn lại là compat shim re-export, không đổi API cho bên gọi. Thêm tests/test_co4e_integration.py — dựng thật Co4ETab qua build_co4e_tab(), lái luồng qua nhiều panel trong cùng instance (thêm node, mở/gập chat, chuyển trang Flow Status rồi quay lại không mất state canvas) — bắt lỗi wiring xuyên-panel mà characterization test từng panel riêng không thấy được. Đã xác minh: pytest 348 passed/1 skipped, tools/check_co4e.py sạch, không file nào >400 dòng, domain/application không import PySide6, và so pixel before/after (git worktree tại HEAD cũ) ra 0/1.125.000 pixel khác biệt. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
+22
-775
@@ -12,780 +12,27 @@ Kept UI-only; the graph model lives in ``core/co4e.py``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
from typing import Dict, Optional
|
||||
|
||||
from PySide6.QtCore import QPointF, QRectF, Qt, Signal
|
||||
from PySide6.QtGui import QBrush, QColor, QPainterPath, QPen, QPolygonF
|
||||
from PySide6.QtWidgets import (
|
||||
QGraphicsItem, QGraphicsObject, QGraphicsPathItem, QGraphicsScene,
|
||||
QGraphicsView, QMenu,
|
||||
# _NodeItem/_EdgeItem (hằng số vẽ + hai lớp QGraphicsItem) đã dời sang
|
||||
# presentation/co4e/canvas_items.py; Co4ECanvas (mutation đồ thị) đã dời sang
|
||||
# presentation/co4e/co4e_canvas_widget.py, phần tương tác view (zoom/pan/
|
||||
# relayout/phím tắt/kéo-thả) nằm trong _CanvasInteractionMixin cùng thư mục.
|
||||
# Không đổi hành vi — xem characterization test cùng tên và docstring ở từng
|
||||
# file đích. Import ĐÍCH DANH tên gốc, không alias — nếu đổi thành
|
||||
# `import co4e_canvas_widget as _w` thì các chỗ gọi bên dưới (và cả test cũ)
|
||||
# vẫn chạy được vì Python cho phép, nhưng lại sai mục đích của việc giữ nguyên
|
||||
# tên: test characterization import trực tiếp các tên này TỪ module này,
|
||||
# alias sẽ làm test đó ngưng chứng minh được điều nó cần chứng minh.
|
||||
from ..presentation.co4e.canvas_items import (
|
||||
_NODE_H, _NODE_W, _PORT_HIT, _PORT_R, _EdgeItem, _NodeItem, _status_color,
|
||||
)
|
||||
|
||||
from ..core.co4e import (
|
||||
STEP_DONE, STEP_ERROR, STEP_PLANNED, STEP_RUNNING, Edge, Node, Step,
|
||||
compute_waves, new_edge_id, new_node_id,
|
||||
# 8 hàm hình học thuần đã dời sang canvas_geometry.py (không đổi hành vi, xem
|
||||
# characterization test cùng tên). Import ĐÍCH DANH tên gốc, không alias — nếu
|
||||
# đổi thành `import canvas_geometry as _g` thì các chỗ gọi bên dưới (và cả
|
||||
# test cũ) vẫn chạy được vì Python cho phép, nhưng lại sai mục đích của việc
|
||||
# giữ nguyên tên: test characterization import trực tiếp các tên này TỪ module
|
||||
# này, alias sẽ làm test đó ngưng chứng minh được điều nó cần chứng minh.
|
||||
from ..presentation.co4e.canvas_geometry import (
|
||||
_dist, _elide, _hits, _ortho_path, _route, _rounded_path, _seg_hits_rect,
|
||||
_towards,
|
||||
)
|
||||
from ..theme import current_palette
|
||||
|
||||
|
||||
def _status_color(status: str) -> str:
|
||||
"""Accent colour for a step's run status. Resolved per paint so the canvas
|
||||
follows a live theme switch."""
|
||||
p = current_palette()
|
||||
return {
|
||||
"idle": p.text_muted, STEP_RUNNING: p.accent, STEP_DONE: p.success,
|
||||
STEP_ERROR: p.danger, STEP_PLANNED: p.purple, "pending": p.text_faint,
|
||||
}.get(status, p.text_muted)
|
||||
|
||||
CO4E_MIME = "application/x-co4e-step"
|
||||
|
||||
_NODE_W, _NODE_H = 210, 96
|
||||
_PORT_R = 6 # output port radius (the drag-to-connect handle)
|
||||
_PORT_HIT = 15 # click tolerance around a port
|
||||
_CORNER_R = 12 # edge elbow corner radius
|
||||
|
||||
|
||||
class _NodeItem(QGraphicsObject):
|
||||
"""One draggable step card. Emits signals via the parent canvas."""
|
||||
|
||||
def __init__(self, node: Node, canvas: "Co4ECanvas"):
|
||||
super().__init__()
|
||||
self.node = node
|
||||
self.canvas = canvas
|
||||
self.status = "idle"
|
||||
self._porting = False
|
||||
self.setFlags(QGraphicsItem.ItemIsMovable | QGraphicsItem.ItemIsSelectable
|
||||
| QGraphicsItem.ItemSendsGeometryChanges)
|
||||
self.setAcceptHoverEvents(True)
|
||||
self.setPos(node.x, node.y)
|
||||
self.setZValue(2)
|
||||
|
||||
def boundingRect(self) -> QRectF:
|
||||
# slack left/right so the input/output ports (now on the sides) paint cleanly
|
||||
return QRectF(-_PORT_R - 2, -3, _NODE_W + 2 * _PORT_R + 4, _NODE_H + 6)
|
||||
|
||||
def _card_rect(self) -> QRectF:
|
||||
return QRectF(1, 1, _NODE_W - 2, _NODE_H - 2)
|
||||
|
||||
def paint(self, p, _opt, _widget=None):
|
||||
tok = current_palette()
|
||||
step = self.node.data
|
||||
accent = QColor(_status_color(self.status))
|
||||
body = QColor(tok.surface_raised)
|
||||
border = QColor(tok.accent) if self.isSelected() else QColor(tok.border)
|
||||
p.setRenderHint(p.RenderHint.Antialiasing)
|
||||
rect = self._card_rect()
|
||||
path = QPainterPath()
|
||||
radius = float(tok.radius_lg)
|
||||
path.addRoundedRect(rect, radius, radius)
|
||||
p.fillPath(path, QBrush(body))
|
||||
p.setPen(QPen(border, 2 if self.isSelected() else 1))
|
||||
p.drawPath(path)
|
||||
# header stripe — a tint of the status colour, not the status colour
|
||||
# itself, so the card's own text stays the brightest thing on it.
|
||||
hdr = QRectF(rect.left(), rect.top(), rect.width(), 26)
|
||||
hpath = QPainterPath()
|
||||
hpath.addRoundedRect(hdr, radius, radius)
|
||||
stripe = QColor(accent)
|
||||
stripe.setAlpha(48)
|
||||
p.fillPath(hpath, QBrush(stripe))
|
||||
# label
|
||||
p.setPen(QColor(tok.text))
|
||||
f = p.font(); f.setBold(True); f.setPointSize(9); p.setFont(f)
|
||||
p.drawText(QRectF(10, 4, _NODE_W - 20, 20), Qt.AlignVCenter | Qt.AlignLeft,
|
||||
_elide(step.label, 26))
|
||||
# role badge + status
|
||||
f.setBold(False); f.setPointSize(8); p.setFont(f)
|
||||
p.setPen(accent)
|
||||
p.drawText(QRectF(10, 30, _NODE_W - 20, 16), Qt.AlignLeft, step.role)
|
||||
# body: instructions preview OR sub-agent chips
|
||||
p.setPen(QColor(tok.text_muted))
|
||||
if step.is_parallel:
|
||||
preview = "⇉ " + ", ".join(s.agent for s in step.sub_agents) if step.sub_agents else "⇉ (no sub-agents)"
|
||||
else:
|
||||
preview = step.instructions or "(no instructions)"
|
||||
p.drawText(QRectF(10, 46, _NODE_W - 20, 30), Qt.TextWordWrap | Qt.AlignTop,
|
||||
_elide(preview, 66))
|
||||
# footer: model + skills + status dot
|
||||
p.setPen(QColor(tok.text_faint))
|
||||
foot = []
|
||||
if step.model:
|
||||
foot.append(step.model)
|
||||
if step.skills:
|
||||
foot.append(f"skills:{len(step.skills)}")
|
||||
foot.append(self.status)
|
||||
p.drawText(QRectF(10, _NODE_H - 18, _NODE_W - 20, 14), Qt.AlignLeft,
|
||||
_elide(" · ".join(foot), 34))
|
||||
# ---- ports ---------------------------------------------------------
|
||||
# input port (top-center): hollow. output port (bottom-center): filled —
|
||||
# the drag handle you pull to wire an edge to another step.
|
||||
port_col = QColor(tok.accent)
|
||||
# input port (left-center): hollow. output port (right-center): filled —
|
||||
# the drag handle you pull to wire an edge to the next step (left→right).
|
||||
p.setBrush(QBrush(body)); p.setPen(QPen(port_col, 1.4))
|
||||
p.drawEllipse(QPointF(1, _NODE_H / 2), _PORT_R - 1, _PORT_R - 1)
|
||||
p.setBrush(QBrush(port_col)); p.setPen(QPen(port_col, 1.4))
|
||||
p.drawEllipse(QPointF(_NODE_W - 1, _NODE_H / 2), _PORT_R, _PORT_R)
|
||||
|
||||
def _in_out_port(self, pos: QPointF) -> bool:
|
||||
d = pos - QPointF(_NODE_W, _NODE_H / 2)
|
||||
return (d.x() * d.x() + d.y() * d.y()) ** 0.5 <= _PORT_HIT
|
||||
|
||||
def itemChange(self, change, value):
|
||||
if change == QGraphicsItem.ItemPositionHasChanged:
|
||||
self.node.x = float(self.pos().x())
|
||||
self.node.y = float(self.pos().y())
|
||||
self.canvas._reposition_edges()
|
||||
self.canvas.graph_changed.emit()
|
||||
elif change == QGraphicsItem.ItemSelectedHasChanged:
|
||||
# a selected/edited node comes to the front (above the edges at z=3)
|
||||
self.setZValue(4 if value else 2)
|
||||
if value:
|
||||
self.canvas.node_selected.emit(self.node.id)
|
||||
return super().itemChange(change, value)
|
||||
|
||||
def hoverMoveEvent(self, e):
|
||||
# a hand cursor over the output port hints it's draggable-to-connect
|
||||
self.setCursor(Qt.PointingHandCursor if self._in_out_port(e.pos()) else Qt.ArrowCursor)
|
||||
super().hoverMoveEvent(e)
|
||||
|
||||
def mousePressEvent(self, e):
|
||||
if self.canvas._connect_from is not None:
|
||||
self.canvas._finish_connect(self.node.id)
|
||||
e.accept()
|
||||
return
|
||||
if e.button() == Qt.LeftButton and self._in_out_port(e.pos()):
|
||||
# start a manual drag-to-connect from this node's output port
|
||||
self._porting = True
|
||||
self.canvas.begin_port_drag(self.node.id, self.mapToScene(QPointF(_NODE_W, _NODE_H / 2)))
|
||||
e.accept()
|
||||
return
|
||||
super().mousePressEvent(e)
|
||||
|
||||
def mouseMoveEvent(self, e):
|
||||
if self._porting:
|
||||
self.canvas.update_port_drag(self.mapToScene(e.pos()))
|
||||
e.accept()
|
||||
return
|
||||
super().mouseMoveEvent(e)
|
||||
|
||||
def mouseReleaseEvent(self, e):
|
||||
if self._porting:
|
||||
self._porting = False
|
||||
self.canvas.finish_port_drag(self.mapToScene(e.pos()))
|
||||
e.accept()
|
||||
return
|
||||
super().mouseReleaseEvent(e)
|
||||
|
||||
def mouseDoubleClickEvent(self, e):
|
||||
self.canvas.node_activated.emit(self.node.id)
|
||||
e.accept()
|
||||
|
||||
def contextMenuEvent(self, e):
|
||||
menu = QMenu()
|
||||
a_add = menu.addAction("＋ Add next step")
|
||||
a_conn = menu.addAction("→ Connect from here")
|
||||
a_del = menu.addAction("🗑 Delete step")
|
||||
chosen = menu.exec(e.screenPos())
|
||||
if chosen is a_add:
|
||||
self.canvas.add_step_below(self.node.id)
|
||||
elif chosen is a_conn:
|
||||
self.canvas.begin_connect(self.node.id)
|
||||
elif chosen is a_del:
|
||||
self.canvas.delete_node(self.node.id)
|
||||
e.accept()
|
||||
|
||||
def center(self) -> QPointF:
|
||||
return self.pos() + QPointF(_NODE_W / 2, _NODE_H / 2)
|
||||
|
||||
|
||||
def _dist(a: QPointF, b: QPointF) -> float:
|
||||
return ((a.x() - b.x()) ** 2 + (a.y() - b.y()) ** 2) ** 0.5
|
||||
|
||||
|
||||
def _towards(a: QPointF, b: QPointF, d: float) -> QPointF:
|
||||
dist = _dist(a, b)
|
||||
if dist < 1e-6:
|
||||
return QPointF(a)
|
||||
t = d / dist
|
||||
return QPointF(a.x() + (b.x() - a.x()) * t, a.y() + (b.y() - a.y()) * t)
|
||||
|
||||
|
||||
def _rounded_path(points, r: float = _CORNER_R) -> QPainterPath:
|
||||
"""Build a path through axis-aligned ``points`` with rounded corners at each
|
||||
bend ("vuông bo cong ở góc")."""
|
||||
if not points:
|
||||
return QPainterPath()
|
||||
path = QPainterPath(points[0])
|
||||
if len(points) == 1:
|
||||
return path
|
||||
for i in range(1, len(points) - 1):
|
||||
prev, cur, nxt = points[i - 1], points[i], points[i + 1]
|
||||
rr = min(r, _dist(prev, cur) / 2.0, _dist(cur, nxt) / 2.0)
|
||||
path.lineTo(_towards(cur, prev, rr))
|
||||
path.quadTo(cur, _towards(cur, nxt, rr))
|
||||
path.lineTo(points[-1])
|
||||
return path
|
||||
|
||||
|
||||
def _seg_hits_rect(p1: QPointF, p2: QPointF, rect: QRectF) -> bool:
|
||||
"""Axis-aligned segment vs rectangle overlap (all routed segments are H or V)."""
|
||||
x1, y1, x2, y2 = p1.x(), p1.y(), p2.x(), p2.y()
|
||||
if abs(y1 - y2) < 0.5: # horizontal
|
||||
if rect.top() <= y1 <= rect.bottom():
|
||||
lo, hi = sorted((x1, x2))
|
||||
return not (hi < rect.left() or lo > rect.right())
|
||||
return False
|
||||
if abs(x1 - x2) < 0.5: # vertical
|
||||
if rect.left() <= x1 <= rect.right():
|
||||
lo, hi = sorted((y1, y2))
|
||||
return not (hi < rect.top() or lo > rect.bottom())
|
||||
return False
|
||||
box = QRectF(QPointF(min(x1, x2), min(y1, y2)), QPointF(max(x1, x2), max(y1, y2)))
|
||||
return rect.intersects(box)
|
||||
|
||||
|
||||
def _hits(points, obstacles) -> bool:
|
||||
for i in range(len(points) - 1):
|
||||
for r in obstacles:
|
||||
if _seg_hits_rect(points[i], points[i + 1], r):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _route(src: QPointF, dst: QPointF, obstacles=None):
|
||||
"""Waypoints for a LEFT→RIGHT orthogonal edge from ``src`` (a node's right
|
||||
output) to ``dst`` (the next node's left input) that AVOIDS the other node
|
||||
rectangles: try the straight elbow, then a clear vertical band, then a
|
||||
top/bottom detour — so a connector never overlaps or hides behind a step."""
|
||||
obstacles = list(obstacles or [])
|
||||
if abs(src.y() - dst.y()) < 1.5:
|
||||
cand = [src, dst]
|
||||
if not _hits(cand, obstacles):
|
||||
return cand
|
||||
mid_x = (src.x() + dst.x()) / 2.0
|
||||
base = [src, QPointF(mid_x, src.y()), QPointF(mid_x, dst.y()), dst]
|
||||
if not _hits(base, obstacles):
|
||||
return base
|
||||
# 1) slide the vertical run to a clear band between the two columns
|
||||
lo, hi = min(src.x(), dst.x()) + 6, max(src.x(), dst.x()) - 6
|
||||
if hi > lo:
|
||||
for frac in (0.5, 0.35, 0.65, 0.2, 0.8):
|
||||
x = lo + (hi - lo) * frac
|
||||
cand = [src, QPointF(x, src.y()), QPointF(x, dst.y()), dst]
|
||||
if not _hits(cand, obstacles):
|
||||
return cand
|
||||
# 2) detour above/below every obstacle, then back in
|
||||
margin = 44.0
|
||||
ys = [src.y(), dst.y()] + [r.top() for r in obstacles] + [r.bottom() for r in obstacles]
|
||||
out_x, in_x = src.x() + 34, dst.x() - 34 # short stubs out of the side ports
|
||||
for side_y in (min(ys) - margin, max(ys) + margin):
|
||||
cand = [src, QPointF(out_x, src.y()), QPointF(out_x, side_y),
|
||||
QPointF(in_x, side_y), QPointF(in_x, dst.y()), dst]
|
||||
if not _hits(cand, obstacles):
|
||||
return cand
|
||||
return base
|
||||
|
||||
|
||||
def _ortho_path(src: QPointF, dst: QPointF, r: float = _CORNER_R) -> QPainterPath:
|
||||
"""Rounded orthogonal elbow (no obstacle avoidance) — used for the transient
|
||||
drag-to-connect line and by callers that pass no obstacles."""
|
||||
return _rounded_path(_route(src, dst), r)
|
||||
|
||||
|
||||
class _EdgeItem(QGraphicsPathItem):
|
||||
def __init__(self, edge: Edge, canvas: "Co4ECanvas"):
|
||||
super().__init__()
|
||||
self.edge = edge
|
||||
self.canvas = canvas
|
||||
self._dst: Optional[QPointF] = None
|
||||
# Above node cards (z=2) so a connecting line is never hidden behind a
|
||||
# step; a selected node bumps itself to the front while being edited.
|
||||
self.setZValue(3)
|
||||
self.setFlag(QGraphicsItem.ItemIsSelectable, True)
|
||||
self.setAcceptHoverEvents(True)
|
||||
self._hover = False
|
||||
self._apply_pen()
|
||||
|
||||
def _apply_pen(self):
|
||||
tok = current_palette()
|
||||
if self.isSelected():
|
||||
color, w = QColor(tok.accent), 3
|
||||
elif self._hover:
|
||||
color, w = QColor(tok.text_muted), 3
|
||||
else:
|
||||
color, w = QColor(tok.border_strong), 2
|
||||
self.setPen(QPen(color, w, Qt.SolidLine, Qt.RoundCap, Qt.RoundJoin))
|
||||
|
||||
def update_path(self, points):
|
||||
self._dst = points[-1] if points else None
|
||||
self.setPath(_rounded_path(points))
|
||||
|
||||
def boundingRect(self):
|
||||
return super().boundingRect().adjusted(-10, -10, 10, 10) # room for the arrowhead
|
||||
|
||||
def shape(self):
|
||||
# Widen the clickable/selectable area so a thin line is easy to grab.
|
||||
from PySide6.QtGui import QPainterPathStroker
|
||||
stroker = QPainterPathStroker()
|
||||
stroker.setWidth(14)
|
||||
return stroker.createStroke(self.path())
|
||||
|
||||
def hoverEnterEvent(self, e):
|
||||
self._hover = True
|
||||
self._apply_pen()
|
||||
self.update()
|
||||
super().hoverEnterEvent(e)
|
||||
|
||||
def hoverLeaveEvent(self, e):
|
||||
self._hover = False
|
||||
self._apply_pen()
|
||||
self.update()
|
||||
super().hoverLeaveEvent(e)
|
||||
|
||||
def paint(self, p, opt, widget=None):
|
||||
self._apply_pen()
|
||||
super().paint(p, opt, widget)
|
||||
# arrowhead at the target, pointing right into its (left) input port
|
||||
if self._dst is not None:
|
||||
p.setRenderHint(p.RenderHint.Antialiasing)
|
||||
tip = self._dst
|
||||
s = 7.0
|
||||
tri = QPolygonF([
|
||||
QPointF(tip.x() + 1, tip.y()),
|
||||
QPointF(tip.x() - s, tip.y() - s * 0.7),
|
||||
QPointF(tip.x() - s, tip.y() + s * 0.7),
|
||||
])
|
||||
col = self.pen().color()
|
||||
p.setBrush(QBrush(col))
|
||||
p.setPen(QPen(col, 1))
|
||||
p.drawPolygon(tri)
|
||||
|
||||
def contextMenuEvent(self, e):
|
||||
menu = QMenu()
|
||||
act_del = menu.addAction("🗑 Delete connection")
|
||||
if menu.exec(e.screenPos()) is act_del:
|
||||
self.canvas.delete_edge(self.edge)
|
||||
e.accept()
|
||||
|
||||
|
||||
def _elide(text: str, n: int) -> str:
|
||||
text = (text or "").replace("\n", " ")
|
||||
return text if len(text) <= n else text[: n - 1] + "…"
|
||||
|
||||
|
||||
class Co4ECanvas(QGraphicsView):
|
||||
node_selected = Signal(str) # a node was clicked (→ config panel)
|
||||
node_activated = Signal(str) # double-clicked
|
||||
graph_changed = Signal() # nodes/edges/positions changed (autosave)
|
||||
|
||||
_ZOOM_MIN, _ZOOM_MAX = 0.3, 3.0
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.setObjectName("co4eCanvas") # themed frame (see theme.py)
|
||||
self._scene = QGraphicsScene(self)
|
||||
self.setScene(self._scene)
|
||||
self.setRenderHint(self.renderHints().Antialiasing)
|
||||
self.setDragMode(QGraphicsView.RubberBandDrag)
|
||||
self.setTransformationAnchor(QGraphicsView.AnchorUnderMouse)
|
||||
self.setAcceptDrops(True)
|
||||
self._nodes: Dict[str, _NodeItem] = {}
|
||||
self._edges: list[_EdgeItem] = []
|
||||
self._connect_from: Optional[str] = None
|
||||
self._zoom = 1.0
|
||||
self._panning = False # middle-mouse drag-to-pan
|
||||
self._pan_start = None
|
||||
self._overlay = None # bottom-left zoom/fit controls (parented to viewport)
|
||||
# manual drag-to-connect state
|
||||
self._port_src: Optional[str] = None
|
||||
self._port_src_pt: Optional[QPointF] = None
|
||||
self._temp_edge: Optional[QGraphicsPathItem] = None
|
||||
|
||||
# ---- bottom-left overlay (zoom / fit) --------------------------------
|
||||
def add_overlay(self, widget) -> None:
|
||||
self._overlay = widget
|
||||
widget.setParent(self.viewport())
|
||||
widget.show()
|
||||
widget.raise_()
|
||||
self._place_overlay()
|
||||
|
||||
def _place_overlay(self) -> None:
|
||||
if self._overlay is not None:
|
||||
self._overlay.adjustSize()
|
||||
vp = self.viewport()
|
||||
self._overlay.move(12, vp.height() - self._overlay.height() - 12)
|
||||
self._overlay.raise_()
|
||||
|
||||
def resizeEvent(self, e): # noqa: N802
|
||||
super().resizeEvent(e)
|
||||
self._place_overlay()
|
||||
|
||||
def scrollContentsBy(self, dx, dy): # noqa: N802
|
||||
# QGraphicsView scrolls the viewport's child widgets along with the
|
||||
# scene, so panning/scrolling would drag the zoom overlay off-corner.
|
||||
# Re-pin it after every scroll so +/−/fit stay fixed in place.
|
||||
super().scrollContentsBy(dx, dy)
|
||||
self._place_overlay()
|
||||
|
||||
def showEvent(self, e): # noqa: N802
|
||||
super().showEvent(e)
|
||||
self._place_overlay() # viewport size is final once shown
|
||||
|
||||
# ---- load / serialize -------------------------------------------------
|
||||
def load(self, nodes, edges) -> None:
|
||||
self._scene.clear()
|
||||
self._nodes.clear()
|
||||
self._edges.clear()
|
||||
self._connect_from = None
|
||||
self._port_src = None
|
||||
self._temp_edge = None
|
||||
for n in nodes:
|
||||
item = _NodeItem(n, self)
|
||||
self._nodes[n.id] = item
|
||||
self._scene.addItem(item)
|
||||
for e in edges:
|
||||
if e.source in self._nodes and e.target in self._nodes:
|
||||
self._add_edge_item(e)
|
||||
self._reposition_edges()
|
||||
|
||||
def nodes(self):
|
||||
return [it.node for it in self._nodes.values()]
|
||||
|
||||
def edges(self):
|
||||
return [it.edge for it in self._edges]
|
||||
|
||||
# ---- mutation ---------------------------------------------------------
|
||||
def add_node(self, step: Step, x: float = 60.0, y: float = 60.0,
|
||||
connect_from: str = "") -> str:
|
||||
node = Node(id=new_node_id(), x=x, y=y, data=step)
|
||||
item = _NodeItem(node, self)
|
||||
self._nodes[node.id] = item
|
||||
self._scene.addItem(item)
|
||||
if connect_from and connect_from in self._nodes:
|
||||
self._make_edge(connect_from, node.id)
|
||||
self._reposition_edges()
|
||||
self.graph_changed.emit()
|
||||
self.node_selected.emit(node.id)
|
||||
return node.id
|
||||
|
||||
def add_step_below(self, node_id: str) -> None:
|
||||
"""Add the next step to the RIGHT of ``node_id`` (horizontal flow)."""
|
||||
parent = self._nodes.get(node_id)
|
||||
if parent is None:
|
||||
return
|
||||
step = Step(label="New Step")
|
||||
self.add_node(step, x=parent.node.x + _NODE_W + 150, y=parent.node.y, connect_from=node_id)
|
||||
|
||||
def _chain_tail(self) -> str:
|
||||
"""A node with no outgoing edge (so a freshly added node chains on)."""
|
||||
sources = {e.edge.source for e in self._edges}
|
||||
tails = [nid for nid in self._nodes if nid not in sources]
|
||||
return tails[-1] if tails else (next(reversed(self._nodes), "") if self._nodes else "")
|
||||
|
||||
def add_palette_step(self, step: Step, pos: QPointF) -> None:
|
||||
tail = self._chain_tail()
|
||||
self.add_node(step, x=pos.x(), y=pos.y(), connect_from=tail)
|
||||
|
||||
def begin_connect(self, source_id: str) -> None:
|
||||
self._connect_from = source_id
|
||||
|
||||
def _finish_connect(self, target_id: str) -> None:
|
||||
src = self._connect_from
|
||||
self._connect_from = None
|
||||
if src and src != target_id:
|
||||
self._make_edge(src, target_id)
|
||||
|
||||
# ---- manual drag-to-connect (from a node's output port) ---------------
|
||||
def begin_port_drag(self, source_id: str, scene_pt: QPointF) -> None:
|
||||
self._port_src = source_id
|
||||
self._port_src_pt = scene_pt
|
||||
self._temp_edge = QGraphicsPathItem()
|
||||
self._temp_edge.setZValue(3.5) # above nodes + edges while connecting
|
||||
self._temp_edge.setPen(
|
||||
QPen(QColor(current_palette().accent), 2, Qt.DashLine, Qt.RoundCap))
|
||||
self._scene.addItem(self._temp_edge)
|
||||
|
||||
def update_port_drag(self, scene_pt: QPointF) -> None:
|
||||
if self._temp_edge is None or self._port_src_pt is None:
|
||||
return
|
||||
self._temp_edge.setPath(_ortho_path(self._port_src_pt, scene_pt))
|
||||
|
||||
def finish_port_drag(self, scene_pt: QPointF) -> None:
|
||||
src = self._port_src
|
||||
if self._temp_edge is not None:
|
||||
self._scene.removeItem(self._temp_edge)
|
||||
self._temp_edge = None
|
||||
self._port_src = None
|
||||
self._port_src_pt = None
|
||||
tgt = self._node_at(scene_pt)
|
||||
if src and tgt and tgt != src:
|
||||
self._make_edge(src, tgt)
|
||||
|
||||
def _node_at(self, scene_pt: QPointF) -> Optional[str]:
|
||||
for it in self._scene.items(scene_pt):
|
||||
if isinstance(it, _NodeItem):
|
||||
return it.node.id
|
||||
return None
|
||||
|
||||
def _make_edge(self, source: str, target: str) -> None:
|
||||
if source == target:
|
||||
return
|
||||
if any(e.edge.source == source and e.edge.target == target for e in self._edges):
|
||||
return
|
||||
edge = Edge(id=new_edge_id(source, target), source=source, target=target)
|
||||
self._add_edge_item(edge)
|
||||
self._reposition_edges()
|
||||
self.graph_changed.emit()
|
||||
|
||||
def _add_edge_item(self, edge: Edge) -> None:
|
||||
item = _EdgeItem(edge, self)
|
||||
self._edges.append(item)
|
||||
self._scene.addItem(item)
|
||||
|
||||
def delete_edge(self, edge: Edge) -> None:
|
||||
for e in list(self._edges):
|
||||
if e.edge is edge or (e.edge.source == edge.source and e.edge.target == edge.target):
|
||||
self._scene.removeItem(e)
|
||||
self._edges.remove(e)
|
||||
self.graph_changed.emit()
|
||||
|
||||
def delete_node(self, node_id: str) -> None:
|
||||
item = self._nodes.pop(node_id, None)
|
||||
if item is None:
|
||||
return
|
||||
self._scene.removeItem(item)
|
||||
for e in list(self._edges):
|
||||
if e.edge.source == node_id or e.edge.target == node_id:
|
||||
self._scene.removeItem(e)
|
||||
self._edges.remove(e)
|
||||
self._reposition_edges()
|
||||
self.graph_changed.emit()
|
||||
|
||||
def delete_selected(self) -> None:
|
||||
for nid in [it.node.id for it in self._nodes.values() if it.isSelected()]:
|
||||
self.delete_node(nid)
|
||||
for e in [it.edge for it in self._edges if it.isSelected()]:
|
||||
self.delete_edge(e)
|
||||
|
||||
# ---- zoom / fit -------------------------------------------------------
|
||||
def _zoom_by(self, factor: float) -> None:
|
||||
# Derive the CURRENT scale from the live transform (never a separate
|
||||
# accumulator that can drift out of sync with fit_view/relayout/reset —
|
||||
# that drift is what made the +/− buttons and Ctrl+wheel randomly stop
|
||||
# working). Clamp the TARGET to the range and apply the exact factor to
|
||||
# reach it, so zooming still works right up to the limits.
|
||||
cur = self.transform().m11() or 1.0
|
||||
target = max(self._ZOOM_MIN, min(self._ZOOM_MAX, cur * factor))
|
||||
if abs(target - cur) < 1e-6:
|
||||
return
|
||||
self.scale(target / cur, target / cur)
|
||||
self._zoom = target
|
||||
|
||||
def zoom_in(self) -> None:
|
||||
self._zoom_by(1.15)
|
||||
|
||||
def zoom_out(self) -> None:
|
||||
self._zoom_by(1 / 1.15)
|
||||
|
||||
def reset_zoom(self) -> None:
|
||||
self.resetTransform()
|
||||
self._zoom = 1.0
|
||||
|
||||
def wheelEvent(self, e):
|
||||
# Ctrl+wheel = zoom (anchored under the cursor); Shift+wheel = pan
|
||||
# horizontally; plain wheel scrolls vertically.
|
||||
if e.modifiers() & Qt.ControlModifier:
|
||||
self._zoom_by(1.15 if e.angleDelta().y() > 0 else 1 / 1.15)
|
||||
e.accept()
|
||||
return
|
||||
if e.modifiers() & Qt.ShiftModifier:
|
||||
bar = self.horizontalScrollBar()
|
||||
bar.setValue(bar.value() - e.angleDelta().y())
|
||||
e.accept()
|
||||
return
|
||||
super().wheelEvent(e)
|
||||
|
||||
# ---- middle-mouse drag-to-pan ----------------------------------------
|
||||
def mousePressEvent(self, e):
|
||||
if e.button() == Qt.MiddleButton:
|
||||
self._panning = True
|
||||
self._pan_start = e.position().toPoint()
|
||||
self.setCursor(Qt.ClosedHandCursor)
|
||||
e.accept()
|
||||
return
|
||||
super().mousePressEvent(e)
|
||||
|
||||
def mouseMoveEvent(self, e):
|
||||
if self._panning and self._pan_start is not None:
|
||||
pos = e.position().toPoint()
|
||||
delta = pos - self._pan_start
|
||||
self._pan_start = pos
|
||||
self.horizontalScrollBar().setValue(self.horizontalScrollBar().value() - delta.x())
|
||||
self.verticalScrollBar().setValue(self.verticalScrollBar().value() - delta.y())
|
||||
e.accept()
|
||||
return
|
||||
super().mouseMoveEvent(e)
|
||||
|
||||
def mouseReleaseEvent(self, e):
|
||||
if e.button() == Qt.MiddleButton and self._panning:
|
||||
self._panning = False
|
||||
self.setCursor(Qt.ArrowCursor)
|
||||
e.accept()
|
||||
return
|
||||
super().mouseReleaseEvent(e)
|
||||
|
||||
def fit_view(self) -> None:
|
||||
"""Auto-fit: zoom/pan so every node is visible with a small margin."""
|
||||
rect = self._scene.itemsBoundingRect()
|
||||
if rect.isNull():
|
||||
return
|
||||
self.setSceneRect(rect.adjusted(-60, -60, 60, 60))
|
||||
self.fitInView(rect.adjusted(-40, -40, 40, 40), Qt.KeepAspectRatio)
|
||||
# keep the zoom accumulator in sync with the transform fitInView applied
|
||||
self._zoom = self.transform().m11() or 1.0
|
||||
|
||||
def relayout(self, hgap: float = 110.0, vgap: float = 40.0) -> None:
|
||||
"""Arrange nodes LEFT→RIGHT by dependency depth: each topological wave is
|
||||
a column (x = wave), siblings stacked vertically within it. Used to turn
|
||||
an old top-down graph into the horizontal flow layout."""
|
||||
nodes = [it.node for it in self._nodes.values()]
|
||||
edges = [it.edge for it in self._edges]
|
||||
if not nodes:
|
||||
return
|
||||
waves = compute_waves(nodes, edges)
|
||||
from collections import defaultdict
|
||||
cols: Dict[int, list] = defaultdict(list)
|
||||
for n in nodes:
|
||||
cols[waves.get(n.id, 0)].append(n)
|
||||
for w in sorted(cols):
|
||||
for row, n in enumerate(sorted(cols[w], key=lambda nn: (nn.y, nn.x))):
|
||||
item = self._nodes.get(n.id)
|
||||
if item is not None:
|
||||
item.setPos(w * (_NODE_W + hgap), row * (_NODE_H + vgap))
|
||||
self._reposition_edges()
|
||||
|
||||
def relayout_if_vertical(self) -> None:
|
||||
"""Convert a graph that's stacked vertically (the old top-down layout, or
|
||||
overlapping nodes) into the horizontal left→right layout — but leave a
|
||||
graph the user already arranged horizontally untouched."""
|
||||
nodes = [it.node for it in self._nodes.values()]
|
||||
if len(nodes) < 2:
|
||||
return
|
||||
xs = [n.x for n in nodes]
|
||||
if max(xs) - min(xs) < _NODE_W: # all in one column → it's vertical
|
||||
self.relayout()
|
||||
|
||||
def add_workflow(self, nodes, edges, at: Optional[QPointF] = None) -> None:
|
||||
"""Drop/merge a whole flow's nodes+edges onto the canvas with fresh ids
|
||||
(so the same template can be dropped several times). Offsets it near
|
||||
``at`` when given, else tiles it beside whatever is already there."""
|
||||
remap: Dict[str, str] = {}
|
||||
# offset so a dropped template doesn't land exactly on existing nodes
|
||||
ox = (at.x() - nodes[0].x) if (at and nodes) else (60 if self._nodes else 0)
|
||||
oy = (at.y() - nodes[0].y) if (at and nodes) else (60 if self._nodes else 0)
|
||||
for n in nodes:
|
||||
new = Node(id=new_node_id(), x=n.x + ox, y=n.y + oy, data=copy.deepcopy(n.data))
|
||||
remap[n.id] = new.id
|
||||
item = _NodeItem(new, self)
|
||||
self._nodes[new.id] = item
|
||||
self._scene.addItem(item)
|
||||
for e in edges:
|
||||
s, t = remap.get(e.source), remap.get(e.target)
|
||||
if s and t:
|
||||
self._add_edge_item(Edge(id=new_edge_id(s, t), source=s, target=t))
|
||||
self._reposition_edges()
|
||||
self.graph_changed.emit()
|
||||
|
||||
def update_node_status(self, node_id: str, status: str) -> None:
|
||||
item = self._nodes.get(node_id)
|
||||
if item is not None:
|
||||
item.status = status
|
||||
item.update()
|
||||
|
||||
def reset_statuses(self) -> None:
|
||||
for it in self._nodes.values():
|
||||
it.status = "idle"
|
||||
it.update()
|
||||
|
||||
def refresh_node(self, node_id: str) -> None:
|
||||
item = self._nodes.get(node_id)
|
||||
if item is not None:
|
||||
item.update()
|
||||
|
||||
def _node_rects(self, exclude):
|
||||
"""Rectangles of every node except ``exclude`` (inflated a little), used
|
||||
as obstacles the edge router steers around."""
|
||||
m = 12.0
|
||||
out = []
|
||||
for nid, item in self._nodes.items():
|
||||
if nid in exclude:
|
||||
continue
|
||||
p = item.pos()
|
||||
out.append(QRectF(p.x(), p.y(), _NODE_W, _NODE_H).adjusted(-m, -m, m, m))
|
||||
return out
|
||||
|
||||
def _reposition_edges(self) -> None:
|
||||
for e in self._edges:
|
||||
s = self._nodes.get(e.edge.source)
|
||||
t = self._nodes.get(e.edge.target)
|
||||
if s is None or t is None:
|
||||
continue
|
||||
src = s.pos() + QPointF(_NODE_W, _NODE_H / 2) # right-center (output)
|
||||
dst = t.pos() + QPointF(0, _NODE_H / 2) # left-center (input)
|
||||
obstacles = self._node_rects({e.edge.source, e.edge.target})
|
||||
e.update_path(_route(src, dst, obstacles))
|
||||
|
||||
# ---- key / drop -------------------------------------------------------
|
||||
def keyPressEvent(self, e):
|
||||
if e.key() in (Qt.Key_Delete, Qt.Key_Backspace):
|
||||
self.delete_selected()
|
||||
return
|
||||
if e.key() == Qt.Key_Escape:
|
||||
self._connect_from = None
|
||||
if self._temp_edge is not None:
|
||||
self._scene.removeItem(self._temp_edge)
|
||||
self._temp_edge = None
|
||||
self._port_src = None
|
||||
return
|
||||
if e.key() in (Qt.Key_Plus, Qt.Key_Equal) and (e.modifiers() & Qt.ControlModifier):
|
||||
self.zoom_in(); return
|
||||
if e.key() == Qt.Key_Minus and (e.modifiers() & Qt.ControlModifier):
|
||||
self.zoom_out(); return
|
||||
if e.key() == Qt.Key_0 and (e.modifiers() & Qt.ControlModifier):
|
||||
self.reset_zoom(); return
|
||||
super().keyPressEvent(e)
|
||||
|
||||
def dragEnterEvent(self, e):
|
||||
if e.mimeData().hasFormat(CO4E_MIME):
|
||||
e.acceptProposedAction()
|
||||
else:
|
||||
super().dragEnterEvent(e)
|
||||
|
||||
def dragMoveEvent(self, e):
|
||||
if e.mimeData().hasFormat(CO4E_MIME):
|
||||
e.acceptProposedAction()
|
||||
else:
|
||||
super().dragMoveEvent(e)
|
||||
|
||||
def dropEvent(self, e):
|
||||
if not e.mimeData().hasFormat(CO4E_MIME):
|
||||
super().dropEvent(e)
|
||||
return
|
||||
try:
|
||||
payload = json.loads(bytes(e.mimeData().data(CO4E_MIME)).decode("utf-8"))
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
return
|
||||
pos = self.mapToScene(e.position().toPoint())
|
||||
if isinstance(payload, dict) and payload.get("kind") == "workflow":
|
||||
# A whole flow dragged from the sidebar → merge its graph in.
|
||||
from ..core.co4e import workflow_from_dict
|
||||
wf = workflow_from_dict(payload.get("workflow", {}))
|
||||
if wf.nodes:
|
||||
self.add_workflow(wf.nodes, wf.edges, at=pos)
|
||||
else:
|
||||
from ..core.co4e import step_from_dict
|
||||
self.add_palette_step(step_from_dict(payload), pos)
|
||||
e.acceptProposedAction()
|
||||
from ..presentation.co4e.co4e_canvas_widget import Co4ECanvas, CO4E_MIME
|
||||
|
||||
Reference in New Issue
Block a user