809 lines
32 KiB
Python
809 lines
32 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""create_skel.py — Skeleton-Generator fuer die JSON-Zwischendatei und ein
|
|
SCL-Grundgeruest, abgeleitet aus den Eingabelisten in data/.
|
|
|
|
Datenquellen (Pfade ueber Umgebungsvariablen aus bin\\setenv.bat):
|
|
PV_DATA Ordner mit den Eingabelisten
|
|
ST500592_10_5-14_ILS-<n>_TIA.xlsx (PLC Tags + Constants)
|
|
ST500592_10_5-9_ILS_positions.json (Positionen, VERW, mappings)
|
|
PV_RESULTS Zielordner fuer die erzeugten Dateien
|
|
PV_CFG Ordner mit den Konfigurationsdateien, geladen beim Start:
|
|
skel.cfg (JSON) Dateinamen, Vokabular, FB-Typ-Regeln
|
|
skel_scl.cfg (INI-Style) SCL-Templates, zeilenweise
|
|
lesbar; jede [sektion] = ein Template,
|
|
Inhalt verbatim (Einrueckung/Leerzeilen
|
|
gehoeren zum Template)
|
|
Fehlende Dateien fallen auf die eingebauten Defaults zurueck.
|
|
|
|
Erzeugt je Steuerung UH01..UH05:
|
|
skeleton_UH0x.json JSON-Zwischendatei-Skeleton (Kap. 14 im
|
|
Json_Layout-Konzept.md)
|
|
FB_Main_skel_UH0x.scl SCL-Grundgeruest (VAR-Deklaration + Aufrufe)
|
|
FB_CallSensors_skel_UH0x.scl Sensor-Aufrufe
|
|
|
|
Mit --con-ini bzw. --con-json wird zusaetzlich die TRO-Topologie
|
|
(Routing / Zusammenhalt, Format hmf-connect-v1) aus data\\connect.ini
|
|
bzw. data\\connect.json eingelesen und als "topology"-Abschnitt in die
|
|
Skeleton-JSONs uebernommen (Verbindungen, offene Enden und je TRO die
|
|
aufgeloeste JamArea-Verdrahtung).
|
|
|
|
Aufruf: bin\\create_skel.bat [--uh 1..5] [--no-scl] [--con-ini | --con-json]
|
|
"""
|
|
|
|
import argparse
|
|
import configparser
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import openpyxl
|
|
except ImportError:
|
|
sys.exit("openpyxl fehlt — bitte bin\\install_py.bat ausfuehren.")
|
|
|
|
# ============================================================================
|
|
# GLOBALE DEFINITIONEN (eingebaute Defaults)
|
|
# Alle folgenden Bloecke sind bewusst als Modul-Konstanten definiert. Sie
|
|
# sind zugleich der Inhalt von cfg\\skel.cfg (JSON) — load_cfg() liest diese
|
|
# Datei aus PV_CFG beim Programmstart und ersetzt die Defaults hier 1:1
|
|
# durch ihren Inhalt. Ohne skel.cfg (z.B. Erststart, Datei geloescht) bleiben
|
|
# die Defaults unten aktiv, das Skript funktioniert also auch dann.
|
|
# ============================================================================
|
|
|
|
#: Dateinamen der Eingabelisten ({n} = ILS-Nummer 1..5)
|
|
FILES = {
|
|
"tia_xlsx": "ST500592_10_5-14_ILS-{n}_TIA.xlsx",
|
|
"positions_json": "ST500592_10_5-9_ILS_positions.json",
|
|
# Routing / Zusammenhalt der TROs (hmf-connect-v1, extrahiert aus
|
|
# doc/TRO_Graph_UH01-UH05.dot) — fuer die kuenftige Topologie-Auswertung
|
|
"connect_json": "connect.json",
|
|
"connect_ini": "connect.ini",
|
|
}
|
|
|
|
#: VERW-Vokabular (Positions-JSON) -> Signalrolle
|
|
#: (siehe doc/EA-Listen-Analyse.md, Abschnitt "Verwendung von Positionen")
|
|
VERW_ROLES = {
|
|
"In sep": "sensorInSep",
|
|
"Sep": "stopperOutput",
|
|
"Jam detector": "sensorJam",
|
|
"Jam detector (LP)": "sensorJam",
|
|
"jam detector (LP)": "sensorJam",
|
|
"ES branch": "switchOutput",
|
|
"ES brunch": "switchOutput", # Schreibvariante in den Daten
|
|
"conveyor full": "sensorConvFull",
|
|
"MT in Position": "sensorCarrInPos",
|
|
"MT in position": "sensorCarrInPos",
|
|
"Finger in position": "sensorFinger",
|
|
"Finger in Position": "sensorFinger",
|
|
"PIN query": "sensorPinQuery",
|
|
"Rad dreht ": "sensorWheel",
|
|
"Rad dreht": "sensorWheel",
|
|
}
|
|
|
|
#: Fallback: Regex auf den Kommentartext -> Signalrolle
|
|
#: (falls kein VERW vorliegt, z.B. Signal nicht im Positions-JSON)
|
|
COMMENT_ROLES = [
|
|
(r"^in Separator", "sensorInSep"),
|
|
(r"^Separator", "stopperOutput"),
|
|
(r"Stausensor|Staumelder", "sensorJam"),
|
|
(r"AE Abzweig", "switchOutput"),
|
|
(r"F.?rderer voll", "sensorConvFull"),
|
|
(r"MT in Position", "sensorCarrInPos"),
|
|
(r"Mitnehmerfinger|Finger in", "sensorFinger"),
|
|
(r"Pin Abfrage|Pr.?fung MT Pos", "sensorPinQuery"),
|
|
(r"Rad dreht", "sensorWheel"),
|
|
(r"^Motor ", "motorOutput"),
|
|
(r"MSS ausgel", "motorProtection"),
|
|
]
|
|
|
|
#: Regeln zur FB-Typ-Bestimmung je Einheit.
|
|
#: Reihenfolge = Prioritaet; "roles" muessen alle vorhanden sein.
|
|
FBTYPE_RULES = [
|
|
{"roles": ["sensorPinQuery"], "fbType": "FB_ILS_MTRO_PinStore_Auto"},
|
|
{"roles": ["sensorFinger"], "fbType": "FB_ILS_MTRO_Vario"},
|
|
{"roles": ["sensorCarrInPos"], "fbType": "FB_ILS_MTRO_Vario"},
|
|
{"roles": ["switchOutput", "sensorInSep"], "fbType": "FB_ILS_MTRO_1Sep1Swi"},
|
|
{"roles": ["sensorInSep"], "fbType": "FB_ILS_MTRO_1Sep"},
|
|
{"roles": ["motorOutput"], "fbType": "FB_Conveyor"},
|
|
{"roles": ["motorProtection"], "fbType": "FB_Conveyor"},
|
|
]
|
|
|
|
#: Erkennung der Foerdertechnik-Einheit (M-Nummer) im Kommentartext
|
|
UNIT_RE = r"\b(KR|CV)[-\s]{0,2}(P?[MK]\d{4}(?:_[A-Z])?)\b"
|
|
|
|
#: Erkennung der Bereichsgruppen im Kommentartext
|
|
AREA_RES = [
|
|
(r"Block ?(\d\.\d(?:\.\d)?)", "pinStoreBlock", "Block {m}"),
|
|
(r"\bLS ?(\d)\b", "loadingStation", "LS{m}"),
|
|
(r"\bWS ?(\d\.\d)\b", "workStation", "WS{m}"),
|
|
(r"\bUZ(\d{4})\b", "safetyZone", "UZ{m}"),
|
|
]
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# SCL-Templates: die "ueblichen Codebloecke" fuer die Aufrufe in FB_Main /
|
|
# FB_CallSensors. Platzhalter im str.format()-Stil.
|
|
# Eingebaute Defaults — die gepflegte, zeilenweise lesbare Fassung liegt in
|
|
# cfg\skel_scl.cfg (INI-Style) und ueberschreibt diese beim Programmstart.
|
|
# ----------------------------------------------------------------------------
|
|
SCL_TEMPLATES = {
|
|
# VAR-Deklaration einer Instanz
|
|
"var_decl": ' {inst} : "{fbtype}"; // {comment}\n',
|
|
|
|
# Aufruf FB_ILS_MTRO_1Sep (Standard-Separator)
|
|
"call_1sep": """\
|
|
REGION {inst} ({shorttype}) — {comment}
|
|
#{inst}.stInSeparator1.Settings.nSeparatorNo := "cSep{tro_no}.1";
|
|
#{inst}.stInSeparator1.Settings.tDelayToNextItem := "TimeForSepWaiting_Short";
|
|
#{inst}.stInSeparator1.Settings.tTrailingTime := T#500ms; // TODO Timing
|
|
#{inst}.stInSeparator1.Settings.tHandlingTime := T#200ms; // TODO Timing
|
|
#{inst}.stInSeparator1.Settings.tJam := "TimeForSepJam";
|
|
#{inst}.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cIn{sen_in_sep}"];
|
|
#{inst}.stInSenJam2 := "DB_Inputs".Sensors["cIn{sen_jam}"];
|
|
|
|
#{inst}.stInPriorityManager.xReleaseOk := TRUE; // TODO Priority
|
|
|
|
#{inst}(nInMainTroNo := "cMainTro{tro_no}",
|
|
xInSftyOk := TRUE, // TODO Safety
|
|
xInAllRdyToStart := TRUE,
|
|
xInCarouselRun := TRUE, // TODO Conveyor
|
|
xInRelease := TRUE, // TODO Release
|
|
stInOutMachineState := #stInOutControlUnit, // TODO ControlUnit
|
|
arInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.arCarrier, // TODO Topologie
|
|
stInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.stData,
|
|
arInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.arCarrier, // TODO Topologie
|
|
stInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.stData,
|
|
xOutStopper1 => "{stopper}",
|
|
nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro{tro_no}"],
|
|
stInOutHMI := "DB_Interface_HMI".stTRO.{inst});
|
|
END_REGION
|
|
|
|
""",
|
|
|
|
# Aufruf FB_ILS_MTRO_1Sep1Swi (Separator + Weiche)
|
|
"call_1sep1swi": """\
|
|
REGION {inst} ({shorttype}) — {comment}
|
|
#{inst}.stInSeparator1.Settings.nSeparatorNo := "cSep{tro_no}.1";
|
|
#{inst}.stInSeparator1.Settings.tDelayToNextItem := "TimeForSepWaiting_Short";
|
|
#{inst}.stInSeparator1.Settings.tTrailingTime := T#100ms; // TODO Timing
|
|
#{inst}.stInSeparator1.Settings.tHandlingTime := T#50ms; // TODO Timing
|
|
#{inst}.stInSeparator1.Settings.tJam := "TimeForSepJam";
|
|
#{inst}.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cIn{sen_in_sep}"];
|
|
#{inst}.stInSenJam2 := "DB_Inputs".Sensors["cIn{sen_jam}"];
|
|
#{inst}.stInSwitch1.Settings.nSwitchNo := "cSwi{tro_no}.1";
|
|
|
|
#{inst}(nInMainTroNo := "cMainTro{tro_no}",
|
|
xInSftyOk := TRUE, // TODO Safety
|
|
xInAllRdyToStart := TRUE,
|
|
xInCarouselRun1 := TRUE, // TODO Conveyor
|
|
xInRelease := TRUE, // TODO Release
|
|
stInOutMachineState := #stInOutControlUnit, // TODO ControlUnit
|
|
arInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.arCarrier, // TODO Topologie
|
|
stInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.stData,
|
|
arInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.arCarrier, // TODO Topologie (Dir1)
|
|
stInOutJamExit2 := "DB_JamArea{tro_no}".stJam{tro_no}_2.stData,
|
|
arInOutJamExit3 := "DB_JamArea{tro_no}".stJam{tro_no}_3.arCarrier, // TODO Topologie (Dir2)
|
|
stInOutJamExit3 := "DB_JamArea{tro_no}".stJam{tro_no}_3.stData,
|
|
xOutStopper1 => "{stopper}",
|
|
xOutSw1ExTo3 => "{switch_out}",
|
|
nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro{tro_no}"],
|
|
stInOutHMI := "DB_Interface_HMI".stTRO.{inst});
|
|
END_REGION
|
|
|
|
""",
|
|
|
|
# Aufruf FB_ILS_MTRO_Vario (Kettenfoerderer)
|
|
"call_vario": """\
|
|
REGION {inst} ({shorttype}) — {comment}
|
|
#{inst}.stInSeparator1.Settings.nSeparatorNo := "cSep{tro_no}.1";
|
|
#{inst}.stInSeparator1.Settings.tDelayToNextItem := "TimeForSepWaiting_Short";
|
|
#{inst}.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cIn{sen_in_sep}"];
|
|
#{inst}.stInVario.stSenCarrInPos := "DB_Inputs".Sensors["cIn{sen_carr_in_pos}"];
|
|
#{inst}.stInVario.stSenFinger := "DB_Inputs".Sensors["cIn{sen_finger}"];
|
|
#{inst}.stInVario.stSenJam := "DB_Inputs".Sensors["cIn{sen_jam}"];
|
|
|
|
#{inst}(nInMainTroNo := "cMainTro{tro_no}",
|
|
xInSftyOk := TRUE, // TODO Safety
|
|
xInAllRdyToStart := TRUE,
|
|
xInRelease := TRUE, // TODO Release
|
|
stInOutMachineState := #stInOutControlUnit, // TODO ControlUnit
|
|
arInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.arCarrier, // TODO Topologie
|
|
stInOutJamEntr1 := "DB_JamArea{tro_no}".stJam{tro_no}_1.stData,
|
|
arInOutJamFinger := "DB_JamArea{tro_no}".stJam{tro_no}_2.arCarrier, // TODO Topologie
|
|
xOutStopper => "{stopper}",
|
|
xOutVarioMotor => "{motor}",
|
|
nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro{tro_no}"],
|
|
stInOutHMI := "DB_Interface_HMI".stTRO.{inst});
|
|
END_REGION
|
|
|
|
""",
|
|
|
|
# Aufruf FB_Conveyor (Kreisel / Foerderer ohne TRO-Logik)
|
|
"call_conveyor": """\
|
|
REGION {inst} — {comment}
|
|
#{inst}.Settings.tEnergySafeTime := "cTimeEnergySafe";
|
|
#{inst}.Settings.xGridActive := FALSE; // TODO Grid
|
|
#{inst}.Settings.sMotorProtectName := '{motor_protection}';
|
|
|
|
#{inst}(xInEnabled := TRUE, // TODO Enable
|
|
xInMotorProtection := "{motor_protection}",
|
|
xInPowerContactor := "{motor}",
|
|
stInOutHMI := "DB_Interface_HMI".stConveyors.{inst});
|
|
END_REGION
|
|
|
|
""",
|
|
|
|
# Eintrag in FB_CallSensors je Sensor
|
|
"call_sensor": """\
|
|
// {comment}
|
|
#fbSensorInput(stInSensorHW := "{bmk}",
|
|
nInSensorNo := "cIn{bmk}",
|
|
stInOutSensor := "DB_Inputs".Sensors["cIn{bmk}"]);
|
|
""",
|
|
|
|
# Rahmen FB_Main
|
|
"fb_main_frame": """\
|
|
FUNCTION_BLOCK "FB_Main_skel_{uh}"
|
|
{{ S7_Optimized_Access := 'TRUE' }}
|
|
VERSION : 0.1
|
|
// ============================================================
|
|
// AUTOMATISCH GENERIERTES GRUNDGERUEST — create_skel.py
|
|
// Quelle: E/A-Listen {uh} (data/)
|
|
// TODO-Marken kennzeichnen manuell zu ergaenzende Stellen
|
|
// (Timings, Topologie/JamAreas, Safety, Release, Priority).
|
|
// ============================================================
|
|
VAR
|
|
{var_block} END_VAR
|
|
|
|
BEGIN
|
|
{call_block}END_FUNCTION_BLOCK
|
|
""",
|
|
|
|
# Rahmen FB_CallSensors
|
|
"fb_sensors_frame": """\
|
|
FUNCTION_BLOCK "FB_CallSensors_skel_{uh}"
|
|
{{ S7_Optimized_Access := 'TRUE' }}
|
|
VERSION : 0.1
|
|
// AUTOMATISCH GENERIERT — create_skel.py ({count} Sensoren)
|
|
VAR
|
|
fbSensorInput : "FB_SensorInput";
|
|
END_VAR
|
|
|
|
BEGIN
|
|
{call_block}END_FUNCTION_BLOCK
|
|
""",
|
|
}
|
|
|
|
# Zuordnung FB-Typ -> Template-Schluessel
|
|
FBTYPE_TEMPLATE = {
|
|
"FB_ILS_MTRO_1Sep": "call_1sep",
|
|
"FB_ILS_MTRO_1Sep1Swi": "call_1sep1swi",
|
|
"FB_ILS_MTRO_Vario": "call_vario",
|
|
"FB_ILS_MTRO_PinStore_Auto": "call_1sep", # eigenes Template folgt spaeter
|
|
"FB_Conveyor": "call_conveyor",
|
|
}
|
|
|
|
# ============================================================================
|
|
# Ende der globalen Definitionen
|
|
# ============================================================================
|
|
|
|
#: Schluessel in skel.cfg -> Modul-Konstante, die beim Laden ersetzt wird
|
|
_CFG_KEYS = {
|
|
"files": "FILES",
|
|
"verw_roles": "VERW_ROLES",
|
|
"comment_roles": "COMMENT_ROLES",
|
|
"fbtype_rules": "FBTYPE_RULES",
|
|
"unit_re": "UNIT_RE",
|
|
"area_res": "AREA_RES",
|
|
"scl_templates": "SCL_TEMPLATES",
|
|
"fbtype_template": "FBTYPE_TEMPLATE",
|
|
}
|
|
|
|
|
|
def load_cfg():
|
|
"""Laedt cfg\\skel.cfg (JSON) aus PV_CFG und ersetzt die globalen
|
|
Definitionen (FILES, VERW_ROLES, ..., SCL_TEMPLATES) durch ihren Inhalt.
|
|
|
|
Fehlt PV_CFG oder die Datei, bleiben die eingebauten Defaults oben aktiv
|
|
(Cold-Start-Faehigkeit ohne Konfigurationsdatei).
|
|
"""
|
|
cfg_dir = os.environ.get("PV_CFG", Path(__file__).resolve().parent.parent / "cfg")
|
|
cfg_file = Path(cfg_dir) / "skel.cfg"
|
|
if not cfg_file.exists():
|
|
print(f"Hinweis: {cfg_file} nicht gefunden — verwende eingebaute Defaults.")
|
|
return
|
|
|
|
print(f"Lade Konfiguration: {cfg_file}")
|
|
data = json.loads(cfg_file.read_text(encoding="utf-8"))
|
|
|
|
globals_ = globals()
|
|
for cfg_key, const_name in _CFG_KEYS.items():
|
|
if cfg_key not in data:
|
|
continue
|
|
value = data[cfg_key]
|
|
# Listen von Listen (comment_roles, fbtype_rules, area_res) als
|
|
# Tupel normalisieren, damit das Unpacking im Code unveraendert bleibt
|
|
if const_name in ("COMMENT_ROLES", "AREA_RES"):
|
|
value = [tuple(item) for item in value]
|
|
globals_[const_name] = value
|
|
|
|
|
|
def load_scl_cfg():
|
|
"""Laedt die SCL-Templates aus PV_CFG\\skel_scl.cfg (INI-Style).
|
|
|
|
Format: jede [sektion] ist ein Template-Schluessel (var_decl, call_1sep,
|
|
...); der Sektionsinhalt wird VERBATIM uebernommen — Einrueckung und
|
|
Leerzeilen (auch am Blockende) gehoeren zum Template. Zeilen, die mit
|
|
';' beginnen, sind Kommentare. SCL selbst beginnt nie mit ';', daher
|
|
ist das kollisionsfrei.
|
|
|
|
Fehlt die Datei, bleiben die Templates aus skel.cfg bzw. den eingebauten
|
|
Defaults aktiv.
|
|
"""
|
|
cfg_dir = os.environ.get("PV_CFG", Path(__file__).resolve().parent.parent / "cfg")
|
|
cfg_file = Path(cfg_dir) / "skel_scl.cfg"
|
|
if not cfg_file.exists():
|
|
print(f"Hinweis: {cfg_file} nicht gefunden — "
|
|
f"verwende SCL-Templates aus skel.cfg/Defaults.")
|
|
return
|
|
|
|
print(f"Lade SCL-Templates: {cfg_file}")
|
|
templates = {}
|
|
name, buf = None, []
|
|
|
|
def flush():
|
|
if name is not None:
|
|
templates[name] = "\n".join(buf) + "\n"
|
|
|
|
for line in cfg_file.read_text(encoding="utf-8").splitlines():
|
|
if line.startswith(";"):
|
|
continue
|
|
m = re.fullmatch(r"\[(\w+)\]\s*", line)
|
|
if m:
|
|
flush()
|
|
name, buf = m.group(1), []
|
|
elif name is not None:
|
|
buf.append(line)
|
|
flush()
|
|
SCL_TEMPLATES.update(templates)
|
|
print(f"SCL-Templates geladen: {', '.join(templates)}")
|
|
|
|
|
|
def project_paths():
|
|
"""Pfade aus den setenv.bat-Umgebungsvariablen (mit Fallbacks)."""
|
|
root = Path(os.environ.get("PROJECT", Path(__file__).resolve().parent.parent))
|
|
data_dir = Path(os.environ.get("PV_DATA", root / "data"))
|
|
results = Path(os.environ.get("PV_RESULTS", root / "results"))
|
|
results.mkdir(parents=True, exist_ok=True)
|
|
return root, data_dir, results
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Einlesen
|
|
# ----------------------------------------------------------------------------
|
|
|
|
def read_positions(data_dir):
|
|
"""Positions-JSON: (BMK, SPS) -> {verw, pos, cabinet, bezeichnung}."""
|
|
path = data_dir / FILES["positions_json"]
|
|
if not path.exists():
|
|
print(f"WARNUNG: {path} nicht gefunden — Positionen/VERW entfallen.")
|
|
return {}
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
index = {}
|
|
for section in ("sensors", "schaltschrank_elemente"):
|
|
for key, val in data.get(section, {}).items():
|
|
if "@" not in key:
|
|
continue
|
|
bmk, sps = key.rsplit("@", 1)
|
|
index[(bmk, sps)] = {
|
|
"verw": (val.get("VERW") or "").strip(),
|
|
"pos": val.get("pos"),
|
|
"bezeichnung": val.get("BEZEICHNUNG") or "",
|
|
"kennzeichnung": val.get("KENNZEICHNUNG") or "",
|
|
}
|
|
# mappings: Verteiler -> [BMK@SPS] umdrehen zu (BMK, SPS) -> Verteiler
|
|
for dist, members in data.get("mappings", {}).items():
|
|
for key in members:
|
|
if "@" not in key:
|
|
continue
|
|
bmk, sps = key.rsplit("@", 1)
|
|
index.setdefault((bmk, sps), {})["cabinet"] = dist
|
|
return index
|
|
|
|
|
|
def read_tia(data_dir, n):
|
|
"""TIA-Liste ILS-n: Liste von Tag-Dicts + Konstanten-Map."""
|
|
path = data_dir / FILES["tia_xlsx"].format(n=n)
|
|
wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
|
|
tags = []
|
|
for row in wb["PLC Tags"].iter_rows(min_row=2, values_only=True):
|
|
name, _path, dtype, addr, comment = (tuple(row) + (None,) * 5)[:5]
|
|
if not name:
|
|
continue
|
|
tags.append({
|
|
"name": str(name),
|
|
"dataType": str(dtype) if dtype else "Bool",
|
|
"address": str(addr) if addr else "",
|
|
"comment": str(comment) if comment else "",
|
|
})
|
|
constants = {}
|
|
if "Constants" in wb.sheetnames:
|
|
for row in wb["Constants"].iter_rows(min_row=2, values_only=True):
|
|
cname, _path, _dt, value = (tuple(row) + (None,) * 4)[:4]
|
|
if cname:
|
|
constants[str(cname)] = value
|
|
return tags, constants
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Connect-Daten (Topologie, Format hmf-connect-v1)
|
|
# ----------------------------------------------------------------------------
|
|
|
|
def expand_jam(short):
|
|
"""JA0101_1 -> "DB_JamArea0101".stJam0101_1 ; sonst unveraendert."""
|
|
m = re.fullmatch(r"JA(\w+)_(\w+)", short)
|
|
if m:
|
|
return f'"DB_JamArea{m.group(1)}".stJam{m.group(1)}_{m.group(2)}'
|
|
return short
|
|
|
|
|
|
def read_connect_json(data_dir):
|
|
"""connect.json -> {UHxx: {nodes, connections, externals}}."""
|
|
path = data_dir / FILES["connect_json"]
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return data.get("plcs", {})
|
|
|
|
|
|
def read_connect_ini(data_dir):
|
|
"""connect.ini -> gleiche Struktur wie read_connect_json()."""
|
|
path = data_dir / FILES["connect_ini"]
|
|
cp = configparser.ConfigParser(interpolation=None)
|
|
cp.optionxform = str # Gross-/Kleinschreibung der TRO-Namen erhalten
|
|
cp.read(path, encoding="utf-8")
|
|
plcs = {}
|
|
for section in cp.sections():
|
|
uh, _, part = section.partition(".")
|
|
plc = plcs.setdefault(uh, {"nodes": {}, "connections": [],
|
|
"externals": []})
|
|
if part == "nodes":
|
|
for name, val in cp[section].items():
|
|
fbtype, _, comment = (x.strip() for x in val.partition("|"))
|
|
plc["nodes"][name] = {"fbType": fbtype, "comment": comment}
|
|
elif part == "connections":
|
|
for _, val in cp[section].items():
|
|
fields = [x.strip() for x in val.split("|")]
|
|
src, _, dst = (x.strip() for x in fields[0].partition("->"))
|
|
plc["connections"].append({
|
|
"from": src, "to": dst, "jamArea": fields[1],
|
|
"kind": fields[2] if len(fields) > 2 else "normal",
|
|
"dbRef": expand_jam(fields[1])})
|
|
elif part == "externals":
|
|
for _, val in cp[section].items():
|
|
route, _, jam = (x.strip() for x in val.partition("|"))
|
|
src, _, dst = (x.strip() for x in route.partition("->"))
|
|
node, direction = (dst, "in") if src == "EXTERN" else (src, "out")
|
|
plc["externals"].append({
|
|
"node": node, "direction": direction, "jamArea": jam,
|
|
"dbRef": expand_jam(jam)})
|
|
return plcs
|
|
|
|
|
|
def derive_topology(topo):
|
|
"""Bereitet die Connect-Daten einer Steuerung fuer das Skeleton auf:
|
|
je TRO die aufgeloeste JamArea-Verdrahtung (entries/exits)."""
|
|
tros = {name: {"id": name, "fbType": nd["fbType"], "comment": nd["comment"],
|
|
"entries": [], "exits": []}
|
|
for name, nd in topo["nodes"].items()}
|
|
for c in topo["connections"]:
|
|
db = c.get("dbRef") or expand_jam(c["jamArea"])
|
|
if c["from"] in tros:
|
|
tros[c["from"]]["exits"].append(
|
|
{"jamArea": c["jamArea"], "dbRef": db, "to": c["to"],
|
|
"kind": c["kind"]})
|
|
if c["to"] in tros:
|
|
tros[c["to"]]["entries"].append(
|
|
{"jamArea": c["jamArea"], "dbRef": db, "from": c["from"],
|
|
"kind": c["kind"]})
|
|
for e in topo["externals"]:
|
|
if e["node"] not in tros:
|
|
continue
|
|
db = e.get("dbRef") or expand_jam(e["jamArea"])
|
|
side = "entries" if e["direction"] == "in" else "exits"
|
|
peer = "from" if e["direction"] == "in" else "to"
|
|
tros[e["node"]][side].append(
|
|
{"jamArea": e["jamArea"], "dbRef": db, peer: "EXTERN",
|
|
"kind": "extern"})
|
|
return {
|
|
"format": "hmf-connect-v1",
|
|
"tros": [tros[k] for k in sorted(tros)],
|
|
"connections": topo["connections"],
|
|
"externals": topo["externals"],
|
|
}
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Klassifikation
|
|
# ----------------------------------------------------------------------------
|
|
|
|
def classify_role(tag, posinfo):
|
|
"""Signalrolle: primaer VERW-Vokabular, sonst Kommentar-Heuristik."""
|
|
verw = (posinfo or {}).get("verw", "")
|
|
if verw in VERW_ROLES:
|
|
return VERW_ROLES[verw]
|
|
for pattern, role in COMMENT_ROLES:
|
|
if re.search(pattern, tag["comment"], re.I):
|
|
return role
|
|
return ""
|
|
|
|
|
|
def extract_unit(comment):
|
|
"""M-Nummer der Foerdertechnik-Einheit aus dem Kommentar."""
|
|
m = re.search(UNIT_RE, comment)
|
|
if m:
|
|
return m.group(2), m.group(1) # ("M0101", "KR")
|
|
return None, None
|
|
|
|
|
|
def extract_areas(comment):
|
|
"""Bereichsgruppen (Block/LS/WS/UZ) aus dem Kommentar."""
|
|
found = []
|
|
for pattern, kind, fmt in AREA_RES:
|
|
for m in re.finditer(pattern, comment, re.I):
|
|
found.append({"id": fmt.format(m=m.group(1)), "kind": kind})
|
|
return found
|
|
|
|
|
|
def guess_fbtype(roles):
|
|
"""FB-Typ einer Einheit aus der Menge ihrer Signalrollen."""
|
|
for rule in FBTYPE_RULES:
|
|
if all(r in roles for r in rule["roles"]):
|
|
return rule["fbType"]
|
|
return "UNBEKANNT"
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Skeleton-Aufbau
|
|
# ----------------------------------------------------------------------------
|
|
|
|
def build_skeleton(n, tags, constants, positions, connect=None):
|
|
"""Baut das JSON-Skeleton fuer eine Steuerung (ILS-n = UH0n)."""
|
|
uh = f"UH0{n}"
|
|
sps = str(n)
|
|
|
|
sensors = []
|
|
units = defaultdict(lambda: {"signals": [], "roles": set(), "areas": [],
|
|
"kind": None, "pos": None, "cabinets": set()})
|
|
areas = defaultdict(lambda: {"kind": None, "members": set(),
|
|
"cabinets": set()})
|
|
unassigned = []
|
|
|
|
for tag in tags:
|
|
posinfo = positions.get((tag["name"], sps))
|
|
role = classify_role(tag, posinfo)
|
|
unit_id, unit_kind = extract_unit(tag["comment"])
|
|
area_hits = extract_areas(tag["comment"])
|
|
cabinet = (posinfo or {}).get("cabinet")
|
|
|
|
entry = {
|
|
"id": tag["name"],
|
|
"constant": f"cIn{tag['name']}" if tag["address"].startswith("%E") else None,
|
|
"hwAddress": tag["address"],
|
|
"comment": tag["comment"],
|
|
"role": role or None,
|
|
"unit": unit_id,
|
|
"cabinet": cabinet,
|
|
"pos": (posinfo or {}).get("pos"),
|
|
}
|
|
if tag["address"].startswith("%E"):
|
|
sensors.append(entry)
|
|
|
|
if unit_id:
|
|
u = units[unit_id]
|
|
u["kind"] = unit_kind
|
|
u["signals"].append(entry)
|
|
if role:
|
|
u["roles"].add(role)
|
|
if u["pos"] is None and entry["pos"]:
|
|
u["pos"] = entry["pos"]
|
|
if cabinet:
|
|
u["cabinets"].add(cabinet)
|
|
for a in area_hits:
|
|
if a not in u["areas"]:
|
|
u["areas"].append(a)
|
|
elif not area_hits and not role:
|
|
unassigned.append(tag["name"])
|
|
|
|
for a in area_hits:
|
|
ar = areas[a["id"]]
|
|
ar["kind"] = a["kind"]
|
|
if unit_id:
|
|
ar["members"].add(unit_id)
|
|
if cabinet:
|
|
ar["cabinets"].add(cabinet)
|
|
|
|
unit_list = []
|
|
for unit_id in sorted(units):
|
|
u = units[unit_id]
|
|
fbtype = guess_fbtype(u["roles"])
|
|
io = {}
|
|
for sig in u["signals"]:
|
|
if sig["role"]:
|
|
io.setdefault(sig["role"], []).append(sig["id"])
|
|
unit_list.append({
|
|
"unit": unit_id,
|
|
"unitKind": u["kind"],
|
|
"fbType": fbtype,
|
|
"areas": u["areas"],
|
|
"cabinet": sorted(u["cabinets"])[0] if u["cabinets"] else None,
|
|
"pos": u["pos"],
|
|
"io": io,
|
|
"signalCount": len(u["signals"]),
|
|
})
|
|
|
|
area_list = [{
|
|
"id": aid,
|
|
"kind": a["kind"],
|
|
"members": sorted(a["members"]),
|
|
"cabinets": sorted(a["cabinets"]),
|
|
} for aid, a in sorted(areas.items())]
|
|
|
|
skel = {
|
|
"$schema": "hmf-layout-v1.schema.json",
|
|
"generatedBy": "create_skel.py",
|
|
"plc": {"id": uh, "configName": f"=A01+{uh}-KF00"},
|
|
"statistics": {
|
|
"tags": len(tags),
|
|
"sensors": len(sensors),
|
|
"constants": len(constants),
|
|
"units": len(unit_list),
|
|
"areas": len(area_list),
|
|
"unassignedSignals": len(unassigned),
|
|
},
|
|
"sensors": sensors,
|
|
"units": unit_list,
|
|
"areas": area_list,
|
|
"unassignedSignals": sorted(unassigned),
|
|
"todo": [
|
|
"connections[] (Topologie) manuell ergaenzen oder aus SCL extrahieren",
|
|
"Timings, Priority, Release, customCode manuell pflegen",
|
|
"TRO-IDs den Einheiten zuweisen (Schema TRO<UH><lfd>)",
|
|
],
|
|
}
|
|
|
|
if connect is not None:
|
|
topo = connect.get(uh)
|
|
if topo:
|
|
skel["topology"] = derive_topology(topo)
|
|
skel["statistics"]["troNodes"] = len(topo["nodes"])
|
|
skel["statistics"]["connections"] = len(topo["connections"])
|
|
skel["statistics"]["externals"] = len(topo["externals"])
|
|
skel["todo"] = [
|
|
"topology[] aus connect-Daten uebernommen — TRO-IDs den "
|
|
"Einheiten (units[]) noch zuordnen",
|
|
"Timings, Priority, Release, customCode manuell pflegen",
|
|
]
|
|
else:
|
|
print(f"WARNUNG: keine Connect-Daten fuer {uh} gefunden.")
|
|
|
|
return skel
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# SCL-Rendering
|
|
# ----------------------------------------------------------------------------
|
|
|
|
def first(io, role, default="BGxxxx"):
|
|
vals = io.get(role)
|
|
return vals[0] if vals else default
|
|
|
|
|
|
def render_scl(skeleton, uh):
|
|
"""FB_Main- und FB_CallSensors-Grundgeruest aus den Templates."""
|
|
var_block = []
|
|
call_block = []
|
|
tro_seq = 0
|
|
for u in skeleton["units"]:
|
|
fbtype = u["fbType"]
|
|
if fbtype == "UNBEKANNT":
|
|
continue
|
|
tro_seq += 1
|
|
inst = f"TRO{uh[-1]}{tro_seq:02d}" if fbtype != "FB_Conveyor" \
|
|
else f"fbConveyor{u['unit'][1:]}"
|
|
comment = f"{u['unitKind']}-{u['unit']}" + \
|
|
(f" / {u['areas'][0]['id']}" if u["areas"] else "")
|
|
var_block.append(SCL_TEMPLATES["var_decl"].format(
|
|
inst=inst, fbtype=fbtype, comment=comment))
|
|
|
|
tpl = SCL_TEMPLATES[FBTYPE_TEMPLATE[fbtype]]
|
|
io = u["io"]
|
|
call_block.append(tpl.format(
|
|
inst=inst,
|
|
shorttype=fbtype.replace("FB_ILS_MTRO_", "").replace("FB_", ""),
|
|
comment=comment,
|
|
tro_no=f"{uh[-1]}{tro_seq:02d}",
|
|
sen_in_sep=first(io, "sensorInSep"),
|
|
sen_jam=first(io, "sensorJam"),
|
|
sen_finger=first(io, "sensorFinger"),
|
|
sen_carr_in_pos=first(io, "sensorCarrInPos"),
|
|
stopper=first(io, "stopperOutput", "MBxxxx"),
|
|
switch_out=first(io, "switchOutput", "MBxxxx"),
|
|
motor=first(io, "motorOutput", "MAxxxx"),
|
|
motor_protection=first(io, "motorProtection", "FCxxxx"),
|
|
))
|
|
|
|
fb_main = SCL_TEMPLATES["fb_main_frame"].format(
|
|
uh=uh, var_block="".join(var_block), call_block="".join(call_block))
|
|
|
|
sensor_calls = [SCL_TEMPLATES["call_sensor"].format(
|
|
bmk=s["id"], comment=s["comment"]) for s in skeleton["sensors"]]
|
|
fb_sensors = SCL_TEMPLATES["fb_sensors_frame"].format(
|
|
uh=uh, count=len(sensor_calls), call_block="".join(sensor_calls))
|
|
|
|
return fb_main, fb_sensors
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# main
|
|
# ----------------------------------------------------------------------------
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--uh", type=int, choices=range(1, 6),
|
|
help="nur diese Steuerung (1..5), sonst alle")
|
|
ap.add_argument("--no-scl", action="store_true",
|
|
help="nur JSON-Skeleton, kein SCL-Grundgeruest")
|
|
con = ap.add_mutually_exclusive_group()
|
|
con.add_argument("--con-ini", action="store_true",
|
|
help="Topologie aus data\\connect.ini uebernehmen")
|
|
con.add_argument("--con-json", action="store_true",
|
|
help="Topologie aus data\\connect.json uebernehmen")
|
|
args = ap.parse_args()
|
|
|
|
load_cfg()
|
|
load_scl_cfg()
|
|
root, data_dir, results = project_paths()
|
|
print(f"Daten: {data_dir}")
|
|
print(f"Results: {results}")
|
|
|
|
positions = read_positions(data_dir)
|
|
print(f"Positions-JSON: {len(positions)} Signale indiziert")
|
|
|
|
connect = None
|
|
if args.con_ini or args.con_json:
|
|
src = FILES["connect_ini"] if args.con_ini else FILES["connect_json"]
|
|
if not (data_dir / src).exists():
|
|
sys.exit(f"FEHLER: {data_dir / src} nicht gefunden.")
|
|
connect = (read_connect_ini if args.con_ini
|
|
else read_connect_json)(data_dir)
|
|
total = sum(len(p["connections"]) for p in connect.values())
|
|
print(f"Connect-Daten: {src} — {len(connect)} PLCs, "
|
|
f"{total} Verbindungen")
|
|
|
|
for n in ([args.uh] if args.uh else range(1, 6)):
|
|
uh = f"UH0{n}"
|
|
tags, constants = read_tia(data_dir, n)
|
|
skel = build_skeleton(n, tags, constants, positions, connect)
|
|
|
|
out_json = results / f"skeleton_{uh}.json"
|
|
out_json.write_text(json.dumps(skel, indent=2, ensure_ascii=False),
|
|
encoding="utf-8", newline="\n")
|
|
s = skel["statistics"]
|
|
topo_info = (f", {s['connections']} Verbindungen"
|
|
if "connections" in s else "")
|
|
print(f"{uh}: {s['tags']} Tags, {s['units']} Einheiten, "
|
|
f"{s['areas']} Bereiche{topo_info} -> {out_json.name}")
|
|
|
|
if not args.no_scl:
|
|
fb_main, fb_sensors = render_scl(skel, uh)
|
|
(results / f"FB_Main_skel_{uh}.scl").write_text(
|
|
fb_main, encoding="utf-8", newline="\n")
|
|
(results / f"FB_CallSensors_skel_{uh}.scl").write_text(
|
|
fb_sensors, encoding="utf-8", newline="\n")
|
|
print(f"{uh}: FB_Main_skel_{uh}.scl, FB_CallSensors_skel_{uh}.scl")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|