#!/usr/bin/env python3 """Wandelt .awl-Dateien (STL/AWL) mechanisch in .scl-Dateien um. Liest sowohl die von stlxml2awl.py gerenderten .awl-Dateien (Kopf "=== FB 92: Name [STL] ===") als auch native STEP7/TIA AWL-Quelldateien (Kopf 'FUNCTION_BLOCK "Name"', BEGIN/NETWORK, klassische deutsche Mnemonik U/UN/O/ON/SPA/SPB/AUF/...). Die Kategorisierung (A/B/C) wird je Datei selbst neu bestimmt (nicht aus einem Log uebernommen), mit denselben Kriterien wie in stlxml2awl.py (Namensmuster aus SCL-Export/README.md + Sprung-/Pointer-Analyse) -- Complexity/classify/classify_by_name werden von dort importiert. Aufruf ueber bin/awl2scl.bat: awl2scl.py ... REM einzelne Dateien awl2scl.py --dir REM rekursiv alle .awl awl2scl.py --category A[,B,C] --target-dir --dir . """ import argparse import datetime import os import re import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import stlxml2awl as base # noqa: E402 # --------------------------------------------------------------------------- # Statement-Modell (gemeinsam fuer beide Quellformate) # --------------------------------------------------------------------------- class Statement: __slots__ = ("label", "mnemonic", "operand", "comment") def __init__(self, label, mnemonic, operand, comment): self.label = label self.mnemonic = mnemonic self.operand = operand self.comment = comment def __repr__(self): return f"Statement({self.label!r}, {self.mnemonic!r}, {self.operand!r})" class Network: def __init__(self, title, statements): self.title = title self.statements = statements class ParsedBlock: def __init__(self, btype, name, num, sections, networks, source_format): self.btype = btype # FB/FC/OB self.name = name self.num = num self.sections = sections # dict section-name -> list[str] (SCL-fertige Member-Zeilen) self.networks = networks # list[Network] self.source_format = source_format # "rendered" | "native" class NotABlockError(Exception): """Datei ist keine Baustein-AWL (z.B. leere/fremde Datei).""" # --------------------------------------------------------------------------- # Mnemonik-Tabellen (nativ -> kanonisch wie im gerenderten Format) # --------------------------------------------------------------------------- NATIVE_TO_CANON = { "U": "A", "UN": "AN", "O": "O", "ON": "ON", "X": "X", "XN": "XN", "U(": "A(", "UN(": "AN(", "O(": "O(", "ON(": "ON(", "X(": "X(", "XN(": "XN(", "SPA": "JU", "SPB": "JC", "SPBN": "JCN", "BEA": "BEU", "BEB": "BEC", "FP": "Rise", "FN": "Fall", "AUF": "OPN", } # Sprungmnemonik, die auf Statuswortbits (CC0/CC1/OV/OS/BR) beruht und mangels # Statuswort-Nachbildung nicht sicher uebersetzt wird (zaehlt aber als Sprung). UNSUPPORTED_JUMP_TOKENS = { "JL", "JP", "JBI", "JNBI", "JZ", "JN", "JM", "JMZ", "JPZ", "JO", "JOS", "JUO", "LOOP", "SPZ", "SPN", "SPP", "SPM", "SPPZ", "SPMZ", "SPO", "SPS", "SPBI", "SPBIN", "SPL", } SUPPORTED_JUMP_TOKENS = {"JU", "JC", "JCN"} NATIVE_POINTER_ARITH = {"LAR1", "LAR2", "TAR1", "TAR2", "+AR1", "+AR2", "ADDAR1", "ADDAR2"} TIMER_TOKENS = {"SI", "SE", "SS", "SA", "SD"} COUNTER_TOKENS = {"ZV", "ZR"} BITWISE_WORD_TOKENS = { "UW": "AND", "OW": "OR", "XW": "XOR", "UD": "AND", "OD": "OR", "XD": "XOR", "AW": "AND", "AD": "AND", # Schreibweise im gerenderten Format (stlxml2awl.py) } SHIFT_TOKENS = {"SLW": "SHL", "SRW": "SHR", "SLD": "SHL", "SRD": "SHR", "RLD": "ROL", "RRD": "ROR"} ARITH_TOKENS = { "+I": "+", "-I": "-", "*I": "*", "/I": "/", "+D": "+", "-D": "-", "*D": "*", "/D": "/", "+R": "+", "-R": "-", "*R": "*", "/R": "/", "+": "+", } COMPARE_TOKENS = { "==I": "=", "<>I": "<>", ">I": ">", "=I": ">=", "<=I": "<=", "==D": "=", "<>D": "<>", ">D": ">", "=D": ">=", "<=D": "<=", "==R": "=", "<>R": "<>", ">R": ">", "=R": ">=", "<=R": "<=", } CONVERT_TOKENS = {"ITD": "INT_TO_DINT", "BTI": "BCD16_TO_INT", "ITB": "INT_TO_BCD16"} # Registerindirekte/zeigerbasierte Operanden in zwei Notationen: # - gerendert (stlxml2awl.py): "MW [AR1,P#3.0]" (Leerzeichen vor "[", Register+Offset per Komma) # - nativ (STEP7-Quelle): "DBW[ #AR1Zeiger]" (kein Leerzeichen vor "[", Zeigerausdruck in Klammer) INDIRECT_OPERAND_RE = re.compile( r"(\[\s*AR[12]\s*,)" r"|(\b(?:DBX|DBW|DBD|DBB|DB|EB|EW|ED|E|AB|AW|AD|A|MB|MW|MD|M|PEB|PEW|PED|PAB|PAW|PAD)\[)" ) def is_indirect_operand(op): return bool(op) and bool(INDIRECT_OPERAND_RE.search(op)) # --------------------------------------------------------------------------- # Formaterkennung # --------------------------------------------------------------------------- def detect_format(text): head = text.lstrip(" \t\r\n")[:20] if head.startswith("==="): return "rendered" if head.startswith(("FUNCTION_BLOCK", "FUNCTION", "ORGANIZATION_BLOCK", "DATA_BLOCK")): return "native" raise NotABlockError(f"unbekanntes .awl-Format (Kopf: {head!r})") # --------------------------------------------------------------------------- # Parser: gerendertes Format (stlxml2awl.py) # --------------------------------------------------------------------------- HEADER_RE = re.compile(r"^===\s*(\w+)\s+(\S*):\s*(.+?)\s*\[STL\]\s*===\s*$") NW_RE = re.compile(r"^---\s*NW\s*\d+:\s*(.*?)\s*---\s*$") STMT_RE = re.compile(r"^(?:(\S+:)\s+)?(\S+)(?:\s+(.*))?$") MEMBER_RE = re.compile(r'^(\s*)(\S+|"[^"]+")\s*:\s*(\S+)(?:\s*:=\s*(\S+))?\s*(?:$|(?<=\S)\s{2,}//\s*(.*)$)') def parse_rendered(text): lines = text.splitlines() i = 0 m = HEADER_RE.match(lines[0].strip()) if not m: raise NotABlockError("Kopfzeile nicht erkannt (gerendertes Format)") btype, num, name = m.group(1), m.group(2), m.group(3) i = 1 sections = {} if i < len(lines) and lines[i].strip() == "INTERFACE:": i += 1 i, sections = parse_rendered_interface(lines, i) networks = [] while i < len(lines): line = lines[i] mnw = NW_RE.match(line.strip()) if mnw: title = mnw.group(1) i += 1 stmts = [] while i < len(lines) and not NW_RE.match(lines[i].strip()): raw = lines[i] i += 1 if not raw.strip(): continue stmts.append(parse_rendered_statement(raw)) networks.append(Network(title, stmts)) else: i += 1 return ParsedBlock(btype, name, num, sections, networks, "rendered") def parse_rendered_statement(raw): line = raw.rstrip("\n") if " // " in line: code, comment = line.split(" // ", 1) code = code.rstrip() else: code, comment = line.rstrip(), "" code = code.strip() if not code: return Statement(None, "NOP", "", comment) if code.startswith("// "): return Statement(None, "COMMENT", "", code[3:] + (" | " + comment if comment else "")) m = STMT_RE.match(code) if not m: return Statement(None, code, "", comment) label = m.group(1)[:-1] if m.group(1) else None mnemonic = m.group(2) if mnemonic.endswith(":") and m.group(3) is None: # reine Marke ohne Instruktion, z.B. "StatEnde:" return Statement(mnemonic[:-1], "NOP", "", comment) operand = (m.group(3) or "").strip() return Statement(label, mnemonic, operand, comment) def parse_rendered_interface(lines, i): """Parst den eingerueckten INTERFACE:-Block in flache SCL-Member-Zeilen je Section.""" sections = {} current_section = None # Stack von (indent, member-dict) fuer verschachtelte Structs stack = [] # list of dicts: {"indent":n, "lines":[...], "open_struct":bool} def close_to(indent): while stack and stack[-1]["indent"] >= indent: top = stack.pop() if top["open_struct"]: target = stack[-1]["lines"] if stack else sections[current_section] target.append(" " * (top["indent"]) + "END_STRUCT;") while i < len(lines): line = lines[i] if not line.strip(): i += 1 continue stripped = line.strip() indent = len(line) - len(line.lstrip(" ")) if indent <= 2 and stripped.endswith(":") and re.match(r"^\w+:$", stripped): close_to(0) current_section = stripped[:-1] sections[current_section] = [] stack = [{"indent": 2, "lines": sections[current_section], "open_struct": False}] i += 1 continue if current_section is None: break if indent < 4: # Ende des INTERFACE-Blocks (naechster Top-Level-Inhalt, z.B. "--- NW") break close_to(indent) mm = MEMBER_RE.match(line) if not mm: i += 1 continue m_indent, m_name, m_type, m_default, m_comment = mm.groups() target = stack[-1]["lines"] out_line = f"{m_name} : {m_type}" if m_default: out_line += f" := {m_default}" out_line += ";" if m_comment: out_line += f" // {m_comment}" target.append(out_line) if m_type == "Struct": new_lines = [] target[-1] = out_line.replace(";", "") stack.append({"indent": indent, "lines": new_lines, "open_struct": True}) # Nachfolgende Zeilen dieser Struct werden separat gesammelt und # beim Schliessen an target angehaengt -> daher hier umverdrahten: stack[-2] = stack[-2] # no-op, Klarheit i += 1 close_to(0) return i, sections # --------------------------------------------------------------------------- # Parser: natives STEP7/TIA AWL-Quellformat # --------------------------------------------------------------------------- NATIVE_HEADER_RE = re.compile(r'^(FUNCTION_BLOCK|FUNCTION|ORGANIZATION_BLOCK)\s+"([^"]+)"(?:\s*:\s*(\S+))?') VAR_SECTION_RE = re.compile(r"^(VAR_INPUT|VAR_OUTPUT|VAR_IN_OUT|VAR_TEMP|VAR)\b") def parse_native(text): text = text.lstrip("") lines = text.splitlines() i = 0 mh = NATIVE_HEADER_RE.match(lines[0].strip()) if not mh: raise NotABlockError("Kopfzeile nicht erkannt (natives Format)") btype = {"FUNCTION_BLOCK": "FB", "FUNCTION": "FC", "ORGANIZATION_BLOCK": "OB"}[mh.group(1)] name = mh.group(2) i = 1 sections = {} while i < len(lines) and lines[i].strip() != "BEGIN": mv = VAR_SECTION_RE.match(lines[i].strip()) if mv: sect_raw = mv.group(1) sect = {"VAR": "Static", "VAR_INPUT": "Input", "VAR_OUTPUT": "Output", "VAR_IN_OUT": "InOut", "VAR_TEMP": "Temp"}[sect_raw] i += 1 body = [] while i < len(lines) and lines[i].strip() != "END_VAR": body.append(lines[i].rstrip()) i += 1 i += 1 # END_VAR sections.setdefault(sect, []).extend(_clean_native_member_lines(body)) else: i += 1 if i < len(lines) and lines[i].strip() == "BEGIN": i += 1 networks = [] body_text = "\n".join(lines[i:]) body_text = re.sub(r"\bEND_(FUNCTION_BLOCK|FUNCTION|ORGANIZATION_BLOCK)\b.*$", "", body_text, flags=re.DOTALL) for block in re.split(r"(?m)^NETWORK\s*$", body_text)[1:]: block_lines = block.splitlines() title = "" start = 0 for j, bl in enumerate(block_lines): if bl.strip().startswith("TITLE"): title = bl.split("=", 1)[1].strip() if "=" in bl else "" start = j + 1 break if bl.strip(): break stmt_text = "\n".join(block_lines[start:]) stmts = tokenize_native_statements(stmt_text) networks.append(Network(title, stmts)) return ParsedBlock(btype, name, num="", sections=sections, networks=networks, source_format="native") def _clean_native_member_lines(body): """Entfernt Leerzeilen, normalisiert Einrueckung minimal -- ist bereits SCL-kompatibel.""" out = [] for raw in body: if not raw.strip(): continue out.append(raw.strip()) return out LOGIC_BRACKET_HEADS = {"U", "O", "UN", "ON", "X", "XN", "A", "AN"} def tokenize_native_statements(text): """Zerlegt den Netzwerk-Text in Statements, getrennt durch ';' auf Klammer-/Quote-Tiefe 0. Kommentare (//...) werden pro Zeile herausgeloest und gesammelt. Wichtig: "(" als Abschluss eines booleschen Klammer-Oeffners (U(/O(/...) ist selbst ein vollstaendiges Statement (endet mit dem naechsten ';') und darf NICHT wie eine echte Klammerung (CALL-Parameterliste ueber mehrere Zeilen) die Tiefenzaehlung erhoehen.""" statements = [] buf = [] comments = [] depth = 0 in_dq = False in_sq = False word = [] for raw_line in text.splitlines(): line = raw_line # Kommentar dieser Zeile abtrennen (ausserhalb von Quotes) code_part = [] j = 0 dq, sq = False, False while j < len(line): ch = line[j] if ch == '"' and not sq: dq = not dq elif ch == "'" and not dq: sq = not sq elif ch == "/" and j + 1 < len(line) and line[j + 1] == "/" and not dq and not sq: comments.append(line[j + 2:].strip()) break code_part.append(ch) j += 1 code_line = "".join(code_part) for ch in code_line: if ch == '"' and not in_sq: in_dq = not in_dq elif ch == "'" and not in_dq: in_sq = not in_sq elif ch == "(" and not in_dq and not in_sq: if "".join(word).upper() not in LOGIC_BRACKET_HEADS: depth += 1 elif ch == "[" and not in_dq and not in_sq: depth += 1 elif ch == ")" and not in_dq and not in_sq: depth = max(0, depth - 1) elif ch == "]" and not in_dq and not in_sq: depth = max(0, depth - 1) if ch.isalnum() or ch == "_": word.append(ch) else: word = [] if ch == ";" and depth == 0 and not in_dq and not in_sq: buf.append(" ") stmt_text = "".join(buf).strip() buf = [] if stmt_text: statements.append(_make_native_statement(stmt_text, comments)) comments = [] continue buf.append(ch) buf.append(" ") tail = "".join(buf).strip() if tail: statements.append(_make_native_statement(tail, comments)) return statements LABEL_RE = re.compile(r"^([A-Za-z_]\w*)\s*:\s*(.*)$", re.DOTALL) def _make_native_statement(stmt_text, comments): comment = " | ".join(c for c in comments if c) text = re.sub(r"\s+", " ", stmt_text).strip() label = None m = LABEL_RE.match(text) if m and " " not in m.group(1) and m.group(2): label, text = m.group(1), m.group(2).strip() if not text: return Statement(label, "NOP", "", comment) parts = text.split(None, 1) mnemonic = parts[0] operand = parts[1].strip() if len(parts) > 1 else "" mnemonic = NATIVE_TO_CANON.get(mnemonic, mnemonic) return Statement(label, mnemonic, operand, comment) # --------------------------------------------------------------------------- # Datei-Ebene: Parsen + Format erkennen # --------------------------------------------------------------------------- def parse_awl_file(path): text = path.read_text(encoding="utf-8-sig", errors="replace") fmt = detect_format(text) if fmt == "rendered": return parse_rendered(text) return parse_native(text) # --------------------------------------------------------------------------- # Kategorisierung (wiederverwendet stlxml2awl.classify/classify_by_name) # --------------------------------------------------------------------------- def analyze_complexity_from_statements(parsed): c = base.Complexity() c.networks = len(parsed.networks) for net in parsed.networks: for st in net.statements: mn = st.mnemonic if mn in base.JUMP_TOKENS or mn in UNSUPPORTED_JUMP_TOKENS or mn in SUPPORTED_JUMP_TOKENS: c.jumps += 1 if mn in base.INDIRECT_CALL_TOKENS: c.indirect_calls += 1 if mn in base.POINTER_ARITH_TOKENS or mn in NATIVE_POINTER_ARITH: c.pointer_arith += 1 if mn == "OPN" and is_indirect_operand(st.operand): c.indirect_db_open += 1 elif is_indirect_operand(st.operand): c.register_indirect += 1 if st.label: c.labels += 1 return c def classify_block(path, parsed): complexity = analyze_complexity_from_statements(parsed) category, reason = base.classify(path, complexity) return category, reason, complexity # --------------------------------------------------------------------------- # Uebersetzung: symbolische STL-Stack-Maschine -> SCL # --------------------------------------------------------------------------- class UnsupportedConstruct(Exception): def __init__(self, reason): super().__init__(reason) self.reason = reason LITERAL_PREFIX_RE = re.compile(r"\b(?:DINT|INT|WORD|DWORD|BYTE|BOOL)#") IDENT_CHAIN_RE = re.compile(r"^[A-Za-z_]\w*(?:\.[A-Za-z_]\w*|\[[^\[\]]*\])*$") def sclize_operand(op): op = op.strip() if not op: return op op = LITERAL_PREFIX_RE.sub("", op) if op.startswith("#") or op.startswith('"'): return op if re.match(r"^[+-]?\d+$", op): return op if re.match(r"^[+-]?\d+\.\d+$", op): return op if re.match(r"^(T|S5T|DT|D|TOD)#", op, re.IGNORECASE): return op if re.match(r"^16#[0-9A-Fa-f]+$", op): return op if op in ("TRUE", "FALSE"): return op # lokale Referenz ohne "#"-Praefix (gerendertes Format haengt es nie an, # auch nicht bei Struct-Pfaden wie "Stat.B_AST") -> SCL verlangt es. if IDENT_CHAIN_RE.match(op): return "#" + op return op class BracketGroup: """Zwei-Ebenen-Faltung je Klammerebene: - chain_expr/chain_op: linksfaltende Kette echter Terme (A/O/X/AN/ON/XN MIT Operand, Vergleiche, geschlossene Unterklammern), exakt wie die STL-RLO sequentiell verknuepft. - outer_expr/outer_op: durch operandenlose "bare" A/O/X-Marken (STL-Idiom fuer UND-vor-ODER ohne Klammern) abgeschlossene Ketten, die als Einheit weiterverknuepft werden. """ def __init__(self): self.chain_expr = None self.chain_op = None self.outer_expr = None self.outer_op = None def add_term(self, term_expr, term_op): if self.chain_expr is None: self.chain_expr = term_expr else: self.chain_expr = f"({self.chain_expr} {term_op} {term_expr})" def break_chain(self, marker_op): seg = self.chain_expr if self.chain_expr is not None else "TRUE" if self.outer_expr is None: self.outer_expr = seg else: self.outer_expr = f"({self.outer_expr} {self.outer_op} {seg})" self.outer_op = marker_op self.chain_expr = None def result(self): if self.outer_expr is None: return self.chain_expr if self.chain_expr is not None else "TRUE" if self.chain_expr is None: return self.outer_expr return f"({self.outer_expr} {self.outer_op} {self.chain_expr})" OP_WORD = {"A": "AND", "O": "OR", "X": "XOR"} class Translator: JUMP_HEAVY_THRESHOLD = 10 def __init__(self, parsed, block_path): self.parsed = parsed self.path = block_path self.temp_vars = [] # zusaetzliche VAR_TEMP-Zeilen self._temp_counter = 0 def new_temp(self, hint, typ="Bool"): self._temp_counter += 1 name = f"tGen{self._temp_counter}_{hint}" name = re.sub(r"[^A-Za-z0-9_]", "", name) self.temp_vars.append(f"{name} : {typ};") return f"#{name}" def translate(self): out_networks = [] for net in self.parsed.networks: lines = self.translate_network(net) out_networks.append((net.title, lines)) return out_networks def translate_network(self, net): lines = [] 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]) # "Erstabfrage"-Flag (wie in echter STL-Hardware): S/R/=/Vergleiche/Kanten # setzen es, damit der naechste Bit-Logik-Term NEU beginnt statt zu # verknuepfen -- lesende Instruktionen (S/R/=/JC/JCN/BEC) selbst # veraendern die aktuell akkumulierte RLO NICHT (Hardware-Register # bleibt bis zur naechsten Bit-Logik-Instruktion unveraendert, daher # liest z.B. ein direkt folgendes "JC" nach "S" dieselbe Bedingung). fresh = [True] def emit(s): lines.append(s) def maybe_reset_top(): if len(groups) == 1 and fresh[0]: groups[0] = BracketGroup() fresh[0] = False def read_top(): return groups[0].result() def negated(expr, negate): return f"NOT ({expr})" if negate else expr for st in net.statements: if st.label: emit(f"{st.label}: ;") fresh[0] = True mn = st.mnemonic op = st.operand if is_indirect_operand(op): raise UnsupportedConstruct(f"registerindirekter/zeigerbasierter Operand: {mn} {op}") if mn == "NOP" or mn == "COMMENT": if st.comment: emit(f"// {st.comment}") continue # --- RLO-Verknuepfung --- if mn in ("A", "O", "X", "AN", "ON", "XN"): negate = mn.endswith("N") base_op = mn[:-1] if negate else mn term_op = OP_WORD[base_op] if not op: # "bare" A/O/X ohne Operand: STL-Idiom "UND-vor-ODER ohne Klammern" # -- schliesst die bisherige Kette ab, verknuepft sie als Einheit. maybe_reset_top() groups[-1].break_chain(term_op) continue maybe_reset_top() term = negated(sclize_operand(op), negate) groups[-1].add_term(term, term_op) continue if mn in ("A(", "O(", "X(", "AN(", "ON(", "XN("): base_mn = mn[:-1] negate = base_mn.endswith("N") base_op = base_mn[:-1] if negate else base_mn pending_bracket_op.append((OP_WORD[base_op], negate)) groups.append(BracketGroup()) continue if mn == ")": inner = groups.pop() term_op, negate = pending_bracket_op.pop() maybe_reset_top() term = negated(f"({inner.result()})", negate) groups[-1].add_term(term, term_op) continue # --- Bit-Operationen --- if mn == "SET": groups[0] = BracketGroup() groups[0].chain_expr = "TRUE" fresh[0] = False continue if mn == "CLR": groups[0] = BracketGroup() groups[0].chain_expr = "FALSE" fresh[0] = False continue if mn == "=": expr = read_top() fresh[0] = True emit(f"{sclize_operand(op)} := {expr};" + (f" // {st.comment}" if st.comment else "")) continue if mn == "S": expr = read_top() fresh[0] = True if expr == "TRUE": emit(f"{sclize_operand(op)} := TRUE;") else: emit(f"IF {expr} THEN {sclize_operand(op)} := TRUE; END_IF;") continue if mn == "R": expr = read_top() fresh[0] = True if expr == "TRUE": emit(f"{sclize_operand(op)} := FALSE;") else: emit(f"IF {expr} THEN {sclize_operand(op)} := FALSE; END_IF;") continue # --- Akku: Laden/Transferieren --- if mn == "L": akku.append(sclize_operand(op)) continue if mn == "T": if not akku: raise UnsupportedConstruct("T ohne vorheriges L (Akku leer)") emit(f"{sclize_operand(op)} := {akku[-1]};") continue if mn == "TAK": if len(akku) < 2: raise UnsupportedConstruct("TAK ohne zwei Akku-Werte") akku[-1], akku[-2] = akku[-2], akku[-1] continue # --- Arithmetik --- if mn in ARITH_TOKENS: if mn == "+" and op: # "+ " -- ADD auf ACCU1 mit Literal if not akku: raise UnsupportedConstruct("+ Konstante ohne Akku-Wert") akku[-1] = f"({akku[-1]} + {sclize_operand(op)})" continue if len(akku) < 2: raise UnsupportedConstruct(f"{mn} ohne zwei Akku-Werte") b = akku.pop() a = akku.pop() akku.append(f"({a} {ARITH_TOKENS[mn]} {b})") continue if mn in BITWISE_WORD_TOKENS: if len(akku) < 2: raise UnsupportedConstruct(f"{mn} ohne zwei Akku-Werte") b = akku.pop() a = akku.pop() fn = BITWISE_WORD_TOKENS[mn] akku.append(f"({a} {fn} {b})") continue if mn in SHIFT_TOKENS: if not akku: raise UnsupportedConstruct(f"{mn} ohne Akku-Wert") if not re.match(r"^\d+$", op.strip()): raise UnsupportedConstruct(f"{mn} mit dynamischer Schiebeweite nicht unterstuetzt") fn = SHIFT_TOKENS[mn] akku[-1] = f"{fn}(IN := {akku[-1]}, N := {op.strip()})" continue if mn in CONVERT_TOKENS: if not akku: raise UnsupportedConstruct(f"{mn} ohne Akku-Wert") akku[-1] = f"{CONVERT_TOKENS[mn]}({akku[-1]})" continue if mn == "NEG": if not akku: raise UnsupportedConstruct("NEG ohne Akku-Wert") akku[-1] = f"-({akku[-1]})" continue # --- RLO invertieren --- if mn == "NOT" and not op: inv = f"NOT ({read_top()})" groups[0] = BracketGroup() groups[0].chain_expr = inv fresh[0] = False continue # --- Vergleiche --- # Ein Vergleich liefert die RLO IMMER frisch aus ACCU1/ACCU2 (ueberschreibt, # kombiniert nicht implizit) -- entspricht der Hardware-Semantik; im # STL-Quelltext steht ein Vergleich daher stets als alleiniger/erster Term # eines Netzwerk- oder Klammerabschnitts. if mn in COMPARE_TOKENS: if len(akku) < 2: raise UnsupportedConstruct(f"{mn} ohne zwei Akku-Werte") a, b = akku[-2], akku[-1] term = f"({a} {COMPARE_TOKENS[mn]} {b})" groups[-1] = BracketGroup() groups[-1].chain_expr = term # Ein Vergleich verhaelt sich wie ein normaler Kontakt: eine # direkt folgende A/O/AN/ON-Instruktion verknuepft sich damit # (z.B. "L a; L b; >=I; U bit; R x" == "(a>=b) AND bit"). fresh[0] = False continue # --- Flankenauswertung --- if mn in ("Rise", "Fall"): cond = read_top() mem = sclize_operand(op) tmp = self.new_temp("Flanke") if mn == "Rise": emit(f"{tmp} := {cond} AND NOT {mem};") else: emit(f"{tmp} := NOT ({cond}) AND {mem};") emit(f"{mem} := {cond};") groups[0] = BracketGroup() groups[0].chain_expr = tmp fresh[0] = False continue # --- Statuswort/BR --- if mn == "SAVE": expr = read_top() emit(f"ENO := {expr};") continue # --- Spruenge --- if mn in SUPPORTED_JUMP_TOKENS: target = op.strip().rstrip(":") if mn == "JU": emit(f"GOTO {target};") elif mn == "JC": expr = read_top() emit(f"IF {expr} THEN GOTO {target}; END_IF;") else: # JCN expr = read_top() emit(f"IF NOT ({expr}) THEN GOTO {target}; END_IF;") continue if mn in UNSUPPORTED_JUMP_TOKENS: raise UnsupportedConstruct(f"Sprungmnemonik {mn} (statuswortabhaengig) nicht unterstuetzt") # --- Bausteinende --- if mn in ("BE", "BEU"): emit("RETURN;") continue if mn == "BEC": expr = read_top() fresh[0] = True emit(f"IF {expr} THEN RETURN; END_IF;") continue # --- CALL --- if mn == "CALL": emit(translate_call(op) + (f" // {st.comment}" if st.comment else "")) continue # --- OPN/AUF --- if mn == "OPN": raise UnsupportedConstruct(f"OPN/AUF DB[...] (indirektes DB-Oeffnen): {op}") if mn in TIMER_TOKENS: raise UnsupportedConstruct(f"Timer-Mnemonik {mn} nicht unterstuetzt") if mn in COUNTER_TOKENS: raise UnsupportedConstruct(f"Zaehler-Mnemonik {mn} nicht unterstuetzt") if mn in ("UC", "CC"): raise UnsupportedConstruct(f"indirekter Aufruf ({mn}) -- CASE-Verteiler noetig") if mn in ("LAR1", "LAR2", "TAR1", "TAR2", "+AR1", "+AR2", "ADDAR1", "ADDAR2"): raise UnsupportedConstruct(f"Pointer-/AR-Arithmetik ({mn}) nicht unterstuetzt") raise UnsupportedConstruct(f"nicht unterstuetzte Mnemonik: {mn} {op}".strip()) return lines def _split_top_level(text, sep=","): """Teilt text am Trennzeichen, aber nicht innerhalb von (), [], "" oder ''.""" parts = [] buf = [] depth = 0 dq = sq = 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 == sep and depth == 0 and not dq and not sq: parts.append("".join(buf)) buf = [] continue buf.append(ch) parts.append("".join(buf)) return parts CALL_RE = re.compile( r'^(?:(?:FB|FC|OB|SFB|SFC)\s+)?(#\w+|"[^"]+")\s*(?:,\s*(#\w+|"[^"]+"))?\s*(\(.*\))?\s*$', re.DOTALL) def translate_call(op): """Wandelt den AWL-CALL-Operanden-Text in SCL-Aufrufsyntax um. Das gerenderte Format (stlxml2awl.py) stellt bei Bausteinaufrufen den Blocktyp voran, z.B. 'FB "Name" , "Instanz" (Param:=Wert)'.""" op = op.strip() m = CALL_RE.match(op) if not m: raise UnsupportedConstruct(f"CALL-Syntax nicht erkannt: {op}") first, second, params_raw = m.group(1), m.group(2), m.group(3) if not second and op.rstrip().endswith(","): raise UnsupportedConstruct(f"CALL-Instanz fehlt: {op}") target = second or first if not params_raw: return f"{target}();" inner = params_raw.strip() assert inner.startswith("(") and inner.endswith(")") inner = inner[1:-1].strip() if not inner: return f"{target}();" parts = [] for piece in _split_top_level(inner, ","): piece = piece.strip() if not piece: continue if ":=" not in piece: raise UnsupportedConstruct(f"CALL-Parameter nicht erkannt: {piece}") name, value = piece.split(":=", 1) parts.append(f"{name.strip()} := {sclize_operand(value.strip())}") return f"{target}(" + ", ".join(parts) + ");" def count_jumps(parsed): n = 0 for net in parsed.networks: for st in net.statements: if st.mnemonic in SUPPORTED_JUMP_TOKENS or st.mnemonic in UNSUPPORTED_JUMP_TOKENS: n += 1 return n # --------------------------------------------------------------------------- # SCL-Ausgabe # --------------------------------------------------------------------------- SECTION_ORDER = [("Input", "VAR_INPUT"), ("Output", "VAR_OUTPUT"), ("InOut", "VAR_IN_OUT"), ("Static", "VAR"), ("Temp", "VAR_TEMP")] BTYPE_KEYWORD = {"FB": "FUNCTION_BLOCK", "FC": "FUNCTION", "OB": "ORGANIZATION_BLOCK"} BTYPE_END = {"FB": "END_FUNCTION_BLOCK", "FC": "END_FUNCTION", "OB": "END_ORGANIZATION_BLOCK"} def render_scl(parsed, translated_networks, temp_vars): out = [] kw = BTYPE_KEYWORD.get(parsed.btype, "FUNCTION_BLOCK") ret_type = " : Void" if parsed.btype == "FC" else "" out.append(f'{kw} "{parsed.name}"{ret_type}') out.append("{ S7_Optimized_Access := 'FALSE' }") out.append("// Mechanische AWL->SCL-Konvertierung (awl2scl.py). Kontrollfluss per GOTO,") out.append("// wo im Original mit Sprungmarken gearbeitet wurde. Vor Uebernahme in TIA") out.append("// Portal pruefen/kompilieren (siehe SCL-Export/README.md).") out.append("") for sect_key, scl_kw in SECTION_ORDER: member_lines = list(parsed.sections.get(sect_key, [])) if sect_key == "Temp": member_lines += temp_vars if not member_lines: continue out.append(f" {scl_kw}") for ml in member_lines: out.append(f" {ml}") out.append(" END_VAR") out.append("") out.append("BEGIN") for title, lines in translated_networks: label = title if title else "" out.append(f" REGION {label}".rstrip()) for ln in lines: out.append(f" {ln}") out.append(" END_REGION") out.append("") out.append(BTYPE_END.get(parsed.btype, "END_FUNCTION_BLOCK")) out.append("") return "\n".join(out) # --------------------------------------------------------------------------- # Datei-/Verzeichnis-Handling # --------------------------------------------------------------------------- def find_awl_files(directory): return sorted(Path(directory).rglob("*.awl")) def station_of(path): for part in Path(path).parts: if part.startswith("=A"): return part return "_" def default_target_dir(): env = os.environ.get("PV_SCL_EXPORT") if env: return Path(env) return Path(__file__).resolve().parent.parent / "SCL-Export" def default_log_path(): project_root = Path(__file__).resolve().parent.parent log_dir = project_root / "log" log_dir.mkdir(parents=True, exist_ok=True) ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") return log_dir / f"awl2scl_{ts}.log" # --------------------------------------------------------------------------- # Ergebnis-Datensatz + Log # --------------------------------------------------------------------------- class Outcome: TRANSLATED = "translated" SKIPPED_EXISTS = "skipped_exists" 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): self.path = path self.outcome = outcome self.category = category self.reason = reason self.scl_path = scl_path self.jumps = jumps def process_file(awl_path, requested_categories, target_dir, console): try: parsed = parse_awl_file(awl_path) except NotABlockError as exc: console(f"UEBERSPRUNGEN [kein Baustein] {awl_path}: {exc}") return Result(awl_path, Outcome.SKIPPED_CATEGORY, reason=str(exc)) except Exception as exc: console(f"FEHLER (Parser) {awl_path}: {exc}") return Result(awl_path, Outcome.ERROR, reason=f"Parserfehler: {exc}") try: category, reason, complexity = classify_block(awl_path, parsed) except Exception as exc: console(f"FEHLER (Klassifikation) {awl_path}: {exc}") return Result(awl_path, Outcome.ERROR, reason=f"Klassifikationsfehler: {exc}") if category not in requested_categories: msg = f"{parsed.name}: Kategorie {category} (nicht angefordert) -- uebersprungen" console(msg) return Result(awl_path, Outcome.SKIPPED_CATEGORY, category=category, reason=reason) 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) translator = Translator(parsed, awl_path) try: translated = translator.translate() except UnsupportedConstruct as exc: console(f"MANUELLE PRUEFUNG {awl_path}: {exc.reason}") return Result(awl_path, Outcome.MANUAL_REVIEW, category=category, reason=exc.reason) except Exception as exc: console(f"FEHLER (Uebersetzung) {awl_path}: {exc}") return Result(awl_path, Outcome.ERROR, category=category, reason=f"Uebersetzungsfehler: {exc}") 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) def write_log(log_path, results, requested_categories, target_dir): lines = [] ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") lines.append(f"AWL -> SCL Konvertierung ({ts})") lines.append("=" * 70) lines.append("") lines.append(f"Angeforderte Kategorie(n) : {','.join(sorted(requested_categories))}") lines.append(f"Zielordner : {target_dir}") 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"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("") manual = [r for r in results if r.outcome == Outcome.MANUAL_REVIEW] if manual: lines.append("-" * 70) lines.append(f"Manuelle Pruefung noetig ({len(manual)})") lines.append("-" * 70) for r in sorted(manual, key=lambda r: str(r.path)): lines.append(f"[{r.category}] {r.path}") lines.append(f" Grund: {r.reason}") lines.append("") errors = [r for r in results if r.outcome == Outcome.ERROR] if errors: lines.append("-" * 70) lines.append(f"Fehler ({len(errors)})") lines.append("-" * 70) for r in sorted(errors, key=lambda r: str(r.path)): lines.append(f"{r.path}: {r.reason}") lines.append("") goto_heavy = [r for r in results if r.outcome == Outcome.TRANSLATED and r.jumps > Translator.JUMP_HEAVY_THRESHOLD] if goto_heavy: lines.append("-" * 70) lines.append(f"GOTO-lastig, Nacharbeit empfohlen ({len(goto_heavy)})") lines.append("-" * 70) for r in sorted(goto_heavy, key=lambda r: -r.jumps): lines.append(f"{r.jumps:4d} Spruenge {r.path} -> {r.scl_path}") lines.append("") skipped_cat = [r for r in results if r.outcome == Outcome.SKIPPED_CATEGORY and r.category] if skipped_cat: lines.append("-" * 70) lines.append(f"Uebersprungen, Kategorie nicht angefordert ({len(skipped_cat)})") lines.append("-" * 70) for r in sorted(skipped_cat, key=lambda r: str(r.path)): lines.append(f"[{r.category}] {r.path} ({r.reason})") lines.append("") skipped_exists = [r for r in results if r.outcome == Outcome.SKIPPED_EXISTS] if skipped_exists: lines.append("-" * 70) lines.append(f"Uebersprungen, Ziel-SCL existiert bereits ({len(skipped_exists)})") lines.append("-" * 70) for r in sorted(skipped_exists, key=lambda r: str(r.path)): lines.append(f"{r.path} -> {r.scl_path}") lines.append("") log_path.parent.mkdir(parents=True, exist_ok=True) log_path.write_text("\n".join(lines) + "\n", encoding="utf-8") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- 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") parser.add_argument("--dir", "--directory", dest="directory", type=Path, help="Verzeichnis rekursiv nach .awl-Dateien durchsuchen") parser.add_argument("--category", default="A", 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("--log", type=Path, help="Pfad der Log-Datei") args = parser.parse_args(argv) targets = list(args.files) if args.directory: targets.extend(find_awl_files(args.directory)) if not targets: parser.error("keine .awl-Dateien angegeben (Pfade und/oder --dir)") requested_categories = {c.strip().upper() for c in args.category.split(",") if c.strip()} target_dir = args.target_dir or default_target_dir() def console(msg): print(msg) results = [] for awl_path in targets: results.append(process_file(awl_path, requested_categories, target_dir, console)) log_path = args.log or default_log_path() write_log(log_path, results, requested_categories, target_dir) 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.") print(f"Log: {log_path}") return 1 if n_err else 0 if __name__ == "__main__": sys.exit(main())