diff --git a/lib/awl2scl.py b/lib/awl2scl.py new file mode 100644 index 0000000..8c1b576 --- /dev/null +++ b/lib/awl2scl.py @@ -0,0 +1,1142 @@ +#!/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()) diff --git a/lib/stlxml2awl.py b/lib/stlxml2awl.py index 2c9799f..723aa74 100644 --- a/lib/stlxml2awl.py +++ b/lib/stlxml2awl.py @@ -268,6 +268,7 @@ INDIRECT_CALL_TOKENS = {"UC", "CC"} POINTER_ARITH_TOKENS = { "LAR1_ACCU1", "LAR2_ACCU1", "TAR1_ACCU1", "TAR2_ACCU1", "ADDAR1", "ADDAR2", } +OPEN_DB_TOKENS = {"OPEN_DB", "OPEN_DI"} def _pat(p): @@ -316,11 +317,12 @@ class Complexity: self.indirect_calls = 0 self.register_indirect = 0 self.pointer_arith = 0 + self.indirect_db_open = 0 self.networks = 0 @property def has_pointer_signal(self): - return self.register_indirect > 0 or self.pointer_arith > 0 + return self.register_indirect > 0 or self.pointer_arith > 0 or self.indirect_db_open > 0 @property def has_branch_signal(self): @@ -340,6 +342,10 @@ def analyze_complexity(blk): c.indirect_calls += 1 if text in POINTER_ARITH_TOKENS: c.pointer_arith += 1 + if text in OPEN_DB_TOKENS: + for acc in st.findall(f"{NS_SL}Access"): + if acc.find(f"{NS_SL}Indirect") is not None: + c.indirect_db_open += 1 if st.find(f"{NS_SL}LabelDeclaration") is not None: c.labels += 1 for ind in blk.iter(f"{NS_SL}Indirect"): @@ -352,7 +358,8 @@ def classify(xml_path, complexity): if complexity.networks == 0: return "A", "reine Datenablage ohne Programmcode (0 Netzwerke, z.B. GlobalDB)" if complexity.has_pointer_signal: - return "C", f"registerindirekte Adressierung ({complexity.register_indirect}x Indirect, {complexity.pointer_arith}x AR-Arithmetik)" + return "C", (f"registerindirekte Adressierung ({complexity.register_indirect}x Indirect, " + f"{complexity.pointer_arith}x AR-Arithmetik, {complexity.indirect_db_open}x OPN DB[Variable])") name_cat = classify_by_name(xml_path) if name_cat == "C": return "C", "Namensmuster (Daten*/Fifo*/DBMOV/Logistik/SS_Host/Visualisierung)" @@ -412,7 +419,8 @@ def default_log_path(): def _format_result_line(r): c = r.complexity stats = (f"Sprünge={c.jumps} Labels={c.labels} IndirekteAufrufe={c.indirect_calls} " - f"RegisterIndirekt={c.register_indirect} AR-Arithmetik={c.pointer_arith} NW={c.networks}") + f"RegisterIndirekt={c.register_indirect} AR-Arithmetik={c.pointer_arith} " + f"OPN-Indirekt={c.indirect_db_open} NW={c.networks}") return (f"{r.btype:10s} {str(r.num or ''):>5s} {r.name or '':30s} [{stats}]\n" f" Grund: {r.reason}\n" f" Pfad : {r.xml_path}")