AG_Prozesse/Projektwissen
6
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
96225b5e0226c860e35b7d9ff0263911415ea667
Projektwissen/.agents/skills/sn-bpmn/md_to_bpmn.py
T
DanielandClaude Opus 5.5 96225b5e02 Rücksprungkanten an allen Elementen vorbeiführen (#12) 
- Mit Lanes: ruecksprung_points() fährt senkrecht nur in den Gängen
  zwischen den Spalten (rechts der Quelle, links des Ziels) und waagrecht
  knapp unter Quelle bzw. Ziel und in einem Kanal am Poolboden. Das
  Kanalband wird nach der Platzierung an die letzte Lane angehängt: Die
  Elemente bleiben, wo sie sind, die Kante bleibt im eigenen Pool, jede
  Rücksprungkante hat eine eigene Kanalhöhe.
- Ohne Lanes: Kanal seitlich an allen Elementen zwischen Quelle und Ziel
  vorbei (parallele Zweige stehen versetzt). Steht ein Nachbar in der
  Zeile im Weg, unten hinaus bzw. oben hinein über die Lücke zwischen den
  Zeilen (zeile(), zeile_frei()).
- beispiel_lanes: Kante S4->S2 läuft nicht mehr durch S5, Pool um das
  Kanalband höher, externer Pool entsprechend tiefer. beispiel_prozess:
  Kanal bei x 50 statt 20. Referenzmodell unverändert.
- xfail für #12 entfernt; fünf neue Tests in test_md_to_bpmn.py.
- Belastungstest (126 Rücksprung-Varianten, mit/ohne Lanes): Kreuzungen
  durch Rücksprungkanten vorher in 31 Fällen, jetzt in keinem.
- Tests: python -m pytest -q -> 102 bestanden, 2 xfail (#13).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 13:13:37 +02:00

1111 lines
47 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
md_to_bpmn.py
=============
Wandelt eine einfache Markdown-Prozessbeschreibung in eine standardkonforme
BPMN-2.0-XML-Datei um (OMG-Standard, offen dokumentiert, kein Bezug zu
internen PICTURE-UUIDs).
Wird bei einem Arbeitsschritt "Verantwortlicher: Aufgabe" geschrieben
(z. B. "S2 Task: Sachbearbeiter: Antrag prüfen"), erkennt das Skript den
Teil vor dem ersten Doppelpunkt als Verantwortlichen und erzeugt daraus
automatisch Swimlanes (Pool + Lanes). Ohne diese Angabe entsteht wie bisher
ein Diagramm ohne Lanes.
Unterstützt neben Task/Gateway/Start/Ende auch die übrigen für PICTURE
relevanten Elemente aus Elementkatalog.md (Aufgaben-Varianten, typisierte
Ereignisse, ereignisbasiertes Gateway, Datenobjekt/-speicher, Gruppe sowie
geschlossene Pools mit Nachrichtenflüssen). Siehe Anleitung.md für die
vollständige Syntax.
Aufruf:
python3 md_to_bpmn.py prozess.md ausgabe.bpmn
"""
import argparse
import math
import re
import sys
import xml.etree.ElementTree as ET
from xml.dom import minidom
from collections import defaultdict
# ---------------------------------------------------------------------------
# 1) DSL
# ---------------------------------------------------------------------------
SIMPLE_TYPE_ALIASES = {
'start': 'StartEvent', 'startereignis': 'StartEvent',
'ende': 'EndEvent', 'end': 'EndEvent', 'endereignis': 'EndEvent',
'zwischenereignis': 'IntermediateEvent', 'intermediate': 'IntermediateEvent',
'task': 'Task', 'aufgabe': 'Task',
'manuelle aufgabe': 'ManualTask', 'manual task': 'ManualTask', 'manualtask': 'ManualTask',
'user aufgabe': 'UserTask', 'user task': 'UserTask', 'usertask': 'UserTask',
'service aufgabe': 'ServiceTask', 'service task': 'ServiceTask', 'servicetask': 'ServiceTask',
'script aufgabe': 'ScriptTask', 'script task': 'ScriptTask', 'scripttask': 'ScriptTask',
'geschäftsregel aufgabe': 'BusinessRuleTask', 'geschaeftsregel aufgabe': 'BusinessRuleTask',
'business rule task': 'BusinessRuleTask', 'businessruletask': 'BusinessRuleTask',
'sendeaufgabe': 'SendTask', 'send task': 'SendTask', 'aufgabe sendend': 'SendTask', 'sendtask': 'SendTask',
'empfangsaufgabe': 'ReceiveTask', 'receive task': 'ReceiveTask', 'aufgabe empfangend': 'ReceiveTask',
'receivetask': 'ReceiveTask',
'gateway xor': 'XorGateway', 'xor': 'XorGateway', 'entscheidung': 'XorGateway',
'gateway or': 'OrGateway', 'gateway oder': 'OrGateway', 'or': 'OrGateway', 'oder': 'OrGateway',
'gateway and': 'AndGateway', 'gateway und': 'AndGateway', 'and': 'AndGateway',
'und': 'AndGateway', 'parallel': 'AndGateway',
'gateway ereignisbasiert': 'EventBasedGateway', 'gateway ebg': 'EventBasedGateway',
'ereignisbasierte verzweigung': 'EventBasedGateway', 'event based gateway': 'EventBasedGateway',
'hinweis': 'TextAnnotation', 'note': 'TextAnnotation', 'anmerkung': 'TextAnnotation',
'datenobjekt': 'DataObject', 'dokument': 'DataObject', 'data object': 'DataObject',
'datenspeicher': 'DataStore', 'data store': 'DataStore',
'gruppe': 'Group', 'group': 'Group',
'teilprozess': 'SubProcess', 'subprocess': 'SubProcess',
'geschlossener teilprozess': 'SubProcess',
'aufruf': 'CallActivity', 'aufrufelement': 'CallActivity', 'aufruf element': 'CallActivity',
'call activity': 'CallActivity', 'callactivity': 'CallActivity',
}
EVENT_USAGE_WORDS = {
'start': 'StartEvent', 'startereignis': 'StartEvent',
'zwischenereignis': 'IntermediateEvent', 'intermediate': 'IntermediateEvent',
'ende': 'EndEvent', 'end': 'EndEvent', 'endereignis': 'EndEvent',
}
EVENT_KIND_WORDS = {
'nachricht': 'message', 'message': 'message',
'zeit': 'timer', 'timer': 'timer',
'eskalation': 'escalation', 'escalation': 'escalation',
'bedingung': 'conditional', 'condition': 'conditional', 'conditional': 'conditional',
'link': 'link',
'fehler': 'error', 'error': 'error',
'signal': 'signal',
'terminierung': 'terminate', 'terminate': 'terminate', 'termination': 'terminate',
}
EVENT_DIRECTION_WORDS = {
'sendend': 'throw', 'send': 'throw', 'werfend': 'throw', 'throw': 'throw',
'empfangend': 'catch', 'receive': 'catch', 'fangend': 'catch', 'catch': 'catch',
}
EVENT_DEFINITION_TAG = {
'message': 'messageEventDefinition', 'timer': 'timerEventDefinition',
'escalation': 'escalationEventDefinition', 'conditional': 'conditionalEventDefinition',
'link': 'linkEventDefinition', 'error': 'errorEventDefinition',
'signal': 'signalEventDefinition', 'terminate': 'terminateEventDefinition',
}
UNCONNECTED_TYPES = {'TextAnnotation', 'DataObject', 'DataStore', 'Group'}
# Für die Verteilung mehrerer Anschlusspunkte (Modellierregel 4) muss die tatsächliche
# Kontur bekannt sein: ein Punkt auf dem rechteckigen Begrenzungsrahmen liegt bei einer
# Raute (Gateway) oder einem Kreis (Ereignis) sonst neben der sichtbaren Form.
SHAPE_KIND = {
'StartEvent': 'circle', 'EndEvent': 'circle', 'IntermediateEvent': 'circle',
'XorGateway': 'diamond', 'OrGateway': 'diamond', 'AndGateway': 'diamond',
'EventBasedGateway': 'diamond',
}
DEFAULT_SIZE = {
'StartEvent': (36, 36), 'EndEvent': (36, 36), 'IntermediateEvent': (36, 36),
'XorGateway': (50, 50), 'OrGateway': (50, 50), 'AndGateway': (50, 50), 'EventBasedGateway': (50, 50),
'Task': (100, 80), 'ManualTask': (100, 80), 'UserTask': (100, 80), 'ServiceTask': (100, 80),
'ScriptTask': (100, 80), 'BusinessRuleTask': (100, 80), 'SendTask': (100, 80), 'ReceiveTask': (100, 80),
'SubProcess': (100, 80), 'CallActivity': (100, 80),
'TextAnnotation': (160, 60), 'DataObject': (36, 50), 'DataStore': (50, 50), 'Group': (220, 140),
}
STEP_RE = re.compile(
r'^(?P<id>[A-Za-z0-9_.\-]+)\s+'
r'(?P<typewords>[A-Za-zÄÖÜäöüß]+(?:\s+[A-Za-zÄÖÜäöüß]+){0,7})'
r'\s*:\s*(?P<label>.*)$',
re.IGNORECASE,
)
ARROW_RE = re.compile(r'^\s*(?:(?P<label>.+?)\s*)?->\s*(?P<target>\S+)\s*$')
MSGFLOW_OUT_RE = re.compile(r'^\s*(?:(?P<label>.+?)\s*)?~>\s*(?P<target>.+?)\s*$')
MSGFLOW_IN_RE = re.compile(r'^\s*(?:(?P<label>.+?)\s*)?<~\s*(?P<target>.+?)\s*$')
FRONTMATTER_RE = re.compile(r'^(?P<key>[A-Za-zÄÖÜäöüß ]+):\s*(?P<value>.*)$')
UNASSIGNED_LANE = 'Nicht zugeordnet'
EXTERN_KEYS = {'extern', 'externer beteiligter', 'externe beteiligte', 'external'}
def resolve_type(typewords_raw, raw_line):
"""Ordnet die Typwörter vor dem Doppelpunkt einem internen Elementtyp zu.
Gibt (typ, event_kind, event_direction) zurück; event_kind/-direction
sind nur für Start-/Zwischen-/Endereignisse mit Typisierung gesetzt.
"""
norm = re.sub(r'\s+', ' ', typewords_raw.strip().lower())
if norm in SIMPLE_TYPE_ALIASES:
return SIMPLE_TYPE_ALIASES[norm], None, None
words = norm.split(' ')
usage = EVENT_USAGE_WORDS.get(words[0])
if usage is None:
raise ValueError(f'Unbekannter Schritt-Typ: {typewords_raw!r} (Zeile: {raw_line!r})')
kind = None
direction = None
for w in words[1:]:
if w in EVENT_KIND_WORDS:
if kind is not None:
raise ValueError(f'Mehrfache Ereignis-Typisierung in {typewords_raw!r} (Zeile: {raw_line!r})')
kind = EVENT_KIND_WORDS[w]
elif w in EVENT_DIRECTION_WORDS:
if direction is not None:
raise ValueError(f'Mehrfache Richtungsangabe in {typewords_raw!r} (Zeile: {raw_line!r})')
direction = EVENT_DIRECTION_WORDS[w]
else:
raise ValueError(f'Unbekanntes Wort {w!r} in Ereignis-Typ {typewords_raw!r} (Zeile: {raw_line!r})')
if direction is None:
direction = 'throw' if usage == 'EndEvent' else 'catch'
if usage == 'StartEvent' and direction == 'throw':
raise ValueError(
f'Startereignisse sind immer eintretend/empfangend, "sendend" ist hier nicht '
f'zulässig (Zeile: {raw_line!r})')
if usage == 'EndEvent' and direction == 'catch':
raise ValueError(
f'Endereignisse sind immer auslösend/sendend, "empfangend" ist hier nicht '
f'zulässig (Zeile: {raw_line!r})')
if kind == 'terminate' and usage != 'EndEvent':
raise ValueError(f'"Terminierung" ist nur als Endereignis zulässig (Zeile: {raw_line!r})')
if kind == 'link' and usage != 'IntermediateEvent':
raise ValueError(
f'"Link" ist nur als Zwischenereignis zulässig, paarweise sendend/empfangend '
f'(Zeile: {raw_line!r})')
return usage, kind, direction
def split_lane(label):
"""Erkennt 'Verantwortlicher: Rest' am Anfang eines Labels.
Gibt (lane_oder_None, bereinigtes_label) zurück."""
if ':' not in label:
return None, label
prefix, rest = label.split(':', 1)
prefix = prefix.strip()
rest = rest.strip()
if not rest or not prefix:
return None, label
if len(prefix) > 30 or len(prefix.split()) > 4:
return None, label
if '?' in prefix or not prefix[0].isalpha():
return None, label
return prefix, rest
class Step:
def __init__(self, sid, typ, label, order, event_kind=None, event_direction=None):
self.id = sid
self.type = typ
self.order = order
self.edges = [] # (label_or_None, target_id)
self.msg_flows = [] # (direction 'out'/'in', label_or_None, externer_name)
self.event_kind = event_kind
self.event_direction = event_direction
if typ in UNCONNECTED_TYPES:
self.lane = None
self.label = label.strip()
else:
lane, clean = split_lane(label.strip())
self.lane = lane
self.label = clean
def parse_markdown(text):
meta = {}
steps = {}
order = []
externals = []
current = None
for raw_line in text.splitlines():
stripped = raw_line.strip()
if not stripped or stripped.startswith('#'):
continue
step_m = STEP_RE.match(stripped)
if not step_m and current is not None:
if '~>' in stripped:
m = MSGFLOW_OUT_RE.match(stripped)
if m:
current.msg_flows.append(('out', m.group('label'), m.group('target').strip()))
continue
if '<~' in stripped:
m = MSGFLOW_IN_RE.match(stripped)
if m:
current.msg_flows.append(('in', m.group('label'), m.group('target').strip()))
continue
if '->' in stripped:
m = ARROW_RE.match(stripped)
if m:
current.edges.append((m.group('label'), m.group('target')))
continue
if step_m:
sid = step_m.group('id')
typ, event_kind, event_direction = resolve_type(step_m.group('typewords'), raw_line)
label = step_m.group('label')
if sid in steps:
raise ValueError(f'Doppelte Schritt-ID: {sid!r}')
st = Step(sid, typ, label, len(order), event_kind, event_direction)
steps[sid] = st
order.append(sid)
current = st
continue
fm = FRONTMATTER_RE.match(stripped)
if fm and current is None:
key = fm.group('key').strip().lower()
value = fm.group('value').strip()
if key in EXTERN_KEYS:
externals.append(value)
else:
meta[key] = value
continue
if current is None:
continue
raise ValueError(f'Zeile konnte nicht interpretiert werden: {raw_line!r}')
flow_order = [sid for sid in order if steps[sid].type not in UNCONNECTED_TYPES]
for i, sid in enumerate(flow_order):
st = steps[sid]
if not st.edges and st.type != 'EndEvent':
if i + 1 < len(flow_order):
st.edges.append((None, flow_order[i + 1]))
for st in steps.values():
for _, tgt in st.edges:
if tgt not in steps:
raise ValueError(f'Schritt {st.id!r} verweist auf unbekanntes Ziel {tgt!r}')
if st.type in UNCONNECTED_TYPES or steps[tgt].type in UNCONNECTED_TYPES:
raise ValueError(
f'Sequenzfluss {st.id!r} -> {tgt!r} nicht erlaubt: Hinweise, Datenobjekte, '
f'Datenspeicher und Gruppen sind keine Ablaufelemente (allenfalls per Notiz '
f'zuordnen, in PICTURE nach dem Import als Assoziation verbinden).')
for _, _, target in st.msg_flows:
if target not in externals:
raise ValueError(
f'Schritt {st.id!r}: unbekannter externer Beteiligter {target!r} - vorher mit '
f'"Extern: {target}" deklarieren.')
incoming_count = defaultdict(int)
for st in steps.values():
for _, tgt in st.edges:
incoming_count[tgt] += 1
for st in steps.values():
if st.type == 'EventBasedGateway' and incoming_count[st.id] > 1:
raise ValueError(
f'Ereignisbasiertes Gateway {st.id!r} hat mehrere eingehende Kanten - es darf laut '
f'Modellierregeln nicht als Zusammenführung genutzt werden.')
return meta, steps, order, externals
# ---------------------------------------------------------------------------
# 2) Layer-Berechnung (gemeinsam für beide Layout-Modi)
# ---------------------------------------------------------------------------
def compute_layers(steps, order):
WHITE, GRAY, BLACK = 0, 1, 2
color = {sid: WHITE for sid in order}
back_edges = set()
def dfs(u):
color[u] = GRAY
for _, v in steps[u].edges:
if color[v] == WHITE:
dfs(v)
elif color[v] == GRAY:
back_edges.add((u, v))
color[u] = BLACK
for sid in order:
if color[sid] == WHITE:
dfs(sid)
fwd_preds = {sid: [] for sid in order}
for sid in order:
for _, tgt in steps[sid].edges:
if (sid, tgt) in back_edges:
continue
fwd_preds[tgt].append(sid)
layer = {}
def longest_path(sid, visiting=None):
if sid in layer:
return layer[sid]
visiting = (visiting or set()) | {sid}
preds = fwd_preds[sid]
layer[sid] = 0 if not preds else 1 + max(longest_path(p, visiting) for p in preds)
return layer[sid]
for sid in order:
longest_path(sid)
return layer, fwd_preds, back_edges
# ---------------------------------------------------------------------------
# 2a) Layout ohne Lanes: Schichten von oben nach unten
# ---------------------------------------------------------------------------
COLUMN_WIDTH = 170
ROW_HEIGHT = 170
MARGIN = 80
def layout_vertical(steps, order):
layer, fwd_preds, back_edges = compute_layers(steps, order)
max_layer = max(layer.values()) if layer else 0
layers = {i: [] for i in range(max_layer + 1)}
for sid in order:
layers[layer[sid]].append(sid)
x_pos = {}
for i in sorted(layers):
nodes_in_layer = sorted(layers[i], key=lambda s: steps[s].order)
if i == 0:
for idx, sid in enumerate(nodes_in_layer):
x_pos[sid] = idx * COLUMN_WIDTH
else:
bary = {}
for sid in nodes_in_layer:
preds = fwd_preds[sid]
bary[sid] = (sum(x_pos[p] for p in preds) / len(preds)) if preds else steps[sid].order * COLUMN_WIDTH
nodes_in_layer.sort(key=lambda s: (bary[s], steps[s].order))
for idx, sid in enumerate(nodes_in_layer):
x_pos[sid] = idx * COLUMN_WIDTH
if nodes_in_layer:
mean_x = sum(x_pos[s] for s in nodes_in_layer) / len(nodes_in_layer)
for sid in nodes_in_layer:
x_pos[sid] -= mean_x
min_x = min(x_pos.values()) if x_pos else 0
shift = MARGIN - min_x
for sid in x_pos:
x_pos[sid] += shift
y_pos = {sid: MARGIN + layer[sid] * ROW_HEIGHT for sid in order}
return x_pos, y_pos, back_edges, None
# ---------------------------------------------------------------------------
# 2b) Layout mit Lanes: Schichten von links nach rechts, Lanes als Zeilen
# ---------------------------------------------------------------------------
LANE_COLUMN_WIDTH = 190
LANE_HEADER_WIDTH = 30
LANE_PADDING = 40
LANE_ROW_HEIGHT = 150
LANE_MIN_HEIGHT = 150
# Rücksprungkanten im Lanes-Layout (#12): Abstand der Kanalhöhen am Poolboden und
# Länge des senkrechten Stücks unter Quelle bzw. Ziel. Unter jedem Element sind im
# eigenen Zeilenslot mindestens (LANE_ROW_HEIGHT - 80) / 2 = 35 frei.
RUECKSPRUNG_ABSTAND = 15
RUECKSPRUNG_STUB = 20
# Ohne Lanes: Abstand des seitlichen Kanals vom äußersten Element (MARGIN 80 lässt
# links Platz für mehrere Kanäle).
RUECKSPRUNG_SEITE = 30
GROESSTE_BREITE = max(w for typ, (w, _) in DEFAULT_SIZE.items() if typ not in UNCONNECTED_TYPES)
def layout_lanes(steps, order, lanes):
layer, fwd_preds, back_edges = compute_layers(steps, order)
lane_index = {name: i for i, name in enumerate(lanes)}
# je Lane: wie viele Knoten teilen sich denselben Layer? -> Zeilen je Lane
lane_layer_nodes = defaultdict(lambda: defaultdict(list))
for sid in order:
st = steps[sid]
if st.type in UNCONNECTED_TYPES:
continue
lane_layer_nodes[st.lane or UNASSIGNED_LANE][layer[sid]].append(sid)
lane_rows = {}
for lane in lanes:
buckets = lane_layer_nodes.get(lane, {})
lane_rows[lane] = max((len(v) for v in buckets.values()), default=1)
lane_height = {lane: max(LANE_MIN_HEIGHT, lane_rows[lane] * LANE_ROW_HEIGHT + LANE_PADDING)
for lane in lanes}
lane_top = {}
running = 0
for lane in lanes:
lane_top[lane] = running
running += lane_height[lane]
pool_height = running
x_pos, y_pos = {}, {}
for lane in lanes:
buckets = lane_layer_nodes.get(lane, {})
for lyr, sids in buckets.items():
sids_sorted = sorted(sids, key=lambda s: steps[s].order)
n = len(sids_sorted)
band_h = lane_height[lane]
for row, sid in enumerate(sids_sorted):
st = steps[sid]
_, h = DEFAULT_SIZE.get(st.type, (100, 80))
slot_h = band_h / n
x_pos[sid] = MARGIN + LANE_HEADER_WIDTH + lyr * LANE_COLUMN_WIDTH
y_pos[sid] = lane_top[lane] + row * slot_h + (slot_h - h) / 2
# Unverbundene Artefakte (Hinweis, Datenobjekt, Datenspeicher, Gruppe): rechts
# neben dem letzten Layer, gestapelt in der ersten Lane.
max_layer = max(layer.values()) if layer else 0
unconnected_i = 0
for sid in order:
st = steps[sid]
if st.type not in UNCONNECTED_TYPES:
continue
x_pos[sid] = MARGIN + LANE_HEADER_WIDTH + (max_layer + 1) * LANE_COLUMN_WIDTH
y_pos[sid] = lane_top[lanes[0]] + LANE_PADDING + unconnected_i * 90
unconnected_i += 1
return x_pos, y_pos, back_edges, (lane_top, lane_height, pool_height)
# ---------------------------------------------------------------------------
# 3) BPMN-2.0-XML aufbauen
# ---------------------------------------------------------------------------
BPMN_TAG = {
'StartEvent': 'startEvent', 'EndEvent': 'endEvent',
'IntermediateEvent': 'intermediateCatchEvent', # bei "sendend" überschrieben, siehe bpmn_tag_for
'Task': 'task', 'ManualTask': 'manualTask', 'UserTask': 'userTask',
'ServiceTask': 'serviceTask', 'ScriptTask': 'scriptTask',
'BusinessRuleTask': 'businessRuleTask', 'SendTask': 'sendTask', 'ReceiveTask': 'receiveTask',
'SubProcess': 'subProcess', 'CallActivity': 'callActivity',
'XorGateway': 'exclusiveGateway', 'OrGateway': 'inclusiveGateway', 'AndGateway': 'parallelGateway',
'EventBasedGateway': 'eventBasedGateway',
}
ID_PREFIX = {
'StartEvent': 'StartEvent', 'EndEvent': 'EndEvent', 'IntermediateEvent': 'IntermediateEvent',
'Task': 'Task', 'ManualTask': 'ManualTask', 'UserTask': 'UserTask',
'ServiceTask': 'ServiceTask', 'ScriptTask': 'ScriptTask',
'BusinessRuleTask': 'BusinessRuleTask', 'SendTask': 'SendTask', 'ReceiveTask': 'ReceiveTask',
'SubProcess': 'SubProcess', 'CallActivity': 'CallActivity',
'XorGateway': 'Gateway', 'OrGateway': 'Gateway', 'AndGateway': 'Gateway', 'EventBasedGateway': 'Gateway',
'TextAnnotation': 'TextAnnotation', 'DataObject': 'DataObject', 'DataStore': 'DataStore', 'Group': 'Group',
}
NS = {
'bpmn': 'http://www.omg.org/spec/BPMN/20100524/MODEL',
'bpmndi': 'http://www.omg.org/spec/BPMN/20100524/DI',
'dc': 'http://www.omg.org/spec/DD/20100524/DC',
'di': 'http://www.omg.org/spec/DD/20100524/DI',
}
for prefix, uri in NS.items():
ET.register_namespace(prefix, uri)
def q(prefix, tag):
return f'{{{NS[prefix]}}}{tag}'
def bpmn_tag_for(st):
if st.type == 'IntermediateEvent':
return 'intermediateThrowEvent' if st.event_direction == 'throw' else 'intermediateCatchEvent'
return BPMN_TAG[st.type]
def sanitize_id(raw):
s = re.sub(r'[^A-Za-z0-9_.\-]', '_', raw)
if not s or not (s[0].isalpha() or s[0] == '_'):
s = 'n_' + s
return s
def spread_point(x, y, w, h, side, index, count, kind='rect'):
"""Verteilt mehrere Anschlusspunkte auf derselben Elementseite, statt sie alle
im Mittelpunkt zusammenlaufen zu lassen (Modellierregel: eigener Anschlusspunkt
je Kante). Bei count=1 identisch zum bisherigen Mittelpunkt-Anschluss.
`kind` legt die tatsächliche Kontur fest, auf der die Punkte liegen müssen:
'rect' (Rechteck, z. B. Task), 'circle' (Ereignis) oder 'diamond' (Gateway).
Ein Punkt auf dem bloßen Begrenzungsrahmen würde bei Kreis/Raute außerhalb
der sichtbaren Form landen und die Kante wirkt dann unverbunden."""
frac = (index + 1) / (count + 1)
cx, cy = x + w / 2, y + h / 2
if kind == 'circle':
r = min(w, h) / 2
base_angle = {'E': 0.0, 'S': 90.0, 'W': 180.0, 'N': 270.0}[side]
spread_deg = 70.0
angle = math.radians(base_angle + (frac - 0.5) * spread_deg)
return cx + r * math.cos(angle), cy + r * math.sin(angle)
if kind == 'diamond':
vertex = {'N': (cx, y), 'S': (cx, y + h), 'W': (x, cy), 'E': (x + w, cy)}[side]
neg_target, pos_target = {
'N': ((x, cy), (x + w, cy)), 'S': ((x, cy), (x + w, cy)),
'W': ((cx, y), (cx, y + h)), 'E': ((cx, y), (cx, y + h)),
}[side]
offset = frac - 0.5
t = min(abs(offset) * 2 * 0.6, 0.6)
tx, ty = pos_target if offset > 0 else neg_target
vx, vy = vertex
return vx + (tx - vx) * t, vy + (ty - vy) * t
if side == 'N':
return x + w * frac, y
if side == 'S':
return x + w * frac, y + h
if side == 'W':
return x, y + h * frac
if side == 'E':
return x + w, y + h * frac
raise ValueError(f'Unbekannte Seite: {side!r}')
GATEWAY_OUTWARD = {'N': (0, -1), 'S': (0, 1), 'E': (1, 0), 'W': (-1, 0)}
# Halbe Kantenlaenge der Raute (Standardgroesse 50x50) reicht, um deren Kontur
# in jedem Fall sicher zu verlassen, bevor die Kante abbiegt.
GATEWAY_STUB = 25
def _dedupe_consecutive(points):
out = [points[0]]
for p in points[1:]:
if p != out[-1]:
out.append(p)
return out
def _orthogonal_middle(ox, oy, src_side, tx, ty, tgt_side):
src_axis = 'v' if src_side in ('N', 'S') else 'h'
tgt_axis = 'v' if tgt_side in ('N', 'S') else 'h'
if src_axis == 'v' and tgt_axis == 'v':
if abs(ox - tx) < 1e-6:
return [(ox, oy), (tx, ty)]
mid_y = oy + (ty - oy) / 2
return [(ox, oy), (ox, mid_y), (tx, mid_y), (tx, ty)]
if src_axis == 'h' and tgt_axis == 'h':
if abs(oy - ty) < 1e-6:
return [(ox, oy), (tx, ty)]
mid_x = ox + (tx - ox) / 2
return [(ox, oy), (mid_x, oy), (mid_x, ty), (tx, ty)]
if src_axis == 'v':
return [(ox, oy), (ox, ty), (tx, ty)]
return [(ox, oy), (tx, oy), (tx, ty)]
# Abstand der Ausweichachse von der Mitte der Ziel-Spalte bzw. -Zeile: halbe
# Breite der größten Elemente (Aufgabe 100x80) plus halber kleinster Gang
# zwischen zwei Spalten/Zeilen (170 - 100 = 70). Der Gang direkt vor dem Ziel
# ist damit in beiden Layouts frei; die Mitte zwischen Quelle und Ziel nicht,
# dort kann eine Spalte mit einer Aufgabe liegen.
AUSWEICHABSTAND = 85
def _ausweichachse(quelle, ziel):
"""Koordinate im freien Gang vor der Ziel-Spalte (bzw. -Zeile), auf der Seite
der Quelle: dort läuft eine Kante an der Ziel-Raute vorbei, ohne ein anderes
Element zu schneiden."""
return ziel + AUSWEICHABSTAND if quelle > ziel else ziel - AUSWEICHABSTAND
def ruecksprung_points(ox, oy, tx, ty, src_spalte, tgt_spalte, kanal_y):
"""Weg einer Rücksprungkante im Lanes-Layout vom Südpunkt der Quelle zum
Südpunkt des Ziels. Senkrecht wird nur in den freien Gängen zwischen den
Spalten gefahren (rechts der Quell-, links der Ziel-Spalte), waagrecht nur
knapp unter Quelle bzw. Ziel und im Kanal am Poolboden - so schneidet die
Kante keine Elemente, die in denselben Spalten weiter unten stehen (#12).
src_spalte/tgt_spalte sind die linken Kanten der Spalten (x_pos)."""
rechts = src_spalte + (GROESSTE_BREITE + LANE_COLUMN_WIDTH) / 2
links = tgt_spalte - (LANE_COLUMN_WIDTH - GROESSTE_BREITE) / 2
unter_quelle = oy + RUECKSPRUNG_STUB
unter_ziel = ty + RUECKSPRUNG_STUB
return [(ox, oy), (ox, unter_quelle), (rechts, unter_quelle), (rechts, kanal_y),
(links, kanal_y), (links, unter_ziel), (tx, unter_ziel), (tx, ty)]
def orthogonal_points(ox, oy, src_side, tx, ty, tgt_side,
src_is_gateway=False, tgt_is_gateway=False):
"""Baut einen rechtwinkligen Kantenverlauf zwischen zwei Anschlusspunkten, der
am Start gerade in Richtung src_side abgeht und am Ziel gerade aus Richtung
tgt_side ankommt - so, wie PICTURE Verbindungen an einem Gateway zeichnet.
An einer Raute (`src_is_gateway`/`tgt_is_gateway`) liegt der Anschlusspunkt
auf der Kontur, nicht am Rand einer Bounding-Box: ein direkter Kurs zum
jeweils anderen Ende kann je nach dessen Lage durchs Innere der Raute
zurücklaufen. Deshalb macht die Kante dort zuerst einen kurzen Schritt
gerade nach aussen (Modellierregel: Kante nie durchs Innere eines
Elements), bevor sie zum Ziel abbiegt - und zwar so, dass die Achse
dieses Schritts (die einzige garantiert freie Koordinate) beim nächsten
Abbiegen zunächst unverändert bleibt. Sonst würde die Kante beim
Umbiegen zum Ziel auf demselben Weg zurück durch die Raute laufen, den
sie gerade erst verlassen hat."""
sx, sy = ox, oy
if src_is_gateway:
dx, dy = GATEWAY_OUTWARD[src_side]
sx, sy = ox + dx * GATEWAY_STUB, oy + dy * GATEWAY_STUB
ex, ey = tx, ty
if tgt_is_gateway:
dx, dy = GATEWAY_OUTWARD[tgt_side]
ex, ey = tx + dx * GATEWAY_STUB, ty + dy * GATEWAY_STUB
src_axis = 'v' if src_side in ('N', 'S') else 'h'
tgt_axis = 'v' if tgt_side in ('N', 'S') else 'h'
if not src_is_gateway and not tgt_is_gateway:
points = _orthogonal_middle(ox, oy, src_side, tx, ty, tgt_side)
elif src_is_gateway and not tgt_is_gateway:
# sy/sx (die Stub-Achse) fixieren, solange die andere Koordinate zum
# Ziel hin schwenkt - erst danach auf der Stub-Achse zum Ziel.
if src_axis == 'v':
points = [(ox, oy), (sx, sy), (tx, sy), (tx, ty)]
else:
points = [(ox, oy), (sx, sy), (sx, ty), (tx, ty)]
elif tgt_is_gateway and not src_is_gateway:
# Spiegelbildlich: erst auf der Stub-Achse des Ziels ankommen, erst
# danach dorthin schwenken.
if tgt_axis == 'v':
points = [(ox, oy), (ox, ey), (ex, ey), (tx, ty)]
else:
points = [(ox, oy), (ex, oy), (ex, ey), (tx, ty)]
else:
# Beide Enden sind Rauten: von der Stub-Achse des Starts direkt auf
# die Stub-Achse des Ziels schwenken, jede nur einmal verlassen.
# Liegt der Stub des Starts schon jenseits des Ziel-Anschlusspunkts
# (z. B. Ankunft am Nordpunkt eines höher liegenden Ziels), liefe das
# letzte gerade Stück durch die Ziel-Raute - dann seitlich daran
# vorbei über eine freie Ausweichachse (Issue #8).
if src_axis == 'v' and tgt_axis == 'v' and (
(tgt_side == 'N' and sy > ty) or (tgt_side == 'S' and sy < ty)):
mx = _ausweichachse(sx, ex)
points = [(ox, oy), (sx, sy), (mx, sy), (mx, ey), (ex, ey), (tx, ty)]
elif src_axis == 'h' and tgt_axis == 'h' and (
(tgt_side == 'W' and sx > tx) or (tgt_side == 'E' and sx < tx)):
my = _ausweichachse(sy, ey)
points = [(ox, oy), (sx, sy), (sx, my), (ex, my), (ex, ey), (tx, ty)]
elif src_axis == 'v':
points = [(ox, oy), (sx, sy), (ex, sy), (ex, ey), (tx, ty)]
else:
points = [(ox, oy), (sx, sy), (sx, ey), (ex, ey), (tx, ty)]
return _dedupe_consecutive(points)
GATEWAY_OUT_PREF = ['S', 'E', 'W', 'N']
GATEWAY_IN_PREF = ['N', 'W', 'E', 'S']
def assign_gateway_sides(steps, flows):
"""PICTURE-Gateways haben nur 4 feste Anschlusspunkte (N/O/S/W); jede Kante
dort verlaeuft gerade. Weist daher jeder (nicht rueckwaertigen) Kante an
einem Gateway einen eigenen Eckpunkt zu. Ab der fuenften Kante an
demselben Gateway muss ein Punkt zweimal genutzt werden - das ist eine
echte Grenze von PICTURE, keine Modellierungsfrage (siehe Einschraenkungen
in Anleitung.md)."""
touching = defaultdict(list)
for flow_id, src, tgt, label, is_back in flows:
if is_back:
continue
if SHAPE_KIND.get(steps[src].type) == 'diamond':
touching[src].append((f'{flow_id}_src', False))
if SHAPE_KIND.get(steps[tgt].type) == 'diamond':
touching[tgt].append((f'{flow_id}_tgt', True))
assigned = {}
for sid, uses in touching.items():
used_sides = set()
overflow = []
for use_id, is_incoming in uses:
pref = GATEWAY_IN_PREF if is_incoming else GATEWAY_OUT_PREF
for side in pref:
if side not in used_sides:
assigned[use_id] = side
used_sides.add(side)
break
else:
overflow.append((use_id, is_incoming))
for use_id, is_incoming in overflow:
pref = GATEWAY_IN_PREF if is_incoming else GATEWAY_OUT_PREF
assigned[use_id] = pref[0]
return assigned
def build_bpmn(meta_md, steps, order, externals):
process_name = (meta_md.get('prozess') or meta_md.get('prozessname')
or meta_md.get('name') or 'Neuer Prozess')
ext_names = list(dict.fromkeys(externals)) # Reihenfolge erhalten, Duplikate entfernen
lanes = []
seen = set()
for sid in order:
st = steps[sid]
if st.type in UNCONNECTED_TYPES:
continue
name = st.lane or None
if name and name not in seen:
seen.add(name)
lanes.append(name)
has_lanes = bool(lanes)
if has_lanes and any((steps[sid].lane is None) for sid in order if steps[sid].type not in UNCONNECTED_TYPES):
lanes.append(UNASSIGNED_LANE)
has_collaboration = has_lanes or bool(ext_names)
if has_lanes:
x_pos, y_pos, back_edges, lane_geo = layout_lanes(steps, order, lanes)
else:
x_pos, y_pos, back_edges, lane_geo = layout_vertical(steps, order)
node_ids = {}
used = set()
for sid in order:
st = steps[sid]
base = f'{ID_PREFIX.get(st.type, "Node")}_{sanitize_id(sid)}'
candidate = base
n = 1
while candidate in used:
n += 1
candidate = f'{base}_{n}'
used.add(candidate)
node_ids[sid] = candidate
definitions = ET.Element(q('bpmn', 'definitions'), {
'id': 'Definitions_1',
'targetNamespace': 'http://picture-tool.local/bpmn',
'exporter': 'md_to_bpmn.py',
})
process_id = 'Process_1'
collaboration = None
ext_ids = {}
msg_flow_records = []
if has_collaboration:
collaboration = ET.SubElement(definitions, q('bpmn', 'collaboration'), {'id': 'Collaboration_1'})
participant_id = 'Participant_1'
ET.SubElement(collaboration, q('bpmn', 'participant'), {
'id': participant_id, 'name': process_name, 'processRef': process_id,
})
for i, name in enumerate(ext_names, start=1):
eid = f'Participant_Ext_{i}'
ext_ids[name] = eid
ET.SubElement(collaboration, q('bpmn', 'participant'), {'id': eid, 'name': name})
msg_n = 0
for sid in order:
st = steps[sid]
for direction, label, target in st.msg_flows:
msg_n += 1
mfid = f'MsgFlow_{msg_n}'
if direction == 'out':
src, tgt = node_ids[sid], ext_ids[target]
else:
src, tgt = ext_ids[target], node_ids[sid]
attrs = {'id': mfid, 'sourceRef': src, 'targetRef': tgt}
if label:
attrs['name'] = label
ET.SubElement(collaboration, q('bpmn', 'messageFlow'), attrs)
msg_flow_records.append((mfid, sid, direction, target))
process = ET.SubElement(definitions, q('bpmn', 'process'), {
'id': process_id, 'name': process_name, 'isExecutable': 'false',
})
lane_ids = {}
if has_lanes:
lane_set = ET.SubElement(process, q('bpmn', 'laneSet'), {'id': 'LaneSet_1'})
for i, lane_name in enumerate(lanes, start=1):
lid = f'Lane_{i}'
lane_ids[lane_name] = lid
lane_el = ET.SubElement(lane_set, q('bpmn', 'lane'), {'id': lid, 'name': lane_name})
for sid in order:
st = steps[sid]
if st.type in UNCONNECTED_TYPES:
continue
if (st.lane or UNASSIGNED_LANE) == lane_name:
ET.SubElement(lane_el, q('bpmn', 'flowNodeRef')).text = node_ids[sid]
incoming = {sid: [] for sid in order}
outgoing = {sid: [] for sid in order}
flows = []
flow_n = 0
for sid in order:
st = steps[sid]
for label, tgt in st.edges:
flow_n += 1
flow_id = f'Flow_{flow_n}'
is_back = (sid, tgt) in back_edges
flows.append((flow_id, sid, tgt, label, is_back))
outgoing[sid].append(flow_id)
incoming[tgt].append(flow_id)
for sid in order:
st = steps[sid]
if st.type in UNCONNECTED_TYPES:
continue
tag = bpmn_tag_for(st)
el = ET.SubElement(process, q('bpmn', tag), {'id': node_ids[sid], 'name': st.label})
for fid in incoming[sid]:
ET.SubElement(el, q('bpmn', 'incoming')).text = fid
for fid in outgoing[sid]:
ET.SubElement(el, q('bpmn', 'outgoing')).text = fid
if st.event_kind:
ET.SubElement(el, q('bpmn', EVENT_DEFINITION_TAG[st.event_kind]), {'id': f'{node_ids[sid]}_ed'})
for sid in order:
st = steps[sid]
if st.type == 'TextAnnotation':
el = ET.SubElement(process, q('bpmn', 'textAnnotation'), {'id': node_ids[sid]})
ET.SubElement(el, q('bpmn', 'text')).text = st.label
elif st.type == 'DataObject':
obj_id = f'{node_ids[sid]}_obj'
ET.SubElement(process, q('bpmn', 'dataObject'), {'id': obj_id})
ET.SubElement(process, q('bpmn', 'dataObjectReference'), {
'id': node_ids[sid], 'name': st.label, 'dataObjectRef': obj_id,
})
elif st.type == 'DataStore':
ET.SubElement(process, q('bpmn', 'dataStoreReference'), {'id': node_ids[sid], 'name': st.label})
elif st.type == 'Group':
category_id = f'{node_ids[sid]}_Category'
category = ET.SubElement(definitions, q('bpmn', 'category'), {'id': category_id})
cv_id = f'{category_id}_Value'
ET.SubElement(category, q('bpmn', 'categoryValue'), {'id': cv_id, 'value': st.label})
ET.SubElement(process, q('bpmn', 'group'), {'id': node_ids[sid], 'categoryValueRef': cv_id})
for flow_id, src, tgt, label, is_back in flows:
attrs = {'id': flow_id, 'sourceRef': node_ids[src], 'targetRef': node_ids[tgt]}
if label:
attrs['name'] = label
ET.SubElement(process, q('bpmn', 'sequenceFlow'), attrs)
# --- Diagram Interchange ---
diagram = ET.SubElement(definitions, q('bpmndi', 'BPMNDiagram'), {'id': 'Diagram_1'})
plane_root = 'Collaboration_1' if has_collaboration else process_id
plane = ET.SubElement(diagram, q('bpmndi', 'BPMNPlane'), {'id': 'Plane_1', 'bpmnElement': plane_root})
def bounds_of(sid):
st = steps[sid]
w, h = DEFAULT_SIZE.get(st.type, (100, 80))
return x_pos[sid], y_pos[sid], w, h
pool_width = max((x_pos[sid] + bounds_of(sid)[2] for sid in order), default=400) + MARGIN
back_flow_ids = [flow_id for flow_id, _, _, _, is_back in flows if is_back]
if has_lanes:
lane_top, lane_height, pool_height = lane_geo
if back_flow_ids:
# Kanal für Rücksprungkanten als Band am Boden der letzten Lane, erst
# nach der Platzierung angehängt - die Elemente bleiben, wo sie sind,
# und der Rückweg verläuft im eigenen Pool (#12).
band = RUECKSPRUNG_ABSTAND * (len(back_flow_ids) + 1)
lane_height[lanes[-1]] += band
pool_height += band
lane_geo = (lane_top, lane_height, pool_height)
else:
pool_height = max((y_pos[sid] + bounds_of(sid)[3] for sid in order), default=300) + MARGIN
ext_pos = {}
if has_collaboration:
ET.SubElement(plane, q('bpmndi', 'BPMNShape'), {
'id': 'Participant_1_di', 'bpmnElement': 'Participant_1', 'isHorizontal': 'true',
}).append(_bounds_el(0, 0, pool_width, pool_height))
if has_lanes:
for lane_name in lanes:
ET.SubElement(plane, q('bpmndi', 'BPMNShape'), {
'id': f'{lane_ids[lane_name]}_di', 'bpmnElement': lane_ids[lane_name], 'isHorizontal': 'true',
}).append(_bounds_el(0, lane_top[lane_name], pool_width, lane_height[lane_name]))
ext_top = pool_height + 60
for i, name in enumerate(ext_names):
eid = ext_ids[name]
ex_y = ext_top + i * 130
ext_pos[name] = (0, ex_y, pool_width, 100)
ET.SubElement(plane, q('bpmndi', 'BPMNShape'), {
'id': f'{eid}_di', 'bpmnElement': eid, 'isHorizontal': 'true',
}).append(_bounds_el(0, ex_y, pool_width, 100))
for sid in order:
attrs = {'id': f'{node_ids[sid]}_di', 'bpmnElement': node_ids[sid]}
if steps[sid].type == 'XorGateway':
# BPMN lässt das "X" im exklusiven Gateway optional; ohne dieses
# Attribut zeichnen Werkzeuge wie bpmn-js eine leere Raute (#15).
attrs['isMarkerVisible'] = 'true'
shape = ET.SubElement(plane, q('bpmndi', 'BPMNShape'), attrs)
x, y, w, h = bounds_of(sid)
shape.append(_bounds_el(x, y, w, h))
# Anschlusspunkte je (Element, Seite) zählen, damit mehrere Kanten an derselben
# Seite nicht alle im selben Punkt zusammenlaufen (Modellierregel 4).
def bounds_for(key):
kind, ident = key
return bounds_of(ident) if kind == 'n' else ext_pos[ident]
edge_sides = {}
anchor_group = defaultdict(list)
use_index = {}
def register_anchor(key, side, use_id):
lst = anchor_group[(key, side)]
use_index[use_id] = len(lst)
lst.append(use_id)
gateway_sides = assign_gateway_sides(steps, flows)
def zeile(sid):
"""Die übrigen verbundenen Elemente, die mit sid auf einer Höhe stehen."""
y, h = y_pos[sid], bounds_of(sid)[3]
return [o for o in order if o != sid and steps[o].type not in UNCONNECTED_TYPES
and y_pos[o] < y + h and y_pos[o] + bounds_of(o)[3] > y]
def zeile_frei(sid, seite):
"""True, wenn in der Zeile von sid zur Seite 'W' bzw. 'E' hin kein Element steht."""
if seite == 'W':
return all(x_pos[o] >= x_pos[sid] for o in zeile(sid))
rechts = x_pos[sid] + bounds_of(sid)[2]
return all(x_pos[o] + bounds_of(o)[2] <= rechts for o in zeile(sid))
for flow_id, src, tgt, label, is_back in flows:
if has_lanes:
src_side, tgt_side = ('E', 'W') if not is_back else ('S', 'S')
elif not is_back:
src_side, tgt_side = 'S', 'N'
else:
# Rücksprung ohne Lanes: seitlich hinaus zum Kanal - außer ein
# Nachbar in derselben Zeile steht im Weg, dann unten hinaus bzw.
# von oben hinein über die freie Lücke zwischen den Zeilen (#12).
same_side = 'W' if x_pos[src] <= x_pos[tgt] else 'E'
src_side = same_side if zeile_frei(src, same_side) else 'S'
tgt_side = same_side if zeile_frei(tgt, same_side) else 'N'
if not is_back:
# PICTURE-Gateways haben nur 4 feste Anschlusspunkte (N/O/S/W) - eigene
# Zuteilung statt der obigen Layout-Konvention, siehe assign_gateway_sides().
src_side = gateway_sides.get(f'{flow_id}_src', src_side)
tgt_side = gateway_sides.get(f'{flow_id}_tgt', tgt_side)
edge_sides[flow_id] = (src_side, tgt_side)
register_anchor(('n', src), src_side, f'{flow_id}_src')
register_anchor(('n', tgt), tgt_side, f'{flow_id}_tgt')
for mfid, sid, direction, target in msg_flow_records:
register_anchor(('n', sid), 'S', f'{mfid}_node')
register_anchor(('x', target), 'N', f'{mfid}_ext')
def shape_kind_for(key):
kind, ident = key
return SHAPE_KIND.get(steps[ident].type, 'rect') if kind == 'n' else 'rect'
def spread_anchor(key, side, use_id):
x, y, w, h = bounds_for(key)
idx = use_index[use_id]
count = len(anchor_group[(key, side)])
return spread_point(x, y, w, h, side, idx, count, shape_kind_for(key))
if has_lanes:
pool_bottom = lane_geo[2]
for flow_id, src, tgt, label, is_back in flows:
edge = ET.SubElement(plane, q('bpmndi', 'BPMNEdge'), {
'id': f'{flow_id}_di', 'bpmnElement': flow_id,
})
src_side, tgt_side = edge_sides[flow_id]
ox, oy = spread_anchor(('n', src), src_side, f'{flow_id}_src')
tx, ty = spread_anchor(('n', tgt), tgt_side, f'{flow_id}_tgt')
if not is_back:
src_is_gateway = SHAPE_KIND.get(steps[src].type) == 'diamond'
tgt_is_gateway = SHAPE_KIND.get(steps[tgt].type) == 'diamond'
points = orthogonal_points(ox, oy, src_side, tx, ty, tgt_side,
src_is_gateway, tgt_is_gateway)
elif has_lanes:
band = RUECKSPRUNG_ABSTAND * (len(back_flow_ids) + 1)
channel_y = pool_bottom - band + RUECKSPRUNG_ABSTAND * (back_flow_ids.index(flow_id) + 1)
points = ruecksprung_points(ox, oy, tx, ty, x_pos[src], x_pos[tgt], channel_y)
else:
# Ohne Lanes stehen die Elemente nicht in festen Spalten (parallele
# Zweige sind zur Mitte versetzt): Kanal seitlich an allen Elementen
# vorbei, die zwischen den Zeilen von Ziel und Quelle stehen (#12).
start, y_start = [(ox, oy)], oy
if src_side == 'S':
y_start = max(y_pos[o] + bounds_of(o)[3] for o in zeile(src) + [src]) + RUECKSPRUNG_STUB
start.append((ox, y_start))
ende, y_ende = [(tx, ty)], ty
if tgt_side == 'N':
y_ende = min(y_pos[o] for o in zeile(tgt) + [tgt]) - RUECKSPRUNG_STUB
ende.insert(0, (tx, y_ende))
oben, unten = min(y_start, y_ende), max(y_start, y_ende)
zwischen = [sid for sid in order if steps[sid].type not in UNCONNECTED_TYPES
and y_pos[sid] <= unten and y_pos[sid] + bounds_of(sid)[3] >= oben]
versatz = RUECKSPRUNG_SEITE + RUECKSPRUNG_ABSTAND * back_flow_ids.index(flow_id)
if x_pos[src] <= x_pos[tgt]:
channel_x = min(x_pos[sid] for sid in zwischen) - versatz
else:
channel_x = max(x_pos[sid] + bounds_of(sid)[2] for sid in zwischen) + versatz
points = _dedupe_consecutive(start + [(channel_x, y_start), (channel_x, y_ende)] + ende)
for px, py in points:
ET.SubElement(edge, q('di', 'waypoint'), {'x': f'{px:g}', 'y': f'{py:g}'})
if label:
# explizite Label-Position setzen, um Überlappungen mit Pfeilen/Nachbarlabels
# zu reduzieren: am ersten Segment, leicht seitlich versetzt
(p0x, p0y), (p1x, p1y) = points[0], points[1] if len(points) > 1 else points[0]
mx, my = (p0x + p1x) / 2, (p0y + p1y) / 2
if abs(p0y - p1y) < 1e-6: # horizontales Segment -> Label darüber
lx, ly = mx - 15, my - 20
else: # vertikales Segment -> Label seitlich
lx, ly = mx + 8, my - 8
bpmn_label = ET.SubElement(edge, q('bpmndi', 'BPMNLabel'))
bpmn_label.append(_bounds_el(lx, ly, max(30, len(label) * 6), 14))
for mfid, sid, direction, target in msg_flow_records:
edge = ET.SubElement(plane, q('bpmndi', 'BPMNEdge'), {'id': f'{mfid}_di', 'bpmnElement': mfid})
ox, oy = spread_anchor(('n', sid), 'S', f'{mfid}_node')
tx, ty = spread_anchor(('x', target), 'N', f'{mfid}_ext')
points = [(ox, oy), (tx, ty)] if direction == 'out' else [(tx, ty), (ox, oy)]
for px, py in points:
ET.SubElement(edge, q('di', 'waypoint'), {'x': f'{px:g}', 'y': f'{py:g}'})
return definitions
def _bounds_el(x, y, w, h):
el = ET.Element(q('dc', 'Bounds'), {
'x': f'{x:g}', 'y': f'{y:g}', 'width': f'{w:g}', 'height': f'{h:g}',
})
return el
def prettify(elem):
rough = ET.tostring(elem, encoding='utf-8')
parsed = minidom.parseString(rough)
return parsed.toprettyxml(indent=' ', encoding='UTF-8')
def main():
# Windows-Konsolen laufen häufig mit einer Legacy-Kodierung (z. B. cp1252) statt UTF-8;
# ohne dies würden Umlaute in Meldungen als Mojibake erscheinen oder das Programm abstürzen.
for stream in (sys.stdout, sys.stderr):
if hasattr(stream, 'reconfigure'):
stream.reconfigure(encoding='utf-8', errors='replace')
ap = argparse.ArgumentParser(description='Markdown-Prozessbeschreibung -> BPMN 2.0 XML')
ap.add_argument('markdown_file')
ap.add_argument('output_bpmn')
args = ap.parse_args()
with open(args.markdown_file, encoding='utf-8') as f:
md_text = f.read()
try:
meta_md, steps, order, externals = parse_markdown(md_text)
definitions = build_bpmn(meta_md, steps, order, externals)
except ValueError as exc:
print(f'Fehler beim Verarbeiten von {args.markdown_file}: {exc}', file=sys.stderr)
sys.exit(1)
xml_bytes = prettify(definitions)
with open(args.output_bpmn, 'wb') as f:
f.write(xml_bytes)
n_edges = sum(len(s.edges) for s in steps.values())
n_msgflows = sum(len(s.msg_flows) for s in steps.values())
lanes_used = len({steps[sid].lane for sid in order if steps[sid].lane})
extra = f', {lanes_used} Lane(s)' if lanes_used else ''
if externals:
extra += f', {n_msgflows} Nachrichtenfluss(e) zu {len(set(externals))} externen Beteiligten'
print(f'OK: {len(order)} Schritte, {n_edges} Kanten{extra} -> {args.output_bpmn}')
if __name__ == '__main__':
main()
Reference in New Issue View Git Blame Copy Permalink