425 lines
15 KiB
Python
425 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""globals2params: baut globale Merker-Zugriffe eines Bausteins in echte
|
|
Parameter um und verwaltet eine persistente toChange-Liste in data/ fuer die
|
|
Anpassung der Aufrufer (auch noch nicht konvertierter .awl-Dateien).
|
|
|
|
Verhalten (verhaltenserhaltend by construction): ein Global, das der Baustein
|
|
bisher direkt las/schrieb, wird Parameter; jeder Aufrufer forwardet exakt
|
|
dasselbe Global an den Parameter (Input ':=', Output/InOut '=>'). Damit bleibt
|
|
die Semantik identisch, solange der Aufrufer das Global vor/nach dem Aufruf
|
|
wie bisher nutzt.
|
|
|
|
Grenzen: indirekte Aufrufe (UC/CC FC[var]) koennen nicht automatisch
|
|
umgeschrieben werden -> in toChange als 'manual' vermerkt.
|
|
"""
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import sclopt
|
|
|
|
|
|
# Globale Operanden: symbolisch ("...") oder absolut (M/E/A/PE/PA + Groesse).
|
|
# Lokale (#...), Literale, Timer/Zaehler (T/C mit Nummer) sind KEINE Kandidaten.
|
|
_ABS_GLOBAL_RE = re.compile(r'^(M|E|A|PE|PA|I|Q)(X|B|W|D)?\d+(\.\d+)?$')
|
|
_SYM_GLOBAL_RE = re.compile(r'^"[^"]+"$')
|
|
|
|
|
|
def is_global_operand(tok):
|
|
tok = tok.strip()
|
|
if not tok or tok.startswith("#"):
|
|
return False
|
|
if _SYM_GLOBAL_RE.match(tok):
|
|
return True
|
|
return bool(_ABS_GLOBAL_RE.match(tok))
|
|
|
|
|
|
def param_name_for(global_tok):
|
|
"""Deterministischer, SCL-gueltiger Parametername aus dem Global-Token."""
|
|
core = global_tok.strip().strip('"')
|
|
name = re.sub(r"[^A-Za-z0-9_]", "_", core)
|
|
if not re.match(r"^[A-Za-z_]", name):
|
|
name = "g_" + name
|
|
return name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Richtungsanalyse ueber die IR-Knoten
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TOKEN_RE = re.compile(r'"[^"]+"|#?[A-Za-z_]\w*(?:\.\w+)*|\b\d+\b')
|
|
|
|
|
|
def _iter_read_tokens(expr):
|
|
"""Alle Operanden-Token eines SCL-Ausdrucks (fuer Lese-Erkennung)."""
|
|
if not expr:
|
|
return
|
|
for m in re.finditer(r'"[^"]+"|#[A-Za-z_][\w.]*|[A-Za-z_][\w.]*', expr):
|
|
yield m.group(0)
|
|
|
|
|
|
def _base_token(tok):
|
|
"""Strukturzugriff auf Basis reduzieren: '"DB".Feld' -> '"DB"'; 'M10.0' bleibt."""
|
|
tok = tok.strip()
|
|
if tok.startswith('"'):
|
|
end = tok.find('"', 1)
|
|
return tok[:end + 1] if end != -1 else tok
|
|
return tok
|
|
|
|
|
|
_CALL_HEAD_RE = re.compile(r'^\s*(?:#\w+|"[^"]+")\s*\(')
|
|
|
|
|
|
def _call_param_values(call_text):
|
|
"""Liefert die Parameter-WERTE (rechts von := / =>) eines SCL-Aufrufs.
|
|
Zielbaustein und Parameternamen werden ausgelassen."""
|
|
m = _CALL_HEAD_RE.match(call_text)
|
|
if not m:
|
|
return
|
|
inner = call_text[m.end():]
|
|
depth = 1
|
|
end = 0
|
|
dq = sq = False
|
|
for i, ch in enumerate(inner):
|
|
if ch == '"' and not sq:
|
|
dq = not dq
|
|
elif ch == "'" and not dq:
|
|
sq = not sq
|
|
elif ch in "([" and not dq and not sq:
|
|
depth += 1
|
|
elif ch in ")]" and not dq and not sq:
|
|
depth -= 1
|
|
if depth == 0:
|
|
end = i
|
|
break
|
|
inner = inner[:end]
|
|
for piece in _split_commas(inner):
|
|
for sep in (":=", "=>"):
|
|
if sep in piece:
|
|
yield piece.split(sep, 1)[1].strip()
|
|
break
|
|
|
|
|
|
def _split_commas(text):
|
|
parts, buf, depth, dq, sq = [], [], 0, False, False
|
|
for ch in text:
|
|
if ch == '"' and not sq:
|
|
dq = not dq
|
|
elif ch == "'" and not dq:
|
|
sq = not sq
|
|
elif ch in "([" and not dq and not sq:
|
|
depth += 1
|
|
elif ch in ")]" and not dq and not sq:
|
|
depth -= 1
|
|
if ch == "," and depth == 0 and not dq and not sq:
|
|
parts.append("".join(buf)); buf = []; continue
|
|
buf.append(ch)
|
|
parts.append("".join(buf))
|
|
return parts
|
|
|
|
|
|
def analyze_directions(nodes, reads, writes):
|
|
"""Fuellt reads/writes (Sets globaler Basis-Token) durch Rekursion ueber die IR."""
|
|
for n in nodes:
|
|
if isinstance(n, sclopt.Assign):
|
|
base = _base_token(n.lhs)
|
|
if is_global_operand(base):
|
|
writes.add(base)
|
|
for t in _iter_read_tokens(n.rhs):
|
|
b = _base_token(t)
|
|
if is_global_operand(b):
|
|
reads.add(b)
|
|
elif isinstance(n, sclopt.IfDo):
|
|
for t in _iter_read_tokens(n.cond):
|
|
b = _base_token(t)
|
|
if is_global_operand(b):
|
|
reads.add(b)
|
|
analyze_directions(n.body, reads, writes)
|
|
elif isinstance(n, sclopt.Case):
|
|
b = _base_token(n.selector)
|
|
if is_global_operand(b):
|
|
reads.add(b)
|
|
for _, body, _ in n.branches:
|
|
analyze_directions(body, reads, writes)
|
|
elif isinstance(n, sclopt.Call):
|
|
# Nur die Parameter-WERTE eines Aufrufs sind Global-Kandidaten,
|
|
# nicht der Zielbaustein und nicht die Parameternamen.
|
|
for value in _call_param_values(n.text):
|
|
b = _base_token(value)
|
|
if is_global_operand(b):
|
|
reads.add(b)
|
|
writes.add(b) # Aufrufparameter: Richtung unklar -> konservativ InOut
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# IR-Umschreibung: Global-Token -> #Parameter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _replace_in_expr(expr, mapping):
|
|
if not expr:
|
|
return expr
|
|
def repl(m):
|
|
tok = m.group(0)
|
|
base = _base_token(tok)
|
|
if base in mapping:
|
|
return mapping[base] + tok[len(base):]
|
|
return tok
|
|
return re.sub(r'"[^"]+"(?:\.\w+)*|#?[A-Za-z_][\w.]*', repl, expr)
|
|
|
|
|
|
def _rewrite_call_values(call_text, mapping):
|
|
"""Ersetzt in einem Aufruf nur die Parameter-Werte (rechts von :=/=>)."""
|
|
m = _CALL_HEAD_RE.match(call_text)
|
|
if not m:
|
|
return call_text
|
|
head = call_text[:m.end()]
|
|
rest = call_text[m.end():]
|
|
depth = 1
|
|
end = 0
|
|
dq = sq = False
|
|
for i, ch in enumerate(rest):
|
|
if ch == '"' and not sq:
|
|
dq = not dq
|
|
elif ch == "'" and not dq:
|
|
sq = not sq
|
|
elif ch in "([" and not dq and not sq:
|
|
depth += 1
|
|
elif ch in ")]" and not dq and not sq:
|
|
depth -= 1
|
|
if depth == 0:
|
|
end = i
|
|
break
|
|
inner, tail = rest[:end], rest[end:]
|
|
new_pieces = []
|
|
for piece in _split_commas(inner):
|
|
for sep in (":=", "=>"):
|
|
if sep in piece:
|
|
name, value = piece.split(sep, 1)
|
|
new_pieces.append(f"{name}{sep}{_replace_in_expr(value, mapping)}")
|
|
break
|
|
else:
|
|
new_pieces.append(piece)
|
|
return head + ",".join(new_pieces) + tail
|
|
|
|
|
|
def rewrite_nodes(nodes, mapping):
|
|
for n in nodes:
|
|
if isinstance(n, sclopt.Assign):
|
|
n.lhs = _replace_in_expr(n.lhs, mapping)
|
|
n.rhs = _replace_in_expr(n.rhs, mapping)
|
|
elif isinstance(n, sclopt.IfDo):
|
|
n.cond = _replace_in_expr(n.cond, mapping)
|
|
rewrite_nodes(n.body, mapping)
|
|
elif isinstance(n, sclopt.Case):
|
|
n.selector = _replace_in_expr(n.selector, mapping)
|
|
for _, body, _ in n.branches:
|
|
rewrite_nodes(body, mapping)
|
|
elif isinstance(n, sclopt.Call):
|
|
n.text = _rewrite_call_values(n.text, mapping)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Aufrufer-Suche in allen .awl-Quellen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _station_of(path):
|
|
for part in Path(path).parts:
|
|
if part.startswith("=A"):
|
|
return part
|
|
return "_"
|
|
|
|
|
|
class CallerIndex:
|
|
"""Findet direkte und indirekte Aufrufer je Bausteinname, ueber .awl-Quellen."""
|
|
|
|
def __init__(self, roots):
|
|
self.direct = {} # name -> list[(station, caller_name, path)]
|
|
self.indirect = {} # name -> list[(station, caller_name, path)] (UC/CC FC[..])
|
|
self._build(roots)
|
|
|
|
def _build(self, roots):
|
|
awl_files = []
|
|
for root in roots:
|
|
awl_files.extend(Path(root).rglob("*.awl"))
|
|
direct_re = re.compile(r'\bCALL\s+(?:FB|FC|SFB|SFC)?\s*"([^"]+)"')
|
|
indirect_re = re.compile(r'\b(?:UC|CC)\s+FC\[')
|
|
header_re = re.compile(r'===\s*\w+\s+\S*:\s*(.+?)\s*\[STL\]\s*===|'
|
|
r'^(?:FUNCTION_BLOCK|FUNCTION|ORGANIZATION_BLOCK)\s+"([^"]+)"')
|
|
for p in awl_files:
|
|
try:
|
|
text = p.read_text(encoding="utf-8-sig", errors="replace")
|
|
except OSError:
|
|
continue
|
|
hm = header_re.search(text)
|
|
caller = (hm.group(1) or hm.group(2)) if hm else p.stem
|
|
station = _station_of(p)
|
|
for m in direct_re.finditer(text):
|
|
self.direct.setdefault(m.group(1), []).append((station, caller, str(p)))
|
|
if indirect_re.search(text):
|
|
self.indirect.setdefault("*", []).append((station, caller, str(p)))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# toChange-Persistenz
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ToChangeStore:
|
|
"""data/globals2params_tochange.json:
|
|
{ "<station>|<caller>": { "path":..., "changes":[ {func, call, status} ] } }"""
|
|
|
|
def __init__(self, path):
|
|
self.path = Path(path)
|
|
self.data = {}
|
|
if self.path.exists():
|
|
try:
|
|
self.data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
self.data = {}
|
|
|
|
@staticmethod
|
|
def key(station, caller):
|
|
return f"{station}|{caller}"
|
|
|
|
def add_change(self, station, caller, path, func, call_text, status):
|
|
k = self.key(station, caller)
|
|
entry = self.data.setdefault(k, {"path": path, "changes": []})
|
|
for ch in entry["changes"]:
|
|
if ch["func"] == func:
|
|
ch["call"] = call_text
|
|
ch["status"] = status
|
|
return
|
|
entry["changes"].append({"func": func, "call": call_text, "status": status})
|
|
|
|
def pending_for(self, station, caller):
|
|
entry = self.data.get(self.key(station, caller))
|
|
if not entry:
|
|
return []
|
|
return [ch for ch in entry["changes"] if ch.get("status") == "pending"]
|
|
|
|
def mark_applied(self, station, caller, func):
|
|
entry = self.data.get(self.key(station, caller))
|
|
if not entry:
|
|
return
|
|
for ch in entry["changes"]:
|
|
if ch["func"] == func:
|
|
ch["status"] = "applied"
|
|
|
|
def save(self):
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.path.write_text(json.dumps(self.data, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Orchestrierung
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Paramizer:
|
|
def __init__(self, caller_index, store):
|
|
self.caller_index = caller_index
|
|
self.store = store
|
|
|
|
def parameterize(self, parsed, translated, translator, station, awl_path):
|
|
"""Baut globale Zugriffe des Bausteins in Parameter um; erzeugt toChange
|
|
fuer alle Aufrufer. Liefert (new_params, callers)."""
|
|
reads, writes = set(), set()
|
|
for _, nodes in translated:
|
|
analyze_directions(nodes, reads, writes)
|
|
globals_used = reads | writes
|
|
if not globals_used:
|
|
return [], []
|
|
|
|
# Richtung + Parametername je Global
|
|
params = [] # (global, name, direction)
|
|
mapping = {} # global-basis -> "#name"
|
|
used_names = set()
|
|
for g in sorted(globals_used):
|
|
r, w = g in reads, g in writes
|
|
direction = "InOut" if (r and w) else ("Output" if w else "Input")
|
|
name = param_name_for(g)
|
|
while name in used_names:
|
|
name += "_"
|
|
used_names.add(name)
|
|
params.append((g, name, direction))
|
|
mapping[g] = "#" + name
|
|
|
|
# Interface ergaenzen (Kommentar: war "<global>")
|
|
for g, name, direction in params:
|
|
sect = {"Input": "Input", "Output": "Output", "InOut": "InOut"}[direction]
|
|
parsed.sections.setdefault(sect, []).append(f'{name} : {_scl_type_for(g)}; // war {g}')
|
|
|
|
# Rumpf umschreiben
|
|
for _, nodes in translated:
|
|
rewrite_nodes(nodes, mapping)
|
|
|
|
# Aufrufer + toChange
|
|
callers = self._record_callers(parsed.name, params)
|
|
return params, callers
|
|
|
|
def _record_callers(self, func_name, params):
|
|
results = []
|
|
call_text = _build_param_call(func_name, params)
|
|
for station, caller, path in self.caller_index.direct.get(func_name, []):
|
|
self.store.add_change(station, caller, path, func_name, call_text, "pending")
|
|
results.append((f"{station}|{caller}", "pending"))
|
|
# indirekte Aufrufe (UC/CC FC[..]) -> manuell
|
|
for station, caller, path in self.caller_index.indirect.get("*", []):
|
|
self.store.add_change(station, caller, path, func_name,
|
|
f"// MANUELL: indirekter Aufruf von {func_name} auf Parameter umstellen",
|
|
"manual")
|
|
results.append((f"{station}|{caller}", "manual"))
|
|
return results
|
|
|
|
def apply_pending_call_rewrites(self, parsed, translated, station):
|
|
"""Wendet frueher vermerkte Aufruf-Rewrites an, wenn DIESER Baustein
|
|
(als Aufrufer) jetzt konvertiert wird."""
|
|
pending = self.store.pending_for(station, parsed.name)
|
|
if not pending:
|
|
return
|
|
by_func = {ch["func"]: ch["call"] for ch in pending}
|
|
for _, nodes in translated:
|
|
_rewrite_calls_in_nodes(nodes, by_func)
|
|
for ch in pending:
|
|
self.store.mark_applied(station, parsed.name, ch["func"])
|
|
|
|
|
|
def _scl_type_for(global_tok):
|
|
"""Grobe Typableitung aus dem Global-Token (mechanisch, im Zweifel Bool/Word)."""
|
|
core = global_tok.strip().strip('"')
|
|
m = re.match(r'^(M|E|A|PE|PA|I|Q)(X|B|W|D)?', core)
|
|
if m:
|
|
return {"X": "Bool", "B": "Byte", "W": "Word", "D": "DWord", None: "Bool"}[m.group(2)]
|
|
if "." in core or re.search(r"\d+\.\d+$", core):
|
|
return "Bool"
|
|
return "Word"
|
|
|
|
|
|
def _build_param_call(func_name, params):
|
|
parts = []
|
|
for g, name, direction in params:
|
|
sep = ":=" if direction == "Input" else "=>"
|
|
parts.append(f"{name} {sep} {g}")
|
|
return f'"{func_name}"(' + ", ".join(parts) + ");"
|
|
|
|
|
|
def _rewrite_calls_in_nodes(nodes, by_func):
|
|
call_name_re = re.compile(r'^"([^"]+)"\s*\(')
|
|
for i, n in enumerate(nodes):
|
|
if isinstance(n, sclopt.Call):
|
|
m = call_name_re.match(n.text.strip())
|
|
if m and m.group(1) in by_func:
|
|
nodes[i] = sclopt.Call(by_func[m.group(1)], n.comment)
|
|
elif isinstance(n, sclopt.IfDo):
|
|
_rewrite_calls_in_nodes(n.body, by_func)
|
|
elif isinstance(n, sclopt.Case):
|
|
for _, body, _ in n.branches:
|
|
_rewrite_calls_in_nodes(body, by_func)
|
|
|
|
|
|
class NullParamizer:
|
|
"""Platzhalter, wenn --globals2params aus ist (keine Umbauten)."""
|
|
def parameterize(self, *a, **k):
|
|
return [], []
|
|
|
|
def apply_pending_call_rewrites(self, *a, **k):
|
|
return
|