524 lines
19 KiB
Python
524 lines
19 KiB
Python
#!/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", "GE_I": ">=I", "LE_I": "<=I",
|
|
"NE_D": "<>D", "EQ_D": "==D", "GT_D": ">D", "LT_D": "<D", "GE_D": ">=D", "LE_D": "<=D",
|
|
"NE_R": "<>R", "EQ_R": "==R", "GT_R": ">R", "LT_R": "<R", "GE_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: <Projekt>/log/stlxml2awl_<Zeitstempel>.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())
|