From 1d5054a26887013deeb50652358b98e72aba7609 Mon Sep 17 00:00:00 2001 From: Michael Stangl Date: Wed, 8 Jul 2026 16:05:32 +0200 Subject: [PATCH] =?UTF-8?q?Library=20noch=20um=20die=20Auswertung=20der=20?= =?UTF-8?q?gemerkten=20globals2params=20erg=C3=A4nzt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/awl2scl.py | 187 ++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 152 insertions(+), 35 deletions(-) diff --git a/lib/awl2scl.py b/lib/awl2scl.py index 8c1b576..305c650 100644 --- a/lib/awl2scl.py +++ b/lib/awl2scl.py @@ -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 /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: /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