erste Fassung eines Python Skripts zur Skeleton Erzeugung über ein json Zwischenformat geschrieben
This commit is contained in:
@@ -0,0 +1,610 @@
|
||||
# -*- 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 optional: create_skel.cfg ueberschreibt die globalen
|
||||
Definitionen (Templates, Vokabular)
|
||||
|
||||
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
|
||||
|
||||
Aufruf: bin\\create_skel.bat [--uh 1..5] [--no-scl]
|
||||
"""
|
||||
|
||||
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
|
||||
# Alle folgenden Bloecke sind bewusst als Modul-Konstanten definiert, damit
|
||||
# sie spaeter unveraendert in eine .cfg (PV_CFG\create_skel.cfg) ausgelagert
|
||||
# werden koennen. load_cfg() ueberschreibt sie, wenn die Datei existiert.
|
||||
# ============================================================================
|
||||
|
||||
#: 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",
|
||||
}
|
||||
|
||||
#: 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.
|
||||
# ----------------------------------------------------------------------------
|
||||
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
|
||||
# ============================================================================
|
||||
|
||||
|
||||
def load_cfg():
|
||||
"""Ueberschreibt die globalen Definitionen aus PV_CFG\\create_skel.cfg.
|
||||
|
||||
Unterstuetzte Sektionen: [files], [verw_roles], [scl_templates].
|
||||
Werte in [scl_templates] sind mehrzeilige Raw-Strings.
|
||||
"""
|
||||
cfg_dir = os.environ.get("PV_CFG")
|
||||
if not cfg_dir:
|
||||
return
|
||||
cfg_file = Path(cfg_dir) / "create_skel.cfg"
|
||||
if not cfg_file.exists():
|
||||
return
|
||||
print(f"Lade Konfiguration: {cfg_file}")
|
||||
cp = configparser.ConfigParser(interpolation=None)
|
||||
cp.read(cfg_file, encoding="utf-8")
|
||||
if cp.has_section("files"):
|
||||
FILES.update(cp["files"])
|
||||
if cp.has_section("verw_roles"):
|
||||
VERW_ROLES.update(cp["verw_roles"])
|
||||
if cp.has_section("scl_templates"):
|
||||
for key, val in cp["scl_templates"].items():
|
||||
SCL_TEMPLATES[key] = val.replace("\\n", "\n")
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# 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):
|
||||
"""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())]
|
||||
|
||||
return {
|
||||
"$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>)",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# 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")
|
||||
args = ap.parse_args()
|
||||
|
||||
load_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")
|
||||
|
||||
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)
|
||||
|
||||
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"]
|
||||
print(f"{uh}: {s['tags']} Tags, {s['units']} Einheiten, "
|
||||
f"{s['areas']} Bereiche -> {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()
|
||||
Reference in New Issue
Block a user