Igor20264
962 lines
33 KiB
Python
962 lines
33 KiB
Python
"""DFD в нотации Гейна–Сарсона (лист Ramus).
|
||
|
||
Оградка ```dfd-gs (алиасы ```gane-sarson, ```uml-dfd-gs, ```uml-gane-sarson).
|
||
Локальный Pillow/SVG, без Graphviz. Йордон остаётся в ```dfd.
|
||
|
||
Раскладка: сущности слева, процессы по центру, хранилища справа.
|
||
Стрелки между соседними колонками идут по зазору. Сквозной поток через
|
||
промежуточную колонку может пересечь блок — обхода препятствий нет.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import math
|
||
import xml.sax.saxutils
|
||
from dataclasses import dataclass, field
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
|
||
from PIL import Image, ImageDraw
|
||
|
||
from .idef0 import _load_font, _text_wh, _wrap
|
||
|
||
_log = logging.getLogger(__name__)
|
||
|
||
CACHE_KEY_PREFIX = "dfd-gs-v1\n"
|
||
|
||
_META = {
|
||
"title": "title",
|
||
"node": "node",
|
||
"number": "number",
|
||
"author": "author",
|
||
"project": "project",
|
||
"date": "date",
|
||
"rev": "rev",
|
||
"revision": "rev",
|
||
"status": "status",
|
||
"used_at": "used_at",
|
||
"used at": "used_at",
|
||
"used-at": "used_at",
|
||
"form": "form",
|
||
}
|
||
_META_ALIASES = sorted(_META, key=len, reverse=True)
|
||
|
||
_KINDS = ("entity", "process", "store")
|
||
_KIND_W = {"entity": 200.0, "process": 248.0, "store": 256.0}
|
||
_KIND_MIN_H = {"entity": 78.0, "process": 96.0, "store": 70.0}
|
||
|
||
|
||
class DfdGsParseError(ValueError):
|
||
"""Неверный DSL DFD Гейна–Сарсона."""
|
||
|
||
|
||
@dataclass
|
||
class GsNode:
|
||
kind: str
|
||
id: str
|
||
label: str
|
||
badge: str = ""
|
||
|
||
|
||
@dataclass
|
||
class GsFlow:
|
||
src: str
|
||
dst: str
|
||
label: str = ""
|
||
|
||
|
||
@dataclass
|
||
class GsDiagram:
|
||
title: str = ""
|
||
node: str = ""
|
||
number: str = ""
|
||
author: str = ""
|
||
project: str = ""
|
||
date: str = ""
|
||
rev: str = ""
|
||
status: str = ""
|
||
used_at: str = ""
|
||
form: str = "kit"
|
||
nodes: list[GsNode] = field(default_factory=list)
|
||
flows: list[GsFlow] = field(default_factory=list)
|
||
|
||
|
||
@dataclass
|
||
class _Placed:
|
||
node: GsNode
|
||
col: int
|
||
x: float
|
||
y: float
|
||
w: float
|
||
h: float
|
||
|
||
@property
|
||
def r(self) -> float:
|
||
return self.x + self.w
|
||
|
||
@property
|
||
def b(self) -> float:
|
||
return self.y + self.h
|
||
|
||
@property
|
||
def cx(self) -> float:
|
||
return self.x + self.w / 2
|
||
|
||
@property
|
||
def cy(self) -> float:
|
||
return self.y + self.h / 2
|
||
|
||
|
||
@dataclass
|
||
class _Route:
|
||
flow: GsFlow
|
||
pts: list[tuple[float, float]]
|
||
|
||
|
||
@dataclass
|
||
class _Layout:
|
||
width: int
|
||
height: int
|
||
frame: tuple[float, float, float, float] | None
|
||
placed: dict[str, _Placed]
|
||
routes: list[_Route]
|
||
|
||
|
||
def _is_proc_num(token: str) -> bool:
|
||
if not token or token[0] not in "0123456789":
|
||
return False
|
||
parts = token.split(".")
|
||
return all(part.isdigit() and part != "" for part in parts)
|
||
|
||
|
||
def _is_store_code(token: str) -> bool:
|
||
if len(token) < 2 or token[0] not in "Dd":
|
||
return False
|
||
return token[1:].isdigit()
|
||
|
||
|
||
def _meta_of(line: str) -> tuple[str, str] | None:
|
||
low = line.lower()
|
||
for alias in _META_ALIASES:
|
||
if low.startswith(alias) and (len(line) == len(alias) or line[len(alias)].isspace()):
|
||
return _META[alias], line[len(alias):].strip()
|
||
return None
|
||
|
||
|
||
def _assign_badges(nodes: list[GsNode]) -> None:
|
||
processes = [n for n in nodes if n.kind == "process"]
|
||
for index, node in enumerate(processes, 1):
|
||
node.badge = node.id if _is_proc_num(node.id) else str(index)
|
||
|
||
stores = [n for n in nodes if n.kind == "store"]
|
||
used = {n.id.upper() for n in stores if _is_store_code(n.id)}
|
||
seq = 1
|
||
for node in stores:
|
||
if _is_store_code(node.id):
|
||
node.badge = node.id.upper()
|
||
continue
|
||
while f"D{seq}" in used:
|
||
seq += 1
|
||
node.badge = f"D{seq}"
|
||
used.add(node.badge)
|
||
seq += 1
|
||
|
||
entities = [n for n in nodes if n.kind == "entity"]
|
||
for index, node in enumerate(entities, 1):
|
||
node.badge = str(index)
|
||
|
||
|
||
def parse_dfd_gs(source: str) -> GsDiagram:
|
||
"""Разобрать DSL. Пустой текст и неизвестные узлы — DfdGsParseError."""
|
||
text = (source or "").lstrip("\ufeff")
|
||
if not text.strip():
|
||
raise DfdGsParseError("пустой DFD")
|
||
|
||
diagram = GsDiagram()
|
||
seen: dict[str, int] = {}
|
||
for lineno, raw in enumerate(text.splitlines(), 1):
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or line.startswith("//"):
|
||
continue
|
||
|
||
kind = next((k for k in _KINDS if line.lower().startswith(k) and (
|
||
len(line) == len(k) or line[len(k)].isspace()
|
||
)), "")
|
||
if kind:
|
||
rest = line[len(kind):].strip()
|
||
parts = rest.split(None, 1)
|
||
if len(parts) < 2 or not parts[1].strip():
|
||
raise DfdGsParseError(f"строка {lineno}: у {kind} нет подписи")
|
||
node_id, label = parts[0], parts[1].strip()
|
||
if node_id in seen:
|
||
raise DfdGsParseError(
|
||
f"строка {lineno}: повтор идентификатора «{node_id}»"
|
||
)
|
||
seen[node_id] = lineno
|
||
diagram.nodes.append(GsNode(kind, node_id, label))
|
||
continue
|
||
|
||
if "-->" in line:
|
||
left, _, right = line.partition("-->")
|
||
src = left.strip()
|
||
tail = right.strip()
|
||
if not src or not tail:
|
||
raise DfdGsParseError(f"строка {lineno}: неполный поток")
|
||
dst, _, label = tail.partition(" ")
|
||
label = label.strip()
|
||
if label.startswith(":"):
|
||
label = label[1:].strip()
|
||
if not dst:
|
||
raise DfdGsParseError(f"строка {lineno}: неполный поток")
|
||
diagram.flows.append(GsFlow(src, dst, label))
|
||
continue
|
||
|
||
meta = _meta_of(line)
|
||
if meta:
|
||
key, value = meta
|
||
if key == "form":
|
||
form = value.lower()
|
||
if form not in {"kit", "plain"}:
|
||
raise DfdGsParseError(
|
||
f"строка {lineno}: form «{value}» — ожидается kit или plain"
|
||
)
|
||
diagram.form = form
|
||
else:
|
||
setattr(diagram, key, value)
|
||
continue
|
||
|
||
raise DfdGsParseError(f"строка {lineno}: неизвестная запись «{line}»")
|
||
|
||
if not diagram.nodes:
|
||
raise DfdGsParseError("DFD без узлов")
|
||
|
||
for flow in diagram.flows:
|
||
if flow.src == flow.dst:
|
||
raise DfdGsParseError(f"поток из «{flow.src}» в себя")
|
||
for end in (flow.src, flow.dst):
|
||
if end not in seen:
|
||
raise DfdGsParseError(f"неизвестный узел «{end}»")
|
||
|
||
_assign_badges(diagram.nodes)
|
||
return diagram
|
||
|
||
|
||
# --- scene -----------------------------------------------------------------
|
||
|
||
def _svg_anchor(anchor: str) -> tuple[str, str, str]:
|
||
a = (anchor or "lt").lower()
|
||
horiz = a[0] if a else "l"
|
||
vert = a[1] if len(a) > 1 else "t"
|
||
ta = {"l": "start", "m": "middle", "r": "end"}.get(horiz, "start")
|
||
dy = {"t": "0.9em", "m": "0.35em", "b": "0"}.get(vert, "0.9em")
|
||
return ta, "0", dy
|
||
|
||
|
||
class _Scene:
|
||
def __init__(self, width: int, height: int):
|
||
self.width = width
|
||
self.height = height
|
||
self.ops: list[tuple] = []
|
||
|
||
def line(self, x1, y1, x2, y2, width: float = 1.4) -> None:
|
||
self.ops.append(("line", float(x1), float(y1), float(x2), float(y2), float(width)))
|
||
|
||
def polyline(self, pts: list[tuple[float, float]], width: float = 1.4) -> None:
|
||
if len(pts) < 2:
|
||
return
|
||
self.ops.append(("polyline", [(float(x), float(y)) for x, y in pts], float(width)))
|
||
|
||
def rect(
|
||
self,
|
||
x, y, w, h,
|
||
width: float = 1.4,
|
||
fill: str | None = None,
|
||
outline: str | None = "#000000",
|
||
) -> None:
|
||
self.ops.append(
|
||
("rect", float(x), float(y), float(w), float(h), float(width), fill, outline)
|
||
)
|
||
|
||
def roundrect(
|
||
self, x, y, w, h, radius: float,
|
||
width: float = 1.4,
|
||
fill: str | None = "#ffffff",
|
||
outline: str | None = "#000000",
|
||
) -> None:
|
||
self.ops.append((
|
||
"roundrect", float(x), float(y), float(w), float(h), float(radius),
|
||
float(width), fill, outline,
|
||
))
|
||
|
||
def polygon(self, pts: list[tuple[float, float]], fill: str = "#000000") -> None:
|
||
if len(pts) < 3:
|
||
return
|
||
self.ops.append(("polygon", [(float(x), float(y)) for x, y in pts], fill))
|
||
|
||
def text(
|
||
self, x, y, text: str, size: float, *,
|
||
bold: bool = False, anchor: str = "lt", fill: str = "#000000",
|
||
) -> None:
|
||
if not text:
|
||
return
|
||
self.ops.append(("text", float(x), float(y), text, float(size), bold, anchor, fill))
|
||
|
||
def text_rot(
|
||
self, x, y, text: str, size: float, *,
|
||
bold: bool = False, angle: float = 90,
|
||
) -> None:
|
||
if not text:
|
||
return
|
||
self.ops.append(("text_rot", float(x), float(y), text, float(size), bold, float(angle)))
|
||
|
||
def to_png(self) -> bytes:
|
||
img = Image.new("RGB", (self.width, self.height), "#ffffff")
|
||
draw = ImageDraw.Draw(img)
|
||
for op in self.ops:
|
||
kind = op[0]
|
||
if kind == "line":
|
||
_, x1, y1, x2, y2, w = op
|
||
draw.line([(x1, y1), (x2, y2)], fill="#000000", width=max(1, int(round(w))))
|
||
elif kind == "polyline":
|
||
_, pts, w = op
|
||
draw.line(pts, fill="#000000", width=max(1, int(round(w))))
|
||
elif kind == "rect":
|
||
_, x, y, w, h, lw, fill, outline = op
|
||
stroke = max(1, int(round(lw))) if outline else 0
|
||
draw.rectangle([x, y, x + w, y + h], fill=fill, outline=outline, width=stroke)
|
||
elif kind == "roundrect":
|
||
_, x, y, w, h, radius, lw, fill, outline = op
|
||
stroke = max(1, int(round(lw))) if outline else 0
|
||
rad = max(0, min(radius, w / 2 - 1, h / 2 - 1))
|
||
draw.rounded_rectangle(
|
||
[x, y, x + w, y + h], radius=rad, fill=fill, outline=outline, width=stroke,
|
||
)
|
||
elif kind == "polygon":
|
||
_, pts, fill = op
|
||
draw.polygon(pts, fill=fill)
|
||
elif kind == "text":
|
||
_, x, y, text, size, bold, anchor, fill = op
|
||
font = _load_font(int(round(size)), bold)
|
||
draw.text((x, y), text, font=font, fill=fill, anchor=anchor)
|
||
elif kind == "text_rot":
|
||
_, x, y, text, size, bold, angle = op
|
||
_paste_rotated(img, text, x, y, int(round(size)), bold, angle)
|
||
buf = BytesIO()
|
||
img.save(buf, format="PNG", optimize=True)
|
||
return buf.getvalue()
|
||
|
||
def to_svg(self) -> str:
|
||
esc = xml.sax.saxutils.escape
|
||
parts = [
|
||
f'<svg xmlns="http://www.w3.org/2000/svg" width="{self.width}" height="{self.height}" '
|
||
f'viewBox="0 0 {self.width} {self.height}">',
|
||
'<rect width="100%" height="100%" fill="#fff"/>',
|
||
]
|
||
for op in self.ops:
|
||
kind = op[0]
|
||
if kind == "line":
|
||
_, x1, y1, x2, y2, w = op
|
||
parts.append(
|
||
f'<line x1="{x1:.1f}" y1="{y1:.1f}" x2="{x2:.1f}" y2="{y2:.1f}" '
|
||
f'stroke="#000" stroke-width="{w:.2f}"/>'
|
||
)
|
||
elif kind == "polyline":
|
||
_, pts, w = op
|
||
d = " ".join(f"{x:.1f},{y:.1f}" for x, y in pts)
|
||
parts.append(
|
||
f'<polyline points="{d}" fill="none" stroke="#000" '
|
||
f'stroke-width="{w:.2f}" stroke-linejoin="miter"/>'
|
||
)
|
||
elif kind == "rect":
|
||
_, x, y, w, h, lw, fill, outline = op
|
||
parts.append(
|
||
f'<rect x="{x:.1f}" y="{y:.1f}" width="{w:.1f}" height="{h:.1f}" '
|
||
f'fill="{fill or "none"}" stroke="{outline or "none"}" '
|
||
f'stroke-width="{0 if outline is None else lw:.2f}"/>'
|
||
)
|
||
elif kind == "roundrect":
|
||
_, x, y, w, h, radius, lw, fill, outline = op
|
||
parts.append(
|
||
f'<rect x="{x:.1f}" y="{y:.1f}" width="{w:.1f}" height="{h:.1f}" '
|
||
f'rx="{radius:.1f}" ry="{radius:.1f}" fill="{fill or "none"}" '
|
||
f'stroke="{outline or "none"}" '
|
||
f'stroke-width="{0 if outline is None else lw:.2f}"/>'
|
||
)
|
||
elif kind == "polygon":
|
||
_, pts, fill = op
|
||
d = " ".join(f"{x:.1f},{y:.1f}" for x, y in pts)
|
||
parts.append(f'<polygon points="{d}" fill="{fill}"/>')
|
||
elif kind == "text":
|
||
_, x, y, text, size, bold, anchor, fill = op
|
||
ta, dx, dy = _svg_anchor(anchor)
|
||
weight = "700" if bold else "400"
|
||
parts.append(
|
||
f'<text x="{x:.1f}" y="{y:.1f}" dx="{dx}" dy="{dy}" text-anchor="{ta}" '
|
||
f'font-family="Arial, DejaVu Sans, sans-serif" font-size="{size:.1f}" '
|
||
f'font-weight="{weight}" fill="{fill}">{esc(text)}</text>'
|
||
)
|
||
elif kind == "text_rot":
|
||
_, x, y, text, size, bold, angle = op
|
||
weight = "700" if bold else "400"
|
||
parts.append(
|
||
f'<text x="{x:.1f}" y="{y:.1f}" text-anchor="middle" '
|
||
f'dominant-baseline="middle" '
|
||
f'transform="rotate({-angle:.0f} {x:.1f} {y:.1f})" '
|
||
f'font-family="Arial, DejaVu Sans, sans-serif" font-size="{size:.1f}" '
|
||
f'font-weight="{weight}" fill="#000">{esc(text)}</text>'
|
||
)
|
||
parts.append("</svg>")
|
||
return "\n".join(parts) + "\n"
|
||
|
||
|
||
def _paste_rotated(
|
||
base: Image.Image, text: str, cx: float, cy: float,
|
||
size: int, bold: bool, angle: float,
|
||
) -> None:
|
||
font = _load_font(size, bold)
|
||
left, top, right, bottom = font.getbbox(text)
|
||
pad = 2
|
||
patch = Image.new("RGBA", (right - left + pad * 2, bottom - top + pad * 2), (0, 0, 0, 0))
|
||
ImageDraw.Draw(patch).text((pad - left, pad - top), text, font=font, fill=(0, 0, 0, 255))
|
||
rotated = patch.rotate(angle, expand=True, resample=Image.Resampling.BICUBIC)
|
||
base.paste(
|
||
rotated,
|
||
(int(round(cx - rotated.width / 2)), int(round(cy - rotated.height / 2))),
|
||
rotated,
|
||
)
|
||
|
||
|
||
# --- layout ----------------------------------------------------------------
|
||
|
||
def _node_size(node: GsNode, scale: float) -> tuple[float, float]:
|
||
s = scale
|
||
width = _KIND_W[node.kind] * s
|
||
font = _load_font(int(round(15 * s)), False)
|
||
if node.kind == "store":
|
||
inner = width - 44 * s
|
||
else:
|
||
inner = width - 28 * s
|
||
lines = _wrap(node.label, font, max(40, int(inner)))
|
||
line_h = 18 * s
|
||
text_h = max(1, len(lines)) * line_h
|
||
extra = 30 * s if node.kind == "process" else 22 * s
|
||
height = max(_KIND_MIN_H[node.kind] * s, text_h + extra)
|
||
return width, height
|
||
|
||
|
||
def _simplify(pts: list[tuple[float, float]]) -> list[tuple[float, float]]:
|
||
out: list[tuple[float, float]] = []
|
||
for point in pts:
|
||
if out and abs(point[0] - out[-1][0]) < 0.4 and abs(point[1] - out[-1][1]) < 0.4:
|
||
continue
|
||
if len(out) >= 2:
|
||
a, b = out[-2], out[-1]
|
||
horiz = abs(a[1] - b[1]) < 0.4 and abs(b[1] - point[1]) < 0.4
|
||
vert = abs(a[0] - b[0]) < 0.4 and abs(b[0] - point[0]) < 0.4
|
||
if horiz or vert:
|
||
out[-1] = point
|
||
continue
|
||
out.append(point)
|
||
return out
|
||
|
||
|
||
def _slot_xs(left: float, right: float, n: int) -> list[float]:
|
||
if n <= 0:
|
||
return []
|
||
if right < left:
|
||
left, right = right, left
|
||
span = right - left
|
||
margin = min(6.0, span / 4) if span else 0.0
|
||
a, b = left + margin, right - margin
|
||
if b <= a or n == 1:
|
||
return [(left + right) / 2] * n
|
||
return [a + (b - a) * (k + 1) / (n + 1) for k in range(n)]
|
||
|
||
|
||
def _arrange(diagram: GsDiagram, scale: float) -> _Layout:
|
||
s = max(float(scale), 1.0)
|
||
groups: list[list[GsNode]] = []
|
||
for kind in _KINDS:
|
||
chunk = [n for n in diagram.nodes if n.kind == kind]
|
||
if chunk:
|
||
groups.append(chunk)
|
||
|
||
sizes = {n.id: _node_size(n, s) for n in diagram.nodes}
|
||
# Vertical gap fits a one-line label between stacked boxes.
|
||
gap_y = 50 * s
|
||
# Horizontal gap fits a wrapped flow label without covering the boxes.
|
||
gap_x = 132 * s
|
||
col_w = [max(sizes[n.id][0] for n in group) for group in groups]
|
||
col_h = []
|
||
for group in groups:
|
||
heights = [sizes[n.id][1] for n in group]
|
||
col_h.append(sum(heights) + gap_y * (len(group) - 1))
|
||
|
||
col_of = {n.id: i for i, group in enumerate(groups) for n in group}
|
||
index_in_col = {
|
||
node.id: (col_i, pos)
|
||
for col_i, group in enumerate(groups)
|
||
for pos, node in enumerate(group)
|
||
}
|
||
|
||
def _neighbors(flow: GsFlow) -> bool:
|
||
a, b = index_in_col[flow.src], index_in_col[flow.dst]
|
||
return a[0] == b[0] and abs(a[1] - b[1]) == 1
|
||
|
||
last = len(groups) - 1
|
||
needs_channel = any(
|
||
col_of[flow.src] == col_of[flow.dst] == last and not _neighbors(flow)
|
||
for flow in diagram.flows
|
||
)
|
||
extra_r = 56 * s if needs_channel else 0.0
|
||
|
||
content_w = sum(col_w) + gap_x * (len(groups) - 1) + extra_r
|
||
content_h = max(col_h) if col_h else 0.0
|
||
inner_pad = 24 * s
|
||
outer = 8 * s
|
||
kit = diagram.form != "plain"
|
||
header_h = 76 * s if kit else 0.0
|
||
footer_h = 56 * s if kit else 0.0
|
||
need_w = content_w + 2 * inner_pad
|
||
need_h = content_h + 2 * inner_pad
|
||
# Stamp needs a readable width; height follows the columns, not a tall empty sheet.
|
||
width = max((760 if kit else 480) * s, need_w + 2 * outer)
|
||
height = need_h + header_h + footer_h + 2 * outer
|
||
width_i, height_i = int(round(width)), int(round(height))
|
||
|
||
frame = (outer, outer, width - 2 * outer, height - 2 * outer)
|
||
if kit:
|
||
work_x, work_y = frame[0], frame[1] + header_h
|
||
work_w, work_h = frame[2], frame[3] - header_h - footer_h
|
||
else:
|
||
work_x, work_y, work_w, work_h = frame
|
||
|
||
block_w = content_w
|
||
ox = work_x + (work_w - block_w) / 2
|
||
oy = work_y + max(inner_pad, (work_h - content_h) / 2)
|
||
|
||
placed: dict[str, _Placed] = {}
|
||
col_box: list[tuple[float, float]] = []
|
||
x = ox
|
||
for index, group in enumerate(groups):
|
||
col_box.append((x, col_w[index]))
|
||
y = oy + (content_h - col_h[index]) / 2
|
||
for node in group:
|
||
nh = sizes[node.id][1]
|
||
placed[node.id] = _Placed(node, index, x, y, col_w[index], nh)
|
||
y += nh + gap_y
|
||
x += col_w[index] + gap_x
|
||
|
||
gaps = [
|
||
(col_box[i][0] + col_box[i][1], col_box[i + 1][0])
|
||
for i in range(len(col_box) - 1)
|
||
]
|
||
|
||
# Side each flow touches. Ports on one side are spread together, so a
|
||
# pair of opposite arrows does not share the same point.
|
||
# Neighbours stacked in one column are linked through the gap between them.
|
||
touch: list[tuple[str, str]] = []
|
||
side_hits: dict[tuple[str, str], list[tuple[int, str]]] = {}
|
||
neighbor_flows: list[int] = []
|
||
for i, flow in enumerate(diagram.flows):
|
||
if _neighbors(flow):
|
||
touch.append(("v", "v"))
|
||
neighbor_flows.append(i)
|
||
continue
|
||
sc, dc = col_of[flow.src], col_of[flow.dst]
|
||
if sc < dc:
|
||
sides = ("right", "left")
|
||
elif sc > dc:
|
||
sides = ("left", "right")
|
||
else:
|
||
sides = ("right", "right")
|
||
touch.append(sides)
|
||
side_hits.setdefault((flow.src, sides[0]), []).append((i, "src"))
|
||
side_hits.setdefault((flow.dst, sides[1]), []).append((i, "dst"))
|
||
|
||
neighbor_dx: dict[int, float] = {}
|
||
neighbor_groups: dict[tuple[int, int, int], list[int]] = {}
|
||
for i in neighbor_flows:
|
||
flow = diagram.flows[i]
|
||
col_i, pos_a = index_in_col[flow.src]
|
||
_col_b, pos_b = index_in_col[flow.dst]
|
||
key = (col_i, min(pos_a, pos_b), max(pos_a, pos_b))
|
||
neighbor_groups.setdefault(key, []).append(i)
|
||
for idxs in neighbor_groups.values():
|
||
count = len(idxs)
|
||
for k, fi in enumerate(idxs):
|
||
neighbor_dx[fi] = (k - (count - 1) / 2) * 18 * s
|
||
|
||
port_y: dict[tuple[int, str], float] = {}
|
||
for (nid, _side), hits in side_hits.items():
|
||
box = placed[nid]
|
||
ordered = sorted(
|
||
hits,
|
||
key=lambda hit: (
|
||
placed[
|
||
diagram.flows[hit[0]].dst
|
||
if hit[1] == "src"
|
||
else diagram.flows[hit[0]].src
|
||
].cy,
|
||
hit[0],
|
||
),
|
||
)
|
||
n = len(ordered)
|
||
for k, (fi, role) in enumerate(ordered):
|
||
y = box.cy if n == 1 else box.y + box.h * (k + 1) / (n + 1)
|
||
port_y[(fi, role)] = y
|
||
|
||
thresh = 10 * s
|
||
elbow_gap: dict[tuple, list[int]] = {}
|
||
for i, flow in enumerate(diagram.flows):
|
||
if i in neighbor_dx:
|
||
continue
|
||
sc, dc = col_of[flow.src], col_of[flow.dst]
|
||
y1, y2 = port_y[(i, "src")], port_y[(i, "dst")]
|
||
if sc != dc and abs(y1 - y2) <= thresh:
|
||
continue
|
||
if sc == dc:
|
||
elbow_gap.setdefault(("same", sc), []).append(i)
|
||
else:
|
||
gap_i = (dc - 1) if sc < dc else (sc - 1)
|
||
elbow_gap.setdefault(("gap", gap_i), []).append(i)
|
||
|
||
gutter_x: dict[int, float] = {}
|
||
for key, idxs in elbow_gap.items():
|
||
ordered = sorted(idxs, key=lambda fi: (port_y[(fi, "src")], fi))
|
||
kind, index = key
|
||
if kind == "gap":
|
||
left, right = gaps[index]
|
||
else:
|
||
left = col_box[index][0] + col_box[index][1] + 8 * s
|
||
right = left + (40 * s if index == last else gap_x - 16 * s)
|
||
for fi, gx in zip(ordered, _slot_xs(left, right, len(ordered))):
|
||
gutter_x[fi] = gx
|
||
|
||
routes: list[_Route] = []
|
||
for i, flow in enumerate(diagram.flows):
|
||
src, dst = placed[flow.src], placed[flow.dst]
|
||
if i in neighbor_dx:
|
||
x = src.cx + neighbor_dx[i]
|
||
if src.y < dst.y:
|
||
pts = [(x, src.b), (x, dst.y)]
|
||
else:
|
||
pts = [(x, src.y), (x, dst.b)]
|
||
routes.append(_Route(flow, pts))
|
||
continue
|
||
src_side, dst_side = touch[i]
|
||
x1 = src.r if src_side == "right" else src.x
|
||
x2 = dst.r if dst_side == "right" else dst.x
|
||
y1, y2 = port_y[(i, "src")], port_y[(i, "dst")]
|
||
if src.col != dst.col and abs(y1 - y2) <= thresh:
|
||
pts = [(x1, y1), (x2, y2)]
|
||
else:
|
||
gx = gutter_x[i]
|
||
pts = _simplify([(x1, y1), (gx, y1), (gx, y2), (x2, y2)])
|
||
routes.append(_Route(flow, pts))
|
||
|
||
return _Layout(
|
||
width_i,
|
||
height_i,
|
||
frame if kit else None,
|
||
placed,
|
||
routes,
|
||
)
|
||
|
||
|
||
# --- draw ------------------------------------------------------------------
|
||
|
||
def _draw_centered(
|
||
scene: _Scene, text: str, x: float, y: float, w: float, h: float,
|
||
size: float, max_width: float,
|
||
) -> None:
|
||
font = _load_font(int(round(size)), False)
|
||
lines = _wrap(text, font, max(20, int(max_width)))
|
||
line_h = size * 1.2
|
||
block = line_h * len(lines)
|
||
top = y + (h - block) / 2
|
||
for i, line in enumerate(lines):
|
||
scene.text(x + w / 2, top + i * line_h + line_h / 2, line, size, anchor="mm")
|
||
|
||
|
||
def _draw_node(scene: _Scene, box: _Placed, scale: float, lw: float) -> None:
|
||
s = scale
|
||
node = box.node
|
||
if node.kind == "entity":
|
||
sh = 5 * s
|
||
scene.rect(box.x + sh, box.y + sh, box.w, box.h, width=0, fill="#c4c4c4", outline=None)
|
||
scene.rect(box.x, box.y, box.w, box.h, width=lw, fill="#ffffff")
|
||
scene.text(box.x + 8 * s, box.y + 6 * s, node.badge, 13 * s, bold=True, anchor="lt")
|
||
_draw_centered(
|
||
scene, node.label, box.x, box.y + 16 * s, box.w, box.h - 22 * s,
|
||
15 * s, box.w - 28 * s,
|
||
)
|
||
elif node.kind == "process":
|
||
radius = min(18 * s, box.h * 0.2, box.w * 0.12)
|
||
scene.roundrect(box.x, box.y, box.w, box.h, radius, width=lw, fill="#ffffff")
|
||
_draw_centered(
|
||
scene, node.label, box.x, box.y + 4 * s, box.w, box.h - 24 * s,
|
||
15 * s, box.w - 28 * s,
|
||
)
|
||
scene.text(box.r - 8 * s, box.b - 6 * s, node.badge, 14 * s, bold=True, anchor="rb")
|
||
else:
|
||
scene.rect(box.x, box.y, box.w, box.h, width=lw, fill="#ffffff")
|
||
strip = 28 * s
|
||
scene.line(box.x + strip, box.y, box.x + strip, box.b, lw)
|
||
scene.text_rot(box.x + strip / 2, box.cy, node.badge, 13 * s, bold=True, angle=90)
|
||
_draw_centered(
|
||
scene, node.label, box.x + strip, box.y, box.w - strip, box.h,
|
||
15 * s, box.w - strip - 16 * s,
|
||
)
|
||
|
||
|
||
def _draw_arrow(scene: _Scene, pts: list[tuple[float, float]], lw: float, scale: float) -> None:
|
||
if len(pts) < 2:
|
||
return
|
||
p1, p2 = pts[-2], pts[-1]
|
||
dx, dy = p2[0] - p1[0], p2[1] - p1[1]
|
||
length = math.hypot(dx, dy) or 1.0
|
||
size = min(12 * scale, length * 0.55)
|
||
ux, uy = dx / length, dy / length
|
||
base = (p2[0] - ux * size, p2[1] - uy * size)
|
||
shaft = list(pts[:-1]) + [base]
|
||
scene.polyline(shaft, lw)
|
||
wing = size * 0.42
|
||
px, py = -uy * wing, ux * wing
|
||
scene.polygon([p2, (base[0] + px, base[1] + py), (base[0] - px, base[1] - py)])
|
||
|
||
|
||
def _point_along(pts: list[tuple[float, float]], frac: float) -> tuple[float, float, float, float]:
|
||
segs = []
|
||
total = 0.0
|
||
for a, b in zip(pts, pts[1:]):
|
||
length = math.hypot(b[0] - a[0], b[1] - a[1])
|
||
segs.append((a, b, length))
|
||
total += length
|
||
if total <= 0:
|
||
return pts[0][0], pts[0][1], 0.0, -1.0
|
||
target = total * min(1.0, max(0.0, frac))
|
||
acc = 0.0
|
||
a, b = segs[-1][0], segs[-1][1]
|
||
for a, b, length in segs:
|
||
if acc + length >= target or length == 0:
|
||
t = 0 if length == 0 else (target - acc) / length
|
||
x = a[0] + (b[0] - a[0]) * t
|
||
y = a[1] + (b[1] - a[1]) * t
|
||
dx, dy = b[0] - a[0], b[1] - a[1]
|
||
norm = math.hypot(dx, dy) or 1.0
|
||
px, py = -dy / norm, dx / norm
|
||
if py > 0 or (abs(py) < 1e-6 and px < 0):
|
||
px, py = -px, -py
|
||
return x, y, px, py
|
||
acc += length
|
||
return a[0], a[1], 0.0, -1.0
|
||
|
||
|
||
def _label_overlap(
|
||
cx: float, cy: float, bw: float, bh: float,
|
||
boxes: list[_Placed],
|
||
busy: list[tuple[float, float, float, float]],
|
||
pad: float,
|
||
) -> float:
|
||
ax, ay = cx - bw / 2 - pad, cy - bh / 2 - pad
|
||
ar, ab = ax + bw + 2 * pad, ay + bh + 2 * pad
|
||
total = 0.0
|
||
for box in boxes:
|
||
ix = max(0.0, min(ar, box.r) - max(ax, box.x))
|
||
iy = max(0.0, min(ab, box.b) - max(ay, box.y))
|
||
total += ix * iy * 4
|
||
for x, y, w, h in busy:
|
||
ix = max(0.0, min(ar, x + w) - max(ax, x))
|
||
iy = max(0.0, min(ab, y + h) - max(ay, y))
|
||
total += ix * iy
|
||
return total
|
||
|
||
|
||
def _clamp_label(
|
||
cx: float, cy: float, bw: float, bh: float,
|
||
canvas_w: float, canvas_h: float, pad: float,
|
||
) -> tuple[float, float]:
|
||
half_w, half_h = bw / 2 + pad, bh / 2 + pad
|
||
cx = min(max(cx, half_w), max(half_w, canvas_w - half_w))
|
||
cy = min(max(cy, half_h), max(half_h, canvas_h - half_h))
|
||
return cx, cy
|
||
|
||
|
||
def _draw_label(
|
||
scene: _Scene,
|
||
text: str,
|
||
pts: list[tuple[float, float]],
|
||
scale: float,
|
||
boxes: list[_Placed],
|
||
canvas_w: float,
|
||
canvas_h: float,
|
||
busy: list[tuple[float, float, float, float]],
|
||
) -> None:
|
||
label = (text or "").strip()
|
||
if not label or len(pts) < 2:
|
||
return
|
||
s = scale
|
||
size = 13 * s
|
||
font = _load_font(int(round(size)), False)
|
||
# Narrow wrap: the label stays in the column gap instead of covering a box.
|
||
lines = _wrap(label, font, max(72, int(108 * s)))
|
||
line_h = size * 1.2
|
||
block_w = max(_text_wh(font, line)[0] for line in lines)
|
||
block_h = line_h * len(lines)
|
||
pad = 3 * s
|
||
best: tuple[float, float, float] | None = None
|
||
for frac in (0.5, 0.34, 0.66, 0.2, 0.8):
|
||
x, y, px, py = _point_along(pts, frac)
|
||
horizontal = abs(py) >= abs(px)
|
||
shift = (5 * s + block_h / 2) if horizontal else (8 * s)
|
||
for sign in (1.0, -1.0):
|
||
cx, cy = _clamp_label(
|
||
x + px * shift * sign, y + py * shift * sign,
|
||
block_w, block_h, canvas_w, canvas_h, pad,
|
||
)
|
||
overlap = _label_overlap(cx, cy, block_w, block_h, boxes, busy, 1.0)
|
||
rank = overlap + (0 if frac == 0.5 and sign > 0 else 0.01)
|
||
if best is None or rank < best[0]:
|
||
best = (rank, cx, cy)
|
||
if overlap <= 0 and frac == 0.5 and sign > 0:
|
||
best = (0.0, cx, cy)
|
||
break
|
||
if best is not None and best[0] <= 0:
|
||
break
|
||
if best is None:
|
||
return
|
||
cx, cy = best[1], best[2]
|
||
scene.rect(
|
||
cx - block_w / 2 - pad, cy - block_h / 2 - pad,
|
||
block_w + 2 * pad, block_h + 2 * pad,
|
||
width=0, fill="#ffffff", outline=None,
|
||
)
|
||
top = cy - block_h / 2
|
||
busy.append((cx - block_w / 2 - pad, top - pad, block_w + 2 * pad, block_h + 2 * pad))
|
||
for i, line in enumerate(lines):
|
||
scene.text(cx, top + i * line_h + line_h / 2, line, size, anchor="mm")
|
||
|
||
|
||
def _cell_text(
|
||
scene: _Scene, x: float, y: float, w: float, h: float,
|
||
caption: str, value: str, scale: float,
|
||
) -> None:
|
||
s = scale
|
||
scene.text(x + 6 * s, y + 4 * s, caption, 9 * s, anchor="lt", fill="#444444")
|
||
if not value:
|
||
return
|
||
font = _load_font(int(round(12 * s)), False)
|
||
lines = _wrap(value, font, max(20, int(w - 12 * s)))[:2]
|
||
line_h = 14 * s
|
||
for i, line in enumerate(lines):
|
||
scene.text(x + w / 2, y + 26 * s + i * line_h, line, 12 * s, anchor="mt")
|
||
|
||
|
||
def _draw_frame(scene: _Scene, diagram: GsDiagram, frame: tuple[float, float, float, float], scale: float) -> None:
|
||
s = scale
|
||
x, y, w, h = frame
|
||
lw = max(1.2, 1.3 * s)
|
||
scene.rect(x, y, w, h, width=lw, fill=None)
|
||
header_h = 76 * s
|
||
footer_h = 56 * s
|
||
scene.line(x, y + header_h, x + w, y + header_h, lw)
|
||
scene.line(x, y + h - footer_h, x + w, y + h - footer_h, lw)
|
||
|
||
fracs = (0.14, 0.18, 0.14, 0.08, 0.28, 0.18)
|
||
cells = (
|
||
("USED AT", diagram.used_at),
|
||
("AUTHOR", diagram.author),
|
||
("DATE", diagram.date),
|
||
("REV", diagram.rev),
|
||
("PROJECT", diagram.project),
|
||
("STATUS", diagram.status),
|
||
)
|
||
cursor = x
|
||
for (caption, value), frac in zip(cells, fracs):
|
||
cw = w * frac
|
||
scene.line(cursor, y, cursor, y + header_h, lw)
|
||
_cell_text(scene, cursor, y, cw, header_h, caption, value, s)
|
||
cursor += cw
|
||
|
||
num_w = 130 * s
|
||
scene.line(x + w - num_w, y + h - footer_h, x + w - num_w, y + h, lw)
|
||
scene.text(x + 8 * s, y + h - footer_h + 4 * s, "NODE / TITLE", 9 * s, anchor="lt", fill="#444444")
|
||
title = " ".join(part for part in (diagram.node, diagram.title) if part)
|
||
if title:
|
||
font = _load_font(int(round(13 * s)), True)
|
||
lines = _wrap(title, font, max(40, int(w - num_w - 16 * s)))[:2]
|
||
for i, line in enumerate(lines):
|
||
scene.text(x + 8 * s, y + h - footer_h + 24 * s + i * 16 * s, line, 13 * s, bold=True, anchor="lt")
|
||
scene.text(
|
||
x + w - num_w / 2, y + h - footer_h + 8 * s, "NUMBER", 9 * s, anchor="mt", fill="#444444",
|
||
)
|
||
if diagram.number:
|
||
scene.text(x + w - num_w / 2, y + h - footer_h / 2 + 6 * s, diagram.number, 14 * s, bold=True, anchor="mm")
|
||
|
||
|
||
def render_dfd_gs(
|
||
diagram: GsDiagram,
|
||
*,
|
||
scale: float = 2.0,
|
||
want_svg: bool = True,
|
||
) -> tuple[bytes, str | None]:
|
||
"""Вернуть (png_bytes, svg или None)."""
|
||
s = max(float(scale), 1.0)
|
||
layout = _arrange(diagram, s)
|
||
scene = _Scene(layout.width, layout.height)
|
||
lw = max(1.2, 1.35 * s)
|
||
if layout.frame is not None:
|
||
_draw_frame(scene, diagram, layout.frame, s)
|
||
for box in layout.placed.values():
|
||
_draw_node(scene, box, s, lw)
|
||
for route in layout.routes:
|
||
_draw_arrow(scene, route.pts, lw, s)
|
||
busy: list[tuple[float, float, float, float]] = []
|
||
for route in layout.routes:
|
||
_draw_label(
|
||
scene, route.flow.label, route.pts, s,
|
||
list(layout.placed.values()), layout.width, layout.height, busy,
|
||
)
|
||
svg = scene.to_svg() if want_svg else None
|
||
return scene.to_png(), svg
|
||
|
||
|
||
def write_dfd_gs_outputs(
|
||
source: str,
|
||
out_png: Path,
|
||
out_svg: Path | None,
|
||
*,
|
||
scale: float = 2.0,
|
||
) -> bool:
|
||
"""Разобрать и нарисовать. DfdGsParseError при ошибке DSL."""
|
||
diagram = parse_dfd_gs(source)
|
||
png, svg = render_dfd_gs(diagram, scale=scale, want_svg=out_svg is not None)
|
||
out_png.parent.mkdir(parents=True, exist_ok=True)
|
||
out_png.write_bytes(png)
|
||
if out_svg is not None and svg is not None:
|
||
out_svg.write_text(svg, encoding="utf-8")
|
||
_log.info("DFD Гейна–Сарсона → %s", out_png.name)
|
||
return True
|
||
|
||
|
||
SAMPLE = """\
|
||
title Зафиксировать решение приёмки
|
||
node A44
|
||
number 1
|
||
author Тарасов И.А.
|
||
project ИС ArtFreelance
|
||
date 05.10.2026
|
||
rev 01
|
||
status WORKING
|
||
used_at учебный проект
|
||
|
||
entity Artist Художник
|
||
process 1 Принять файлы результата
|
||
store D2 Файлы работ
|
||
|
||
Artist --> 1 файлы работы
|
||
1 --> D2 превью и хеш
|
||
"""
|