#!/usr/bin/env python3 """Wandelt TIA-Openness SimaticML XML (StatementList) in lesbaren AWL-Text um.""" import argparse import datetime import re import sys from pathlib import Path import xml.etree.ElementTree as ET NS_IF = "{http://www.siemens.com/automation/Openness/SW/Interface/v5}" NS_SL = "{http://www.siemens.com/automation/Openness/SW/NetworkSource/StatementList/v5}" TOKEN_MAP = { "A_BRACK": "A(", "O_BRACK": "O(", "AN_BRACK": "AN(", "ON_BRACK": "ON(", "X_BRACK": "X(", "XN_BRACK": "XN(", "BRACKET": ")", "NE_I": "<>I", "EQ_I": "==I", "GT_I": ">I", "LT_I": "=I", "LE_I": "<=I", "NE_D": "<>D", "EQ_D": "==D", "GT_D": ">D", "LT_D": "=D", "LE_D": "<=D", "NE_R": "<>R", "EQ_R": "==R", "GT_R": ">R", "LT_R": "=R", "LE_R": "<=R", "ADD_I": "+I", "SUB_I": "-I", "MUL_I": "*I", "DIV_I": "/I", "ADD_D": "+D", "SUB_D": "-D", "MUL_D": "*D", "DIV_D": "/D", "ADD_R": "+R", "SUB_R": "-R", "MUL_R": "*R", "DIV_R": "/R", "ADD": "+", "NOP_0": "NOP 0", "NOP_1": "NOP 1", "TAR2_ACCU1": "TAR2", "TAR1_ACCU1": "TAR1", "LAR1_ACCU1": "LAR1", "LAR2_ACCU1": "LAR2", "OPEN_DB": "OPN", "OPEN_DI": "OPN", "EMPTY_LINE": "", "COMMENT": "//", "Assign": "=", } def fmt_symbol(access): parts = [] for comp in access.findall(f"./{NS_SL}Symbol/{NS_SL}Component"): name = comp.get("Name") token = name if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) else f'"{name}"' idx = [fmt_access(a) for a in comp.findall(f"{NS_SL}Access")] if idx: token += "[" + ",".join(idx) + "]" if comp.get("SliceAccessModifier"): token += "." + comp.get("SliceAccessModifier") parts.append(token) sym = ".".join(parts) if access.get("Scope") == "GlobalVariable" and parts and not parts[0].startswith('"'): sym = f'"{parts[0][1:-1] if parts[0].startswith(chr(34)) else parts[0]}"' + ( "." + ".".join(parts[1:]) if len(parts) > 1 else "") return sym def fmt_address(addr): area = addr.get("Area") typ = addr.get("Type") bit = int(addr.get("BitOffset", 0)) blk = addr.get("BlockNumber") width = {"Bool": 1, "Byte": 8, "Word": 16, "DWord": 32, "Int": 16, "DInt": 32}.get(typ, 0) sizech = {1: "", 8: "B", 16: "W", 32: "D"}.get(width, "?") areach = {"Local": "L", "Memory": "M", "Input": "I", "Output": "Q", "DB": "DB", "DI": "DI", "Timer": "T", "Counter": "C", "PeripheryInput": "PI", "PeripheryOutput": "PQ"}.get(area, area) if area in ("DB", "DI"): pre = f"DB{blk}." if blk else "" core = f"{pre}{'DBX' if width == 1 else 'DB' + sizech}{bit // 8}" elif area == "Timer": return f"T {bit}" elif area == "Counter": return f"C {bit}" else: core = f"{areach}{sizech}{bit // 8}" if width == 1: core += f".{bit % 8}" return core def fmt_indirect(ind): width = ind.get("Width") area = ind.get("Area") reg = ind.get("Register", "") inner = [fmt_access(a) for a in ind.findall(f"{NS_SL}Access")] const = ind.find(f"{NS_SL}Constant") off = "" if const is not None: off = const.findtext(f"{NS_SL}ConstantValue", "") if width == "Block": # OPN DB[var] return f"DB[{inner[0]}]" if inner else f"DB[{off}]" sizech = {"Bit": "X", "Byte": "B", "Word": "W", "Double": "D", "DWord": "D"}.get(width, "?") areach = {"DB": "DB", "DI": "DI", "Local": "L", "Memory": "M", "Input": "I", "Output": "Q"}.get(area, area or "") base = f"{areach}{sizech}" if areach not in ("DB", "DI") else f"{areach}{sizech}" if reg: # registerindirekt return f"{base} [{reg},{off}]" if inner: # speicherindirekt return f"{base} [{inner[0]}]" return f"{base} [{off}]" def fmt_access(access): scope = access.get("Scope") if scope in ("LocalVariable", "GlobalVariable", "LocalConstant", "GlobalConstant"): return fmt_symbol(access) if scope in ("LiteralConstant", "TypedConstant"): c = access.find(f"{NS_SL}Constant") val = c.findtext(f"{NS_SL}ConstantValue", "") if c is not None else "" typ = c.findtext(f"{NS_SL}ConstantType", "") if c is not None else "" pref = {"Word": "W#16#", "DWord": "DW#16#", "Time": "", "S5Time": "", "Byte": "B#16#"} if typ in ("Word", "DWord", "Byte") and not val.startswith(("2#", "16#", "W#", "DW#", "B#")): return pref[typ] + val return val if scope == "AddressConstant": c = access.find(f"{NS_SL}Constant") return "P#" + c.findtext(f"{NS_SL}ConstantValue", "") if c is not None else "P#?" if scope == "Address": addr = access.find(f"{NS_SL}Address") if addr is not None: return fmt_address(addr) ind = access.find(f"{NS_SL}Indirect") if ind is not None: return fmt_indirect(ind) return "?ADDR?" if scope == "Label": lab = access.find(f"{NS_SL}Label") return lab.get("Name") if lab is not None else "?LBL?" if scope == "LabelDefinition": lab = access.find(f"{NS_SL}Label") return (lab.get("Name") + ":") if lab is not None else "?LBL?:" if scope == "Statusword": return access.find(f"{NS_SL}Statusword").get("Name", "?SW?") if scope == "Call": ci = access.find(f"{NS_SL}CallInfo") if ci is None: ins = access.find(f"{NS_SL}Instruction") name = ins.get("Name") params = [] for p in ins.findall(f"{NS_SL}Parameter"): pa = p.find(f"{NS_SL}Access") params.append(f"{p.get('Name')}:={fmt_access(pa) if pa is not None else '?'}") return f"CALLINFO SYS \"{name}\" (" + ", ".join(params) + ")" name, btype = ci.get("Name"), ci.get("BlockType") inst = ci.find(f"{NS_SL}Instance") s = f'"{name}"' if inst is not None: s += " , " + fmt_symbol(inst) params = [] for p in ci.findall(f"{NS_SL}Parameter"): pa = p.find(f"{NS_SL}Access") params.append(f"{p.get('Name')}:={fmt_access(pa) if pa is not None else '?'}") if params: s += " (" + ", ".join(params) + ")" return f"CALLINFO {btype} {s}" return f"?{scope}?" def render_statement(st): label = "" ld = st.find(f"{NS_SL}LabelDeclaration/{NS_SL}Label") if ld is not None: label = ld.get("Name") + ": " toks = st.findall(f"{NS_SL}StlToken") accesses = st.findall(f"{NS_SL}Access") comments = [] for lc in st.iter(f"{NS_SL}LineComment"): t = lc.findtext(f"{NS_SL}Text", "") if t: comments.append(t) for cm in st.findall(f"{NS_SL}Comment"): t = cm.findtext(f"{NS_SL}Text", "") if t: comments.append(t) if not toks: if label: return label.rstrip() return "// " + " | ".join(comments) if comments else "" tok = toks[0].get("Text") mn = TOKEN_MAP.get(tok, tok) if tok == "EMPTY_LINE": return label.rstrip() if label else "" if tok == "COMMENT": return "// " + " | ".join(comments) ops = [] for a in accesses: f = fmt_access(a) if f.startswith("CALLINFO"): return f"{label}CALL {f[9:]}" + (" // " + " | ".join(comments) if comments else "") ops.append(f) line = f"{label:6s}{mn:6s} " + ", ".join(ops) if comments: line = f"{line:60s} // " + " | ".join(comments) return line.rstrip() def render_interface(root): out = [] for sec in root.iter(f"{NS_IF}Section"): members = sec.findall(f"{NS_IF}Member") out.append(f" {sec.get('Name')}:") for m in members: out.extend(render_member(m, 4)) return out def render_member(m, indent): out = [] cm = m.find(f"{NS_IF}Comment/{NS_IF}MultiLanguageText") sv = m.findtext(f"{NS_IF}StartValue") line = " " * indent + f"{m.get('Name')} : {m.get('Datatype')}" if sv: line += f" := {sv}" if cm is not None and cm.text: line = f"{line:55s} // {cm.text}" out.append(line) for sub in m.findall(f"{NS_IF}Member"): out.extend(render_member(sub, indent + 2)) return out def parse_block(path): text = path.read_text(encoding="utf-8") root = ET.fromstring(text) for tag in ("FB", "FC", "OB", "GlobalDB", "InstanceDB"): blk = root.find(f".//SW.Blocks.{tag}") if blk is not None: return blk, tag return None, None def render_block(blk, btype): out = [] attrs = blk.find("AttributeList") name = attrs.findtext("Name") num = attrs.findtext("Number") lang = attrs.findtext("ProgrammingLanguage") out.append(f"=== {btype} {num}: {name} [{lang}] ===") iface = attrs.find("Interface") if iface is not None: secs = iface.find(f"{NS_IF}Sections") if secs is not None: out.append("INTERFACE:") out.extend(render_interface(secs)) nwno = 0 for cu in blk.iter("SW.Blocks.CompileUnit"): nwno += 1 title = "" for mt in cu.iter("MultilingualTextItem"): al = mt.find("AttributeList") if al is not None and al.findtext("Culture") == "de-DE" and al.findtext("Text"): title = al.findtext("Text") break out.append(f"\n--- NW {nwno}: {title} ---") src = cu.find(f".//{NS_SL}StatementList") if src is None: out.append(" (kein STL-Quelltext)") continue for st in src.findall(f"{NS_SL}StlStatement"): out.append(render_statement(st)) return out, name, num # --------------------------------------------------------------------------- # Komplexitaets-Analyse / Kategorisierung # # Kategorie A: einfache Verknuepfungslogik, nahezu mechanische 1:1-Uebersetzung # Kategorie B: Ablaufsteuerung mit vielen Sprüngen, Kontrollfluss muss zu # IF/CASE rekonstruiert werden # Kategorie C: Pointer-/Indirektzugriffe (registerindirekte Adressierung), # hoechster Aufwand, Registeremulation nicht sinnvoll # --------------------------------------------------------------------------- JUMP_TOKENS = { "JU", "JC", "JCN", "JL", "JP", "JBI", "JNBI", "JZ", "JN", "JM", "JMZ", "JPZ", "JO", "JOS", "JUO", "LOOP", } INDIRECT_CALL_TOKENS = {"UC", "CC"} POINTER_ARITH_TOKENS = { "LAR1_ACCU1", "LAR2_ACCU1", "TAR1_ACCU1", "TAR2_ACCU1", "ADDAR1", "ADDAR2", } def _pat(p): return re.compile(p, re.IGNORECASE) # Namensmuster aus der Projekt-Klassifizierung (bekannte Bausteingruppen) C_NAME_PATTERNS = [ _pat(r"^Daten\w*$"), _pat(r"^Fifo"), _pat(r"^DBMOV$"), _pat(r"(^|[\\/])Logistik([\\/]|$)"), _pat(r"(^|[\\/])SS_Host([\\/]|$)"), _pat(r"(^|[\\/])Visualisierung([\\/]|$)"), ] B_NAME_PATTERNS = [ _pat(r"^MHM"), _pat(r"_Bereich$"), _pat(r"_OmniFahren$"), _pat(r"_Ziel_FC$"), ] A_NAME_PATTERNS = [ _pat(r"^MZylinder$"), _pat(r"^MSensoren$"), _pat(r"^MBlinken$"), _pat(r"^MAntrieb_"), _pat(r"(^|[\\/])Basics[_\\/]"), _pat(r"^Set_(Input|Output)$"), ] def classify_by_name(xml_path): stem = xml_path.stem full = str(xml_path) for patterns, cat in ((C_NAME_PATTERNS, "C"), (B_NAME_PATTERNS, "B"), (A_NAME_PATTERNS, "A")): for pat in patterns: if pat.search(stem) or pat.search(full): return cat return None class Complexity: def __init__(self): self.jumps = 0 self.labels = 0 self.indirect_calls = 0 self.register_indirect = 0 self.pointer_arith = 0 self.networks = 0 @property def has_pointer_signal(self): return self.register_indirect > 0 or self.pointer_arith > 0 @property def has_branch_signal(self): return self.indirect_calls > 0 def analyze_complexity(blk): c = Complexity() c.networks = sum(1 for _ in blk.iter("SW.Blocks.CompileUnit")) for st in blk.iter(f"{NS_SL}StlStatement"): toks = st.findall(f"{NS_SL}StlToken") if toks: text = toks[0].get("Text") if text in JUMP_TOKENS: c.jumps += 1 if text in INDIRECT_CALL_TOKENS: c.indirect_calls += 1 if text in POINTER_ARITH_TOKENS: c.pointer_arith += 1 if st.find(f"{NS_SL}LabelDeclaration") is not None: c.labels += 1 for ind in blk.iter(f"{NS_SL}Indirect"): if ind.get("Register"): c.register_indirect += 1 return c 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)" name_cat = classify_by_name(xml_path) if name_cat == "C": return "C", "Namensmuster (Daten*/Fifo*/DBMOV/Logistik/SS_Host/Visualisierung)" if complexity.has_branch_signal: return "B", f"{complexity.indirect_calls}x indirekter Aufruf (UC/CC) - CASE-Verteiler noetig" if name_cat == "B": return "B", "Namensmuster (HM-Module/_Bereich/_OmniFahren/_Ziel_FC)" if name_cat == "A": return "A", "Namensmuster (einfache Verknuepfungslogik)" return "A", "kein Namensmuster erkannt, keine Pointer-/Sprungauffaelligkeiten" class NotABlockError(Exception): """XML enthaelt keinen SW.Blocks-Export (z.B. UDT, Tag-Tabelle).""" class ConversionResult: def __init__(self, xml_path, awl_path, btype, name, num, complexity, category, reason): self.xml_path = xml_path self.awl_path = awl_path self.btype = btype self.name = name self.num = num self.complexity = complexity self.category = category self.reason = reason def convert_file(xml_path): blk, btype = parse_block(xml_path) if blk is None: raise NotABlockError("kein SW.Blocks-Element gefunden (kein Baustein-Export)") lines, name, num = render_block(blk, btype) awl_path = xml_path.with_suffix(".awl") awl_path.write_text("\n".join(lines) + "\n", encoding="utf-8") complexity = analyze_complexity(blk) category, reason = classify(xml_path, complexity) return ConversionResult(xml_path, awl_path, btype, name, num, complexity, category, reason) # --------------------------------------------------------------------------- # CLI / Massenkonvertierung # --------------------------------------------------------------------------- def find_xml_files(directory): return sorted(Path(directory).rglob("*.xml")) 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"stlxml2awl_{ts}.log" 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}") 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}") def write_log(log_path, results, skipped, errors): log_path.parent.mkdir(parents=True, exist_ok=True) lines = [] ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") lines.append(f"STL -> AWL Konvertierung ({ts})") lines.append("=" * 70) lines.append("") lines.append(f"Konvertiert : {len(results)}") lines.append(f"Uebersprungen (kein Baustein) : {len(skipped)}") lines.append(f"Fehler : {len(errors)}") lines.append("") by_cat = {"A": [], "B": [], "C": []} for r in results: by_cat[r.category].append(r) lines.append(f"Kategorie A (einfache Verknuepfungslogik) : {len(by_cat['A'])}") lines.append(f"Kategorie B (Ablaufsteuerung/Sprungmarken) : {len(by_cat['B'])}") lines.append(f"Kategorie C (Pointer-/Indirektzugriffe) : {len(by_cat['C'])}") lines.append("") lines.append("Hinweis: automatische Einschaetzung anhand Codeanalyse (registerindirekte") lines.append("Zugriffe, indirekte Aufrufe) und bekannter Namensmuster. Ersetzt keine") lines.append("fachliche Pruefung, dient als Priorisierungshilfe fuer die Massenkonvertierung.") lines.append("") if by_cat["C"]: lines.append("-" * 70) lines.append(f"Kategorie C - Pointer-/Indirektzugriffe: MANUELLE UEBERARBEITUNG NOETIG ({len(by_cat['C'])})") lines.append("-" * 70) for r in sorted(by_cat["C"], key=lambda r: str(r.xml_path)): lines.append(_format_result_line(r)) lines.append("") if by_cat["B"]: lines.append("-" * 70) lines.append(f"Kategorie B - Ablaufsteuerung/viele Sprungmarken: REVIEW ERFORDERLICH ({len(by_cat['B'])})") lines.append("-" * 70) for r in sorted(by_cat["B"], key=lambda r: str(r.xml_path)): lines.append(_format_result_line(r)) lines.append("") if errors: lines.append("-" * 70) lines.append(f"Fehler bei der Konvertierung ({len(errors)})") lines.append("-" * 70) for path, exc in errors: lines.append(f"{path}: {exc}") lines.append("") if skipped: lines.append("-" * 70) lines.append(f"Uebersprungen ({len(skipped)}, kein Baustein-Export, z.B. UDT/Tag-Tabelle)") lines.append("-" * 70) for path in skipped: lines.append(str(path)) lines.append("") log_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def main(argv=None): parser = argparse.ArgumentParser( description="Wandelt TIA-Openness SimaticML XML (StatementList) in lesbaren AWL-Text um.") parser.add_argument("files", nargs="*", type=Path, help="Einzelne XML-Dateien") parser.add_argument("--dir", "--directory", dest="directory", type=Path, help="Verzeichnis rekursiv nach .xml-Dateien durchsuchen und alle konvertieren") parser.add_argument("--log", type=Path, help="Pfad der Log-Datei (Default: /log/stlxml2awl_.log)") args = parser.parse_args(argv) targets = list(args.files) if args.directory: targets.extend(find_xml_files(args.directory)) if not targets: parser.error("keine XML-Dateien angegeben (Pfade und/oder --dir)") results, skipped, errors = [], [], [] for xml_path in targets: try: result = convert_file(xml_path) results.append(result) print(f"OK [{result.category}] {xml_path} -> {result.awl_path.name}") except NotABlockError: skipped.append(xml_path) except Exception as exc: errors.append((xml_path, exc)) print(f"FEHLER {xml_path}: {exc}", file=sys.stderr) log_path = args.log or default_log_path() write_log(log_path, results, skipped, errors) print() print(f"{len(results)} konvertiert, {len(skipped)} uebersprungen, {len(errors)} Fehler.") print(f"Kategorie A/B/C: " f"{sum(1 for r in results if r.category == 'A')}/" f"{sum(1 for r in results if r.category == 'B')}/" f"{sum(1 for r in results if r.category == 'C')}") print(f"Log: {log_path}") return 1 if errors else 0 if __name__ == "__main__": sys.exit(main())