# -*- coding: utf-8 -*- # Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3" # ableiten und scheitert, wenn keine PythonCore-Installation registriert ist. """ validate_anlage.py - Ist die Anlagen-JSON bereit fuer die SCL? Eine Pruefung fuer beide Wege des Ablaufs (doc/Konzept_Layout_Hardware_SCL.md): direkt nach merge_hw_layout (keine Aenderungen) und nach den Handkorrekturen in BricsCAD (tro_annotate --read-tros/--read-jams, dann tro_flow --json und merge_hw_layout neu). scl_skeleton ruft sie vor dem Schreiben auf. Geprueft wird (FEHLER = so keine SCL, WARNUNG = SCL mit offenen Stellen): Struktur TRO-IDs eindeutig, jeder Typ im Katalog, Verbindungen und Staubereiche zeigen auf vorhandene TROs/Stationen, kein Separator in zwei TROs (FEHLER); leerer Separator-Slot, TRO ohne Verbindung, TRO vom Materialeingang aus nicht erreichbar (WARNUNG) Zuordnungen Binding auf einen unbekannten TRO/Slot, Status widerspruch, BMK fehlt in der Hardwareplanung (FEHLER); offene Pflichtrollen (WARNUNG, je Rolle gezaehlt) Aktualitaet Anlagen-JSON aelter als TRO-JSON, Hardwareplanung oder hw_overrides.ini (FEHLER: merge_hw_layout neu); TRO-JSON aelter als die BricsCAD-Korrekturen in tro_overrides.ini (WARNUNG: tro_flow --json neu) Korrekturen Eintraege in hw_overrides.ini / _jam_names.ini, die zu keinem TRO bzw. Staubereich passen - weder ueber die ID noch ueber den stabilen Schluessel (WARNUNG) Aufruf ueber bin/validate_anlage.bat bzw. bin/validate_anlage.sh: validate_anlage --json mubea_anlage.json [--hw mubea_hw.json] [--strict] Exit-Codes: 0 bereit (mit --strict: auch ohne Warnungen), 1 Eingabefehler, 5 Fehler gefunden (mit --strict auch bei Warnungen). """ from __future__ import annotations import argparse import collections import configparser import json import sys from dataclasses import dataclass, field from pathlib import Path from pydantic import ValidationError from hw_import import HardwareDocument from material_flow import env_dir from model import ANLAGE_SCHEMA, AnlageDocument, TroDocument from scl_context import JAM_NAMES_SUFFIX, jam_key, jam_stable_key, load_jam_overrides from tro_overrides import load_overrides, override_path FEHLER = "FEHLER" WARNUNG = "WARNUNG" # Name der Datei mit den Handkorrekturen der Hardware (merge_hw_layout) HW_OVERRIDES_FILE = "hw_overrides.ini" EXIT_FEHLER = 5 @dataclass class Befund: stufe: str bereich: str text: str def __str__(self) -> str: return f"{self.stufe:<7} {self.bereich:<11} {self.text}" @dataclass class Pruefung: """Ergebnis einer Pruefung, Befunde in der Reihenfolge der Bereiche.""" befunde: list[Befund] = field(default_factory=list) def fehler(self, bereich: str, text: str) -> None: self.befunde.append(Befund(FEHLER, bereich, text)) def warnung(self, bereich: str, text: str) -> None: self.befunde.append(Befund(WARNUNG, bereich, text)) @property def anzahl_fehler(self) -> int: return sum(1 for b in self.befunde if b.stufe == FEHLER) @property def anzahl_warnungen(self) -> int: return sum(1 for b in self.befunde if b.stufe == WARNUNG) def bereit(self, strict: bool = False) -> bool: return not self.anzahl_fehler and not (strict and self.anzahl_warnungen) # --------------------------------------------------------------------------- # Einzelpruefungen # --------------------------------------------------------------------------- def check_structure(doc: TroDocument, p: Pruefung) -> None: """IDs, Typen, Verbindungen, Staubereiche, Separatoren, Erreichbarkeit.""" ids = collections.Counter(t.tro_id for t in doc.tros + doc.boundaries) for tro_id, n in sorted(ids.items()): if n > 1: p.fehler("Struktur", f"TRO-ID {tro_id} kommt {n}x vor") for tro in doc.tros: if tro.definition() is None: p.fehler("Struktur", f"{tro.tro_id}: Typ '{tro.type_name}' steht nicht im Katalog") bekannt = set(ids) for c in doc.connections: for ende in (c.src, c.dst): if ende not in bekannt: p.fehler("Struktur", f"Verbindung {c.src} -> {c.dst}: {ende} unbekannt") for area in doc.jam_areas_or_derived(): if area.tro not in bekannt: p.fehler("Struktur", f"Staubereich {area.jam_id}: TRO {area.tro} unbekannt") besitzer: dict[str, str] = {} for tro in doc.tros: for sep in dict.fromkeys(tro.separators): if not sep: continue if sep in besitzer: p.fehler("Struktur", f"Separator {sep} steckt in {besitzer[sep]} und " f"{tro.tro_id}") besitzer.setdefault(sep, tro.tro_id) for slot in tro.slots: if slot.slot.startswith("separator") and not slot.layout_id: p.warnung("Struktur", f"{tro.tro_id}: {slot.slot} ohne Layout-Bauteil") # Erreichbarkeit vom Materialeingang (Systemgrenzen ohne Vorgaenger) nach: dict[str, set[str]] = collections.defaultdict(set) for c in doc.connections: nach[c.src].add(c.dst) for tro in doc.tros + doc.boundaries: nach[tro.tro_id].update(tro.successors) for tro in doc.tros: if not (tro.successors or tro.predecessors or tro.colocated): p.warnung("Struktur", f"{tro.tro_id}: ohne Verbindung") starts = [b.tro_id for b in doc.boundaries if not b.predecessors] if starts: gesehen, stapel = set(starts), list(starts) while stapel: for nxt in nach.get(stapel.pop(), ()): if nxt not in gesehen: gesehen.add(nxt) stapel.append(nxt) weg = [t.tro_id for t in doc.tros if t.tro_id not in gesehen and (t.successors or t.predecessors)] if weg: p.warnung("Struktur", f"vom Materialeingang nicht erreichbar: {', '.join(weg)}") def check_bindings(anlage: AnlageDocument, hw: HardwareDocument | None, p: Pruefung) -> None: """Zuordnungen der Hardwareplanung gegen Layout und TIA-Import.""" doc = anlage.layout slots = {(t.tro_id, s.slot) for t in doc.tros for s in t.slots} tros = {t.tro_id for t in doc.tros} geraete = hw.by_bmk() if hw is not None else None offen: collections.Counter[str] = collections.Counter() for b in anlage.bindings: wo = f"{b.tro}.{b.slot}.{b.role}" if b.tro else (b.layout_id or b.bmk) if b.tro and b.tro not in tros: p.fehler("Zuordnung", f"{wo}: TRO {b.tro} steht nicht im Layout") continue if b.tro and (b.tro, b.slot) not in slots: p.fehler("Zuordnung", f"{wo}: Slot {b.slot} gibt es an {b.tro} nicht") continue if b.status == "widerspruch": p.fehler("Zuordnung", f"{wo}: {b.bmk or '-'} - {b.note or 'Widerspruch'}") elif b.status == "offen": offen[b.role] += 1 if geraete is not None and b.bmk and b.status != "widerspruch" and b.bmk not in geraete: p.fehler("Zuordnung", f"{wo}: {b.bmk} fehlt in der Hardwareplanung") if offen: p.warnung("Zuordnung", "offene Pflichtrollen (TODO(E-Planung)): " + ", ".join(f"{r} {n}x" for r, n in sorted(offen.items()))) def _mtime(path: Path | None) -> float | None: return path.stat().st_mtime if path is not None and path.is_file() else None def _has_sections(path: Path, csv_name: str) -> bool: """Hat die INI einen Abschnitt [.*]?""" if not path.is_file(): return False prefix = f"[{csv_name.lower()}." try: return any(line.strip().lower().startswith(prefix) for line in path.read_text(encoding="utf-8-sig").splitlines()) except OSError: return False def check_freshness(anlage_file: Path, anlage: AnlageDocument | None, tro_file: Path | None, hw_file: Path | None, cfg_dir: Path, csv_name: str, p: Pruefung) -> None: """Ist die JSON juenger als alles, woraus sie gebaut wird?""" stand = _mtime(anlage_file) if stand is None: return if anlage is not None: hw_ini = cfg_dir / HW_OVERRIDES_FILE # hw_overrides.ini zaehlt nur mit Abschnitten zu dieser Anlage if not _has_sections(hw_ini, csv_name): hw_ini = None for quelle, was in ((tro_file, "TRO-JSON"), (hw_file, "Hardwareplanung"), (hw_ini, HW_OVERRIDES_FILE)): t = _mtime(quelle) if t is not None and t > stand: p.fehler("Stand", f"{anlage_file.name} ist aelter als {was} ({quelle.name}) " f"- merge_hw_layout neu laufen lassen") # BricsCAD-Korrekturen wirken in tro_flow - die TRO-JSON muss juenger sein layout_stand = _mtime(tro_file) if anlage is not None else stand korrektur = override_path(cfg_dir) t = _mtime(korrektur) if layout_stand is not None and t is not None and t > layout_stand: try: if load_overrides(csv_name, cfg_dir).edits: p.warnung("Stand", f"{korrektur.name} (Handkorrekturen aus BricsCAD) ist " f"juenger als die TRO-JSON - tro_flow --json neu laufen lassen") except (OSError, configparser.Error): pass def check_corrections(doc: TroDocument, cfg_dir: Path, csv_name: str, p: Pruefung) -> None: """Handkorrekturen, die ins Leere zeigen (z. B. nach einer Umbenennung).""" hw_ini = cfg_dir / HW_OVERRIDES_FILE if hw_ini.is_file(): parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"), interpolation=None) parser.optionxform = str try: with hw_ini.open(encoding="utf-8-sig") as fh: parser.read_file(fh) except (OSError, configparser.Error) as exc: p.fehler("Korrektur", f"{hw_ini.name} nicht lesbar: {exc}") else: section = f"{csv_name}.bindings" for key in (parser[section] if parser.has_section(section) else {}): teile = key.split(".") tro = doc.resolve(teile[0]) if len(teile) == 3 else None if tro is None or not any(s.slot == teile[1] for s in tro.slots): p.warnung("Korrektur", f"{hw_ini.name} [{section}] {key}: kein TRO/Slot " f"dazu (ID und Schluessel geprueft)") jam_ini = cfg_dir / f"{Path(csv_name).stem}{JAM_NAMES_SUFFIX}" try: jam = load_jam_overrides(jam_ini) except (OSError, ValueError, configparser.Error) as exc: p.fehler("Korrektur", f"{jam_ini.name} nicht lesbar: {exc}") return if jam: areas = doc.jam_areas_or_derived() treffer = {k for a in areas for k in (jam_stable_key(a, doc), jam_key(a))} for key in sorted(set(jam) - treffer): p.warnung("Korrektur", f"{jam_ini.name} {key}: kein Staubereich dazu") # --------------------------------------------------------------------------- # Gesamt # --------------------------------------------------------------------------- def validate(json_file: Path, *, hw_file: Path | None = None, results: Path | None = None, cfg_dir: Path | None = None) -> Pruefung: """TRO- oder Anlagen-JSON pruefen. Ladefehler werden zu einem FEHLER.""" results = results or json_file.parent cfg_dir = cfg_dir or env_dir("SKEL_CFG", "cfg") p = Pruefung() try: text = json_file.read_text(encoding="utf-8") if json.loads(text).get("schema") == ANLAGE_SCHEMA: anlage: AnlageDocument | None = AnlageDocument.from_json(text) doc = anlage.layout else: anlage, doc = None, TroDocument.from_json(text) except (OSError, ValueError, ValidationError) as exc: p.fehler("Modell", f"{json_file.name} nicht lesbar oder passt nicht zum Modell: " f"{str(exc).splitlines()[0]}") return p csv_name = Path(doc.source).name check_structure(doc, p) hw = None tro_file = None if anlage is not None: tro_file = results / Path(anlage.layout_file).name if anlage.layout_file else None if hw_file is None and anlage.hw_file: hw_file = results / Path(anlage.hw_file).name if hw_file is not None and hw_file.is_file(): try: hw = HardwareDocument.from_json(hw_file.read_text(encoding="utf-8")) except (OSError, ValidationError) as exc: p.fehler("Zuordnung", f"{hw_file.name} nicht lesbar: {str(exc).splitlines()[0]}") elif hw_file is not None: p.warnung("Zuordnung", f"{hw_file.name} nicht gefunden - BMK nicht gegengeprueft") check_bindings(anlage, hw, p) check_freshness(json_file, anlage, tro_file, hw_file, cfg_dir, csv_name, p) check_corrections(doc, cfg_dir, csv_name, p) return p def report(json_file: Path, p: Pruefung, strict: bool, out=sys.stdout) -> None: print("", file=out) print("================================================================", file=out) print("PRUEFUNG ANLAGEN-JSON", file=out) print("================================================================", file=out) print(f"Datei = {json_file}", file=out) print(f"Fehler = {p.anzahl_fehler}", file=out) print(f"Warnungen = {p.anzahl_warnungen}", file=out) print(f"Ergebnis = {'bereit fuer die SCL' if p.bereit(strict) else 'NICHT bereit'}" + (" (--strict: Warnungen zaehlen)" if strict else ""), file=out) print("================================================================", file=out) for b in p.befunde: print(f" {b}", file=out) print("", file=out) def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="validate_anlage", description="Prueft eine Anlagen-JSON (oder TRO-JSON), bevor daraus SCL wird.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" " validate_anlage.bat --json mubea_anlage.json\n" " validate_anlage.bat --json dematic_merge_anlage.json --strict\n" "\nExit-Codes: 0 bereit, 1 Aufruffehler, 5 nicht bereit\n", ) parser.add_argument("--json", required=True, metavar="NAME", help="Anlagen-JSON (_anlage.json) oder TRO-JSON, in " "%%SKEL_RESULTS%% oder ein Pfad.") parser.add_argument("--hw", metavar="NAME", help="Hardwareplanung (_hw.json). Standard: hwFile " "der Anlagen-JSON.") parser.add_argument("--strict", action="store_true", help="Auch Warnungen machen die JSON 'nicht bereit'.") return parser.parse_args(argv) def _in_results(name: str, results: Path) -> Path: path = Path(name) return results / name if not path.is_absolute() and path.parent == Path(".") else path def main(argv: list[str] | None = None) -> int: args = parse_args(argv) results = env_dir("SKEL_RESULTS", "results") json_file = _in_results(args.json, results) if not json_file.is_file(): print(f"FEHLER: JSON nicht gefunden: {json_file}", file=sys.stderr) return 1 hw_file = _in_results(args.hw, results) if args.hw else None p = validate(json_file, hw_file=hw_file, results=json_file.parent) report(json_file, p, args.strict) return 0 if p.bereit(args.strict) else EXIT_FEHLER if __name__ == "__main__": sys.exit(main())