Schoenenberger_Systeme_GmbH/HundM_awl
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Library noch um die Auswertung der gemerkten globals2params ergänzt

Browse Source
This commit is contained in:
Michael Stangl
2026-07-08 16:05:32 +02:00
parent 9663daf75b
commit 1d5054a268
1 changed files with 152 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/awl2scl.py
+152 -35
View File
@@ -25,6 +25,7 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import stlxml2awl as base # noqa: E402
import sclopt # noqa: E402
# ---------------------------------------------------------------------------
@@ -555,14 +556,15 @@ class Translator:
return f"#{name}"
def translate(self):
"""Liefert Liste (Netzwerktitel, IR-Knotenliste)."""
out_networks = []
for net in self.parsed.networks:
lines = self.translate_network(net)
out_networks.append((net.title, lines))
nodes = self.translate_network(net)
out_networks.append((net.title, nodes))
return out_networks
def translate_network(self, net):
lines = []
nodes = []
groups = [BracketGroup()] # Stack der Klammerebenen, [0] = Top-Level
pending_bracket_op = [] # je offener Klammer: (op, negate) fuer den Term bei ")"
akku = [] # symbolischer Wert-Stack (ACCU1 = akku[-1])
@@ -574,8 +576,8 @@ class Translator:
# liest z.B. ein direkt folgendes "JC" nach "S" dieselbe Bedingung).
fresh = [True]
def emit(s):
lines.append(s)
def add(node):
nodes.append(node)
def maybe_reset_top():
if len(groups) == 1 and fresh[0]:
@@ -590,7 +592,7 @@ class Translator:
for st in net.statements:
if st.label:
emit(f"{st.label}: ;")
add(sclopt.Label(st.label))
fresh[0] = True
mn = st.mnemonic
@@ -601,7 +603,7 @@ class Translator:
if mn == "NOP" or mn == "COMMENT":
if st.comment:
emit(f"// {st.comment}")
add(sclopt.Comment(st.comment))
continue
# --- RLO-Verknuepfung ---
@@ -648,23 +650,25 @@ class Translator:
if mn == "=":
expr = read_top()
fresh[0] = True
emit(f"{sclize_operand(op)} := {expr};" + (f" // {st.comment}" if st.comment else ""))
add(sclopt.Assign(sclize_operand(op), expr, st.comment))
continue
if mn == "S":
expr = read_top()
fresh[0] = True
assign = sclopt.Assign(sclize_operand(op), "TRUE")
if expr == "TRUE":
emit(f"{sclize_operand(op)} := TRUE;")
add(assign)
else:
emit(f"IF {expr} THEN {sclize_operand(op)} := TRUE; END_IF;")
add(sclopt.IfDo(expr, [assign]))
continue
if mn == "R":
expr = read_top()
fresh[0] = True
assign = sclopt.Assign(sclize_operand(op), "FALSE")
if expr == "TRUE":
emit(f"{sclize_operand(op)} := FALSE;")
add(assign)
else:
emit(f"IF {expr} THEN {sclize_operand(op)} := FALSE; END_IF;")
add(sclopt.IfDo(expr, [assign]))
continue
# --- Akku: Laden/Transferieren ---
@@ -674,7 +678,7 @@ class Translator:
if mn == "T":
if not akku:
raise UnsupportedConstruct("T ohne vorheriges L (Akku leer)")
emit(f"{sclize_operand(op)} := {akku[-1]};")
add(sclopt.Assign(sclize_operand(op), akku[-1]))
continue
if mn == "TAK":
if len(akku) < 2:
@@ -755,10 +759,10 @@ class Translator:
mem = sclize_operand(op)
tmp = self.new_temp("Flanke")
if mn == "Rise":
emit(f"{tmp} := {cond} AND NOT {mem};")
add(sclopt.Assign(tmp, f"{cond} AND NOT {mem}"))
else:
emit(f"{tmp} := NOT ({cond}) AND {mem};")
emit(f"{mem} := {cond};")
add(sclopt.Assign(tmp, f"NOT ({cond}) AND {mem}"))
add(sclopt.Assign(mem, cond))
groups[0] = BracketGroup()
groups[0].chain_expr = tmp
fresh[0] = False
@@ -767,37 +771,37 @@ class Translator:
# --- Statuswort/BR ---
if mn == "SAVE":
expr = read_top()
emit(f"ENO := {expr};")
add(sclopt.Assign("ENO", expr))
continue
# --- Spruenge ---
if mn in SUPPORTED_JUMP_TOKENS:
target = op.strip().rstrip(":")
if mn == "JU":
emit(f"GOTO {target};")
add(sclopt.Goto(target))
elif mn == "JC":
expr = read_top()
emit(f"IF {expr} THEN GOTO {target}; END_IF;")
add(sclopt.IfDo(expr, [sclopt.Goto(target)]))
else: # JCN
expr = read_top()
emit(f"IF NOT ({expr}) THEN GOTO {target}; END_IF;")
add(sclopt.IfDo(f"NOT ({expr})", [sclopt.Goto(target)]))
continue
if mn in UNSUPPORTED_JUMP_TOKENS:
raise UnsupportedConstruct(f"Sprungmnemonik {mn} (statuswortabhaengig) nicht unterstuetzt")
# --- Bausteinende ---
if mn in ("BE", "BEU"):
emit("RETURN;")
add(sclopt.Return())
continue
if mn == "BEC":
expr = read_top()
fresh[0] = True
emit(f"IF {expr} THEN RETURN; END_IF;")
add(sclopt.IfDo(expr, [sclopt.Return()]))
continue
# --- CALL ---
if mn == "CALL":
emit(translate_call(op) + (f" // {st.comment}" if st.comment else ""))
add(sclopt.Call(translate_call(op), st.comment))
continue
# --- OPN/AUF ---
@@ -815,7 +819,7 @@ class Translator:
raise UnsupportedConstruct(f"nicht unterstuetzte Mnemonik: {mn} {op}".strip())
return lines
return nodes
def _split_top_level(text, sep=","):
@@ -920,10 +924,10 @@ def render_scl(parsed, translated_networks, temp_vars):
out.append("")
out.append("BEGIN")
for title, lines in translated_networks:
for title, nodes in translated_networks:
label = title if title else ""
out.append(f" REGION {label}".rstrip())
for ln in lines:
for ln in sclopt.render_ir(nodes):
out.append(f" {ln}")
out.append(" END_REGION")
out.append("")
@@ -966,25 +970,41 @@ def default_log_path():
# Ergebnis-Datensatz + Log
# ---------------------------------------------------------------------------
GENERATED_MARKER = "// Mechanische AWL->SCL-Konvertierung (awl2scl.py)"
class Outcome:
TRANSLATED = "translated"
SKIPPED_EXISTS = "skipped_exists"
SKIPPED_PROTECTED = "skipped_protected"
SKIPPED_CATEGORY = "skipped_category"
MANUAL_REVIEW = "manual_review"
ERROR = "error"
class Result:
def __init__(self, path, outcome, category=None, reason="", scl_path=None, jumps=0):
def __init__(self, path, outcome, category=None, reason="", scl_path=None, jumps=0,
params=None, callers=None):
self.path = path
self.outcome = outcome
self.category = category
self.reason = reason
self.scl_path = scl_path
self.jumps = jumps
self.params = params or [] # neue Parameter (globals2params)
self.callers = callers or [] # (caller_name, status) fuer toChange
def process_file(awl_path, requested_categories, target_dir, console):
def _is_generated(scl_path):
"""True, wenn die vorhandene .scl von awl2scl erzeugt wurde (Marker-Kopfzeile)."""
try:
head = scl_path.read_text(encoding="utf-8", errors="replace")[:600]
except OSError:
return False
return GENERATED_MARKER in head
def process_file(awl_path, requested_categories, target_dir, console, opts):
try:
parsed = parse_awl_file(awl_path)
except NotABlockError as exc:
@@ -1007,8 +1027,12 @@ def process_file(awl_path, requested_categories, target_dir, console):
scl_path = target_dir / station_of(awl_path) / f"{parsed.name}.scl"
if scl_path.exists():
console(f"UEBERSPRUNGEN [Ziel existiert bereits] {awl_path} -> {scl_path}")
return Result(awl_path, Outcome.SKIPPED_EXISTS, category=category, scl_path=scl_path)
if not opts.force:
console(f"UEBERSPRUNGEN [Ziel existiert bereits] {awl_path} -> {scl_path}")
return Result(awl_path, Outcome.SKIPPED_EXISTS, category=category, scl_path=scl_path)
if not _is_generated(scl_path):
console(f"GESCHUETZT [handgeschrieben, nicht ueberschrieben] {scl_path}")
return Result(awl_path, Outcome.SKIPPED_PROTECTED, category=category, scl_path=scl_path)
translator = Translator(parsed, awl_path)
try:
@@ -1020,15 +1044,40 @@ def process_file(awl_path, requested_categories, target_dir, console):
console(f"FEHLER (Uebersetzung) {awl_path}: {exc}")
return Result(awl_path, Outcome.ERROR, category=category, reason=f"Uebersetzungsfehler: {exc}")
new_params = []
callers = []
if opts.globals2params:
try:
new_params, callers = opts.paramizer.parameterize(
parsed, translated, translator, station_of(awl_path), awl_path)
except Exception as exc:
console(f"FEHLER (globals2params) {awl_path}: {exc}")
return Result(awl_path, Outcome.ERROR, category=category,
reason=f"globals2params-Fehler: {exc}")
# Aufruf-Rewrites, die frueher fuer DIESEN Baustein vermerkt wurden, anwenden
if opts.globals2params:
opts.paramizer.apply_pending_call_rewrites(parsed, translated, station_of(awl_path))
if opts.improve_gotos:
translated = [(title, sclopt.improve_gotos(nodes)) for title, nodes in translated]
translated = sclopt.synth_case_across_networks(translated)
scl_text = render_scl(parsed, translated, translator.temp_vars)
scl_path.parent.mkdir(parents=True, exist_ok=True)
scl_path.write_text(scl_text, encoding="utf-8")
jumps = count_jumps(parsed)
console(f"OK [{category}] {awl_path} -> {scl_path}" + (f" (GOTO-lastig: {jumps} Spruenge)" if jumps > Translator.JUMP_HEAVY_THRESHOLD else ""))
return Result(awl_path, Outcome.TRANSLATED, category=category, scl_path=scl_path, jumps=jumps)
extra = ""
if jumps > Translator.JUMP_HEAVY_THRESHOLD and not opts.improve_gotos:
extra = f" (GOTO-lastig: {jumps} Spruenge)"
if new_params:
extra += f" (+{len(new_params)} Parameter, {len(callers)} Aufrufer)"
console(f"OK [{category}] {awl_path} -> {scl_path}{extra}")
return Result(awl_path, Outcome.TRANSLATED, category=category, scl_path=scl_path,
jumps=jumps, params=new_params, callers=callers)
def write_log(log_path, results, requested_categories, target_dir):
def write_log(log_path, results, requested_categories, target_dir, opts=None):
lines = []
ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
lines.append(f"AWL -> SCL Konvertierung ({ts})")
@@ -1036,14 +1085,31 @@ def write_log(log_path, results, requested_categories, target_dir):
lines.append("")
lines.append(f"Angeforderte Kategorie(n) : {','.join(sorted(requested_categories))}")
lines.append(f"Zielordner : {target_dir}")
if opts is not None:
lines.append(f"--improve-gotos : {'ja' if opts.improve_gotos else 'nein'}")
lines.append(f"--globals2params : {'ja' if opts.globals2params else 'nein'}")
lines.append(f"Dateien gesamt : {len(results)}")
lines.append(f"Uebersetzt : {sum(1 for r in results if r.outcome == Outcome.TRANSLATED)}")
lines.append(f"Uebersprungen (Kategorie) : {sum(1 for r in results if r.outcome == Outcome.SKIPPED_CATEGORY)}")
lines.append(f"Uebersprungen (Ziel da) : {sum(1 for r in results if r.outcome == Outcome.SKIPPED_EXISTS)}")
lines.append(f"Geschuetzt (handgeschr.) : {sum(1 for r in results if r.outcome == Outcome.SKIPPED_PROTECTED)}")
lines.append(f"Manuelle Pruefung noetig : {sum(1 for r in results if r.outcome == Outcome.MANUAL_REVIEW)}")
lines.append(f"Fehler : {sum(1 for r in results if r.outcome == Outcome.ERROR)}")
lines.append("")
parametrized = [r for r in results if r.outcome == Outcome.TRANSLATED and r.params]
if parametrized:
lines.append("-" * 70)
lines.append(f"Parametrisiert (globals2params) ({len(parametrized)})")
lines.append("-" * 70)
for r in sorted(parametrized, key=lambda r: str(r.path)):
n_manual = sum(1 for _, status in r.callers if status == "manual")
lines.append(f"{r.scl_path}")
for g, name, direction in r.params:
lines.append(f" {direction:6s} {name} (war {g})")
lines.append(f" toChange: {len(r.callers)} Aufrufer, davon {n_manual} manuell")
lines.append("")
manual = [r for r in results if r.outcome == Outcome.MANUAL_REVIEW]
if manual:
lines.append("-" * 70)
@@ -1081,6 +1147,15 @@ def write_log(log_path, results, requested_categories, target_dir):
lines.append(f"[{r.category}] {r.path} ({r.reason})")
lines.append("")
protected = [r for r in results if r.outcome == Outcome.SKIPPED_PROTECTED]
if protected:
lines.append("-" * 70)
lines.append(f"Geschuetzt, handgeschriebene .scl nicht ueberschrieben ({len(protected)})")
lines.append("-" * 70)
for r in sorted(protected, key=lambda r: str(r.path)):
lines.append(f"{r.scl_path}")
lines.append("")
skipped_exists = [r for r in results if r.outcome == Outcome.SKIPPED_EXISTS]
if skipped_exists:
lines.append("-" * 70)
@@ -1098,6 +1173,18 @@ def write_log(log_path, results, requested_categories, target_dir):
# CLI
# ---------------------------------------------------------------------------
class Options:
def __init__(self, improve_gotos=False, globals2params=False, force=False, paramizer=None):
self.improve_gotos = improve_gotos
self.globals2params = globals2params
self.force = force
self.paramizer = paramizer
def default_tochange_path():
return Path(__file__).resolve().parent.parent / "data" / "globals2params_tochange.json"
def main(argv=None):
parser = argparse.ArgumentParser(description="Wandelt .awl-Dateien mechanisch in .scl-Dateien um.")
parser.add_argument("files", nargs="*", type=Path, help="Einzelne .awl-Dateien")
@@ -1107,6 +1194,14 @@ def main(argv=None):
help="Kommaliste der zu uebersetzenden Kategorien, z.B. A oder A,B (Default: A)")
parser.add_argument("--target-dir", type=Path, default=None,
help="Zielordner fuer .scl-Dateien (Default: $PV_SCL_EXPORT oder <Projekt>/SCL-Export)")
parser.add_argument("--improve-gotos", action="store_true",
help="Kontrollfluss aufraeumen (Guard->IF, dedup, CASE, Klammern) - verlustfrei")
parser.add_argument("--globals2params", action="store_true",
help="Globale Merker-Zugriffe in Parameter umbauen + toChange fuer Aufrufer (data/)")
parser.add_argument("--force", action="store_true",
help="Vorhandene generierte .scl ueberschreiben (handgeschriebene bleiben geschuetzt)")
parser.add_argument("--tochange", type=Path, default=None,
help="Pfad der toChange-Liste (Default: <Projekt>/data/globals2params_tochange.json)")
parser.add_argument("--log", type=Path, help="Pfad der Log-Datei")
args = parser.parse_args(argv)
@@ -1119,21 +1214,43 @@ def main(argv=None):
requested_categories = {c.strip().upper() for c in args.category.split(",") if c.strip()}
target_dir = args.target_dir or default_target_dir()
# globals2params impliziert Regenerierung
force = args.force or args.globals2params or args.improve_gotos
store = None
if args.globals2params:
import paramize
roots = [args.directory] if args.directory else [t.parent for t in targets]
roots = list(dict.fromkeys(str(Path(r)) for r in roots))
store = paramize.ToChangeStore(args.tochange or default_tochange_path())
paramizer = paramize.Paramizer(paramize.CallerIndex(roots), store)
else:
import paramize
paramizer = paramize.NullParamizer()
opts = Options(improve_gotos=args.improve_gotos, globals2params=args.globals2params,
force=force, paramizer=paramizer)
def console(msg):
print(msg)
results = []
for awl_path in targets:
results.append(process_file(awl_path, requested_categories, target_dir, console))
results.append(process_file(awl_path, requested_categories, target_dir, console, opts))
if store is not None:
store.save()
log_path = args.log or default_log_path()
write_log(log_path, results, requested_categories, target_dir)
write_log(log_path, results, requested_categories, target_dir, opts)
n_ok = sum(1 for r in results if r.outcome == Outcome.TRANSLATED)
n_manual = sum(1 for r in results if r.outcome == Outcome.MANUAL_REVIEW)
n_err = sum(1 for r in results if r.outcome == Outcome.ERROR)
print()
print(f"{len(results)} Dateien geprueft, {n_ok} uebersetzt, {n_manual} manuelle Pruefung, {n_err} Fehler.")
if args.globals2params and store is not None:
print(f"toChange-Liste: {store.path}")
print(f"Log: {log_path}")
return 1 if n_err else 0