1302 lines
50 KiB
Python
1302 lines
50 KiB
Python
#!/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 <datei.awl> ... REM einzelne Dateien
|
|
awl2scl.py --dir <verzeichnis> REM rekursiv alle .awl
|
|
awl2scl.py --category A[,B,C] --target-dir <pfad> --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
|
|
import sclopt # 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 uebersetzbare Baustein-AWL (z.B. leere/fremde Datei)."""
|
|
|
|
|
|
class DataBlockError(NotABlockError):
|
|
"""Reiner Datenbaustein (DB) ohne Programmlogik -- nichts zu uebersetzen."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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": ">=", "<=I": "<=",
|
|
"==D": "=", "<>D": "<>", ">D": ">", "<D": "<", ">=D": ">=", "<=D": "<=",
|
|
"==R": "=", "<>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*$")
|
|
# Kopfzeile mit beliebigem Sprach-/Typ-Tag (z.B. "[DB]") -- fuer die Erkennung
|
|
# reiner Datenbausteine, die kein STL enthalten.
|
|
ANY_HEADER_RE = re.compile(r"^===\s*(\w+)\s+(\S*):\s*(.+?)\s*\[(\w+)\]\s*===\s*$")
|
|
DATA_BLOCK_TYPES = {"GlobalDB", "InstanceDB", "DB", "ArrayDB"}
|
|
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:
|
|
any_m = ANY_HEADER_RE.match(lines[0].strip())
|
|
if any_m and (any_m.group(4) == "DB" or any_m.group(1) in DATA_BLOCK_TYPES):
|
|
raise DataBlockError(
|
|
f"Datenbaustein ({any_m.group(1)} {any_m.group(2)}: {any_m.group(3)}) - keine Logik")
|
|
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:
|
|
mdb = re.match(r'^DATA_BLOCK\s+"([^"]+)"', lines[0].strip())
|
|
if mdb:
|
|
raise DataBlockError(f"Datenbaustein (DATA_BLOCK {mdb.group(1)}) - keine Logik")
|
|
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):
|
|
"""Liefert Liste (Netzwerktitel, IR-Knotenliste)."""
|
|
out_networks = []
|
|
for net in self.parsed.networks:
|
|
nodes = self.translate_network(net)
|
|
out_networks.append((net.title, nodes))
|
|
return out_networks
|
|
|
|
def translate_network(self, net):
|
|
nodes = []
|
|
groups = [BracketGroup()] # Stack der Klammerebenen, [0] = Top-Level
|
|
pending_bracket_op = [] # je offener Klammer: (op, negate) fuer den Term bei ")"
|
|
akku = [] # symbolischer Wert-Stack (ACCU1 = akku[-1])
|
|
# "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 add(node):
|
|
nodes.append(node)
|
|
|
|
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:
|
|
add(sclopt.Label(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:
|
|
add(sclopt.Comment(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
|
|
add(sclopt.Assign(sclize_operand(op), expr, st.comment))
|
|
continue
|
|
if mn == "S":
|
|
expr = read_top()
|
|
fresh[0] = True
|
|
assign = sclopt.Assign(sclize_operand(op), "TRUE")
|
|
if expr == "TRUE":
|
|
add(assign)
|
|
else:
|
|
add(sclopt.IfDo(expr, [assign]))
|
|
continue
|
|
if mn == "R":
|
|
expr = read_top()
|
|
fresh[0] = True
|
|
assign = sclopt.Assign(sclize_operand(op), "FALSE")
|
|
if expr == "TRUE":
|
|
add(assign)
|
|
else:
|
|
add(sclopt.IfDo(expr, [assign]))
|
|
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)")
|
|
add(sclopt.Assign(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:
|
|
# "+ <Konstante>" -- 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":
|
|
add(sclopt.Assign(tmp, f"{cond} AND NOT {mem}"))
|
|
else:
|
|
add(sclopt.Assign(tmp, f"NOT ({cond}) AND {mem}"))
|
|
add(sclopt.Assign(mem, cond))
|
|
groups[0] = BracketGroup()
|
|
groups[0].chain_expr = tmp
|
|
fresh[0] = False
|
|
continue
|
|
|
|
# --- Statuswort/BR ---
|
|
if mn == "SAVE":
|
|
expr = read_top()
|
|
add(sclopt.Assign("ENO", expr))
|
|
continue
|
|
|
|
# --- Spruenge ---
|
|
if mn in SUPPORTED_JUMP_TOKENS:
|
|
target = op.strip().rstrip(":")
|
|
if mn == "JU":
|
|
add(sclopt.Goto(target))
|
|
elif mn == "JC":
|
|
expr = read_top()
|
|
add(sclopt.IfDo(expr, [sclopt.Goto(target)]))
|
|
else: # JCN
|
|
expr = read_top()
|
|
add(sclopt.IfDo(f"NOT ({expr})", [sclopt.Goto(target)]))
|
|
continue
|
|
if mn in UNSUPPORTED_JUMP_TOKENS:
|
|
raise UnsupportedConstruct(f"Sprungmnemonik {mn} (statuswortabhaengig) nicht unterstuetzt")
|
|
|
|
# --- Bausteinende ---
|
|
if mn in ("BE", "BEU"):
|
|
add(sclopt.Return())
|
|
continue
|
|
if mn == "BEC":
|
|
expr = read_top()
|
|
fresh[0] = True
|
|
add(sclopt.IfDo(expr, [sclopt.Return()]))
|
|
continue
|
|
|
|
# --- CALL ---
|
|
if mn == "CALL":
|
|
add(sclopt.Call(translate_call(op), st.comment))
|
|
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 nodes
|
|
|
|
|
|
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, nodes in translated_networks:
|
|
label = title if title else ""
|
|
out.append(f" REGION {label}".rstrip())
|
|
for ln in sclopt.render_ir(nodes):
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
GENERATED_MARKER = "// Mechanische AWL->SCL-Konvertierung (awl2scl.py)"
|
|
|
|
|
|
class Outcome:
|
|
TRANSLATED = "translated"
|
|
SKIPPED_EXISTS = "skipped_exists"
|
|
SKIPPED_PROTECTED = "skipped_protected"
|
|
SKIPPED_CATEGORY = "skipped_category"
|
|
MANUAL_REVIEW = "manual_review"
|
|
ERROR = "error"
|
|
|
|
|
|
class Result:
|
|
def __init__(self, path, outcome, category=None, reason="", scl_path=None, jumps=0,
|
|
params=None, callers=None):
|
|
self.path = path
|
|
self.outcome = outcome
|
|
self.category = category
|
|
self.reason = reason
|
|
self.scl_path = scl_path
|
|
self.jumps = jumps
|
|
self.params = params or [] # neue Parameter (globals2params)
|
|
self.callers = callers or [] # (caller_name, status) fuer toChange
|
|
|
|
|
|
def _is_generated(scl_path):
|
|
"""True, wenn die vorhandene .scl von awl2scl erzeugt wurde (Marker-Kopfzeile)."""
|
|
try:
|
|
head = scl_path.read_text(encoding="utf-8", errors="replace")[:600]
|
|
except OSError:
|
|
return False
|
|
return GENERATED_MARKER in head
|
|
|
|
|
|
def process_file(awl_path, requested_categories, target_dir, console, opts):
|
|
try:
|
|
parsed = parse_awl_file(awl_path)
|
|
except DataBlockError as exc:
|
|
console(f"UEBERSPRUNGEN [Datenbaustein, keine Logik] {awl_path}: {exc}")
|
|
return Result(awl_path, Outcome.SKIPPED_CATEGORY, reason=f"Datenbaustein (no logic): {exc}")
|
|
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():
|
|
if not opts.force:
|
|
console(f"UEBERSPRUNGEN [Ziel existiert bereits] {awl_path} -> {scl_path}")
|
|
return Result(awl_path, Outcome.SKIPPED_EXISTS, category=category, scl_path=scl_path)
|
|
if not _is_generated(scl_path):
|
|
console(f"GESCHUETZT [handgeschrieben, nicht ueberschrieben] {scl_path}")
|
|
return Result(awl_path, Outcome.SKIPPED_PROTECTED, category=category, scl_path=scl_path)
|
|
|
|
translator = Translator(parsed, awl_path)
|
|
try:
|
|
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}")
|
|
|
|
new_params = []
|
|
callers = []
|
|
if opts.globals2params:
|
|
try:
|
|
new_params, callers = opts.paramizer.parameterize(
|
|
parsed, translated, translator, station_of(awl_path), awl_path)
|
|
except Exception as exc:
|
|
console(f"FEHLER (globals2params) {awl_path}: {exc}")
|
|
return Result(awl_path, Outcome.ERROR, category=category,
|
|
reason=f"globals2params-Fehler: {exc}")
|
|
|
|
# Aufruf-Rewrites, die frueher fuer DIESEN Baustein vermerkt wurden, anwenden
|
|
if opts.globals2params:
|
|
opts.paramizer.apply_pending_call_rewrites(parsed, translated, station_of(awl_path))
|
|
|
|
if opts.improve_gotos:
|
|
translated = [(title, sclopt.improve_gotos(nodes)) for title, nodes in translated]
|
|
translated = sclopt.synth_case_across_networks(translated)
|
|
|
|
scl_text = render_scl(parsed, translated, translator.temp_vars)
|
|
scl_path.parent.mkdir(parents=True, exist_ok=True)
|
|
scl_path.write_text(scl_text, encoding="utf-8")
|
|
jumps = count_jumps(parsed)
|
|
extra = ""
|
|
if jumps > Translator.JUMP_HEAVY_THRESHOLD and not opts.improve_gotos:
|
|
extra = f" (GOTO-lastig: {jumps} Spruenge)"
|
|
if new_params:
|
|
extra += f" (+{len(new_params)} Parameter, {len(callers)} Aufrufer)"
|
|
console(f"OK [{category}] {awl_path} -> {scl_path}{extra}")
|
|
return Result(awl_path, Outcome.TRANSLATED, category=category, scl_path=scl_path,
|
|
jumps=jumps, params=new_params, callers=callers)
|
|
|
|
|
|
def write_log(log_path, results, requested_categories, target_dir, opts=None):
|
|
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}")
|
|
if opts is not None:
|
|
lines.append(f"--improve-gotos : {'ja' if opts.improve_gotos else 'nein'}")
|
|
lines.append(f"--globals2params : {'ja' if opts.globals2params else 'nein'}")
|
|
lines.append(f"Dateien gesamt : {len(results)}")
|
|
lines.append(f"Uebersetzt : {sum(1 for r in results if r.outcome == Outcome.TRANSLATED)}")
|
|
lines.append(f"Uebersprungen (Kategorie) : {sum(1 for r in results if r.outcome == Outcome.SKIPPED_CATEGORY)}")
|
|
lines.append(f"Uebersprungen (Ziel da) : {sum(1 for r in results if r.outcome == Outcome.SKIPPED_EXISTS)}")
|
|
lines.append(f"Geschuetzt (handgeschr.) : {sum(1 for r in results if r.outcome == Outcome.SKIPPED_PROTECTED)}")
|
|
lines.append(f"Manuelle Pruefung noetig : {sum(1 for r in results if r.outcome == Outcome.MANUAL_REVIEW)}")
|
|
lines.append(f"Fehler : {sum(1 for r in results if r.outcome == Outcome.ERROR)}")
|
|
lines.append("")
|
|
|
|
parametrized = [r for r in results if r.outcome == Outcome.TRANSLATED and r.params]
|
|
if parametrized:
|
|
lines.append("-" * 70)
|
|
lines.append(f"Parametrisiert (globals2params) ({len(parametrized)})")
|
|
lines.append("-" * 70)
|
|
for r in sorted(parametrized, key=lambda r: str(r.path)):
|
|
n_manual = sum(1 for _, status in r.callers if status == "manual")
|
|
lines.append(f"{r.scl_path}")
|
|
for g, name, direction in r.params:
|
|
lines.append(f" {direction:6s} {name} (war {g})")
|
|
lines.append(f" toChange: {len(r.callers)} Aufrufer, davon {n_manual} manuell")
|
|
lines.append("")
|
|
|
|
manual = [r for r in results if r.outcome == Outcome.MANUAL_REVIEW]
|
|
if manual:
|
|
lines.append("-" * 70)
|
|
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("")
|
|
|
|
protected = [r for r in results if r.outcome == Outcome.SKIPPED_PROTECTED]
|
|
if protected:
|
|
lines.append("-" * 70)
|
|
lines.append(f"Geschuetzt, handgeschriebene .scl nicht ueberschrieben ({len(protected)})")
|
|
lines.append("-" * 70)
|
|
for r in sorted(protected, key=lambda r: str(r.path)):
|
|
lines.append(f"{r.scl_path}")
|
|
lines.append("")
|
|
|
|
skipped_exists = [r for r in results if r.outcome == Outcome.SKIPPED_EXISTS]
|
|
if skipped_exists:
|
|
lines.append("-" * 70)
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Options:
|
|
def __init__(self, improve_gotos=False, globals2params=False, force=False, paramizer=None):
|
|
self.improve_gotos = improve_gotos
|
|
self.globals2params = globals2params
|
|
self.force = force
|
|
self.paramizer = paramizer
|
|
|
|
|
|
def default_tochange_path():
|
|
return Path(__file__).resolve().parent.parent / "data" / "globals2params_tochange.json"
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="Wandelt .awl-Dateien mechanisch in .scl-Dateien um.")
|
|
parser.add_argument("files", nargs="*", type=Path, help="Einzelne .awl-Dateien")
|
|
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 <Projekt>/SCL-Export)")
|
|
parser.add_argument("--improve-gotos", action="store_true",
|
|
help="Kontrollfluss aufraeumen (Guard->IF, dedup, CASE, Klammern) - verlustfrei")
|
|
parser.add_argument("--globals2params", action="store_true",
|
|
help="Globale Merker-Zugriffe in Parameter umbauen + toChange fuer Aufrufer (data/)")
|
|
parser.add_argument("--force", action="store_true",
|
|
help="Vorhandene generierte .scl ueberschreiben (handgeschriebene bleiben geschuetzt)")
|
|
parser.add_argument("--tochange", type=Path, default=None,
|
|
help="Pfad der toChange-Liste (Default: <Projekt>/data/globals2params_tochange.json)")
|
|
parser.add_argument("--log", type=Path, help="Pfad der Log-Datei")
|
|
parser.add_argument("--show-missing", action="store_true",
|
|
help="Am Ende alle nicht uebersetzten Bausteine (manuelle Pruefung + Fehler) "
|
|
"mit Grund auf der Konsole auflisten")
|
|
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()
|
|
|
|
# globals2params impliziert Regenerierung
|
|
force = args.force or args.globals2params or args.improve_gotos
|
|
|
|
store = None
|
|
if args.globals2params:
|
|
import paramize
|
|
roots = [args.directory] if args.directory else [t.parent for t in targets]
|
|
roots = list(dict.fromkeys(str(Path(r)) for r in roots))
|
|
store = paramize.ToChangeStore(args.tochange or default_tochange_path())
|
|
paramizer = paramize.Paramizer(paramize.CallerIndex(roots), store)
|
|
else:
|
|
import paramize
|
|
paramizer = paramize.NullParamizer()
|
|
|
|
opts = Options(improve_gotos=args.improve_gotos, globals2params=args.globals2params,
|
|
force=force, paramizer=paramizer)
|
|
|
|
def console(msg):
|
|
print(msg)
|
|
|
|
results = []
|
|
for awl_path in targets:
|
|
results.append(process_file(awl_path, requested_categories, target_dir, console, opts))
|
|
|
|
if store is not None:
|
|
store.save()
|
|
|
|
log_path = args.log or default_log_path()
|
|
write_log(log_path, results, requested_categories, target_dir, opts)
|
|
|
|
n_ok = sum(1 for r in results if r.outcome == Outcome.TRANSLATED)
|
|
n_manual = sum(1 for r in results if r.outcome == Outcome.MANUAL_REVIEW)
|
|
n_err = sum(1 for r in results if r.outcome == Outcome.ERROR)
|
|
print()
|
|
print(f"{len(results)} Dateien geprueft, {n_ok} uebersetzt, {n_manual} manuelle Pruefung, {n_err} Fehler.")
|
|
if args.globals2params and store is not None:
|
|
print(f"toChange-Liste: {store.path}")
|
|
print(f"Log: {log_path}")
|
|
|
|
if args.show_missing:
|
|
print_missing(results)
|
|
|
|
return 1 if n_err else 0
|
|
|
|
|
|
def print_missing(results):
|
|
"""Listet alle nicht uebersetzten Bausteine (manuelle Pruefung + Fehler) mit Grund auf."""
|
|
missing = [r for r in results if r.outcome in (Outcome.MANUAL_REVIEW, Outcome.ERROR)]
|
|
print()
|
|
print("=" * 70)
|
|
print(f"Nicht uebersetzt ({len(missing)})")
|
|
print("=" * 70)
|
|
if not missing:
|
|
print("(keine)")
|
|
return
|
|
for r in sorted(missing, key=lambda r: str(r.path)):
|
|
tag = "MANUELL" if r.outcome == Outcome.MANUAL_REVIEW else "FEHLER "
|
|
cat = f"[{r.category}] " if r.category else ""
|
|
print(f"{tag} {cat}{r.path}")
|
|
print(f" Grund: {r.reason}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|