Schalter --improve-gotos und --globasl2params implementiert, um Code lesbarer zu machen. So kann man sicheren Lauffähigen Code in jedem Fall erzeugen, aber auch gleich die Lesbarkeit erhöhene
This commit is contained in:
+424
@@ -0,0 +1,424 @@
|
||||
#!/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
|
||||
Reference in New Issue
Block a user