"""Layout linting tools for Node-RED flows.""" import json from nodered_mcp.client import get_client from nodered_mcp.models.layout import LayoutIssue, LayoutReport from nodered_mcp.server import mcp # Node dimension lookup: maps common node types to (width, height) in pixels # Unknown types fall back to (120, 30) NODE_DIMENSIONS: dict[str, tuple[int, int]] = { "tab": (120, 30), "inject": (120, 30), "debug": (120, 30), "function": (180, 60), "switch": (180, 60), "change": (180, 60), "template": (180, 80), "http response": (180, 40), "http request": (180, 60), "websocket in": (180, 60), "websocket out": (180, 60), "mqtt in": (180, 60), "mqtt out": (180, 60), "tcp in": (180, 60), "tcp out": (180, 60), "udp in": (180, 60), "udp out": (180, 60), "split": (120, 30), "join": (120, 30), "sort": (120, 30), "batch": (180, 60), "link out": (180, 50), "link in": (180, 50), "trigger": (180, 60), "delay": (180, 60), "rbe": (180, 60), "comment": (200, 40), "catch": (180, 50), "status": (180, 50), "complete": (180, 50), "unknown": (120, 30), } # Default fallback for node types not in the lookup DEFAULT_NODE_WIDTH = 120 DEFAULT_NODE_HEIGHT = 30 MIN_SPACING = 20 def _get_node_dims(node: dict) -> tuple[int, int]: node_type = node.get("type", "unknown") return NODE_DIMENSIONS.get(node_type, (DEFAULT_NODE_WIDTH, DEFAULT_NODE_HEIGHT)) def _is_placed(node: dict) -> bool: x = node.get("x", 0) y = node.get("y", 0) return not (x == 0 and y == 0) def _node_label(node: dict) -> str: return node.get("name") or node.get("type") or node.get("id", "?") def _segment_intersects_aabb( x1: float, y1: float, x2: float, y2: float, bx: float, by: float, bw: float, bh: float, ) -> bool: dx = x2 - x1 dy = y2 - y1 p = [-dx, dx, -dy, dy] q = [ x1 - (bx - bw), (bx + bw) - x1, y1 - (by - bh), (by + bh) - y1, ] t_enter = 0.0 t_exit = 1.0 for pi, qi in zip(p, q): if pi == 0: if qi < 0: return False else: t = qi / pi if pi < 0: t_enter = max(t_enter, t) else: t_exit = min(t_exit, t) if t_enter > t_exit: return False return True def _lint_nodes(nodes: list[dict], node_ids: list[str] | None = None) -> LayoutReport: placed = [n for n in nodes if _is_placed(n)] scope = set(node_ids) if node_ids else None issues: list[LayoutIssue] = [] overlap_pairs: set[tuple[str, str]] = set() node_dims: dict[str, tuple[int, int]] = {} for n in placed: nid = n.get("id", "") node_dims[nid] = _get_node_dims(n) for i, a in enumerate(placed): aid = a.get("id", "") ax, ay = a.get("x", 0), a.get("y", 0) aw, ah = node_dims[aid] ahw, ahh = aw / 2, ah / 2 for b in placed[i + 1:]: bid = b.get("id", "") if scope and aid not in scope and bid not in scope: continue bx, by = b.get("x", 0), b.get("y", 0) bw, bh = node_dims[bid] bhw, bhh = bw / 2, bh / 2 dx = abs(ax - bx) dy = abs(ay - by) combined_w = ahw + bhw combined_h = ahh + bhh if dx < combined_w and dy < combined_h: pair = (min(aid, bid), max(aid, bid)) overlap_pairs.add(pair) issues.append(LayoutIssue( severity="error", rule="overlap", message=( f"{_node_label(a)} and {_node_label(b)} overlap " f"at ({ax}, {ay})" ), node_ids=[aid, bid], )) elif dx < combined_w + MIN_SPACING and dy < combined_h + MIN_SPACING: gap = min( max(dx - combined_w, 0), max(dy - combined_h, 0), ) issues.append(LayoutIssue( severity="warning", rule="spacing", message=( f"{_node_label(a)} and {_node_label(b)} are only " f"{int(gap)}px apart at ({ax}, {ay})" ), node_ids=[aid, bid], )) node_by_id = {n.get("id", ""): n for n in placed} for node in placed: nid = node.get("id", "") if scope and nid not in scope: continue nw, nh = node_dims.get(nid, (DEFAULT_NODE_WIDTH, DEFAULT_NODE_HEIGHT)) nhw = nw / 2 for output_wires in node.get("wires", []): for target_id in output_wires: target = node_by_id.get(target_id) if not target: continue tw, _ = node_dims.get(target_id, (DEFAULT_NODE_WIDTH, DEFAULT_NODE_HEIGHT)) thw = tw / 2 sx = node.get("x", 0) + nhw sy = node.get("y", 0) tx = target.get("x", 0) - thw ty = target.get("y", 0) for other in placed: oid = other.get("id", "") if oid in (nid, target_id): continue ox = other.get("x", 0) oy = other.get("y", 0) ow, oh = node_dims.get(oid, (DEFAULT_NODE_WIDTH, DEFAULT_NODE_HEIGHT)) ohw, ohh = ow / 2, oh / 2 if _segment_intersects_aabb(sx, sy, tx, ty, ox, oy, ohw, ohh): issues.append(LayoutIssue( severity="warning", rule="wire_crossing", message=( f"Wire from {_node_label(node)} to " f"{_node_label(target)} passes through " f"{_node_label(other)}" ), node_ids=[nid, target_id, oid], )) error_count = sum(1 for i in issues if i.severity == "error") warning_count = sum(1 for i in issues if i.severity == "warning") parts = [] if error_count: parts.append(f"{error_count} error{'s' if error_count != 1 else ''}") if warning_count: parts.append(f"{warning_count} warning{'s' if warning_count != 1 else ''}") summary = ", ".join(parts) if parts else "no issues" return LayoutReport( issues=issues, nodes_checked=len(placed), summary=summary, ) def format_lint_report(report: LayoutReport) -> str: if not report.issues: return "" lines = [f"\nLayout ({report.summary}):"] for issue in report.issues: lines.append(f"- [{issue.severity}] {issue.rule}: {issue.message}") return "\n".join(lines) @mcp.tool() async def lint_flow(flow_id: str, node_ids: str | None = None) -> LayoutReport: client = get_client() raw_flow = await client.get_flow(flow_id) nodes = raw_flow.get("nodes", []) scoped_ids = json.loads(node_ids) if node_ids else None return _lint_nodes(nodes, scoped_ids)