#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ vf_spec_export.py - Eingabe-Journale eines VF-Linienzugs in eine SPEC uebersetzen und zurueck. Eine Spec beschreibt eine Kette in Domaenenwerten ("links", "winkel", "aussen", ja/nein) statt in Menue-Codes und Journal-Tokens: lesbar, schreibbar und damit die Eingabeform fuer den nicht-interaktiven Bau (Lisp/vf_spec.lsp, siehe doc/TODO-plan-vf-interactive.md Stufe 1). Zwei Richtungen, die einander pruefen: spec_aus_journal(tokens) -> spec (Leser) journal_aus_spec(spec) -> tokens (Schreiber) Fuer jede echte Kette muss gelten: journal_aus_spec(spec_aus_journal(tok)) == tok Genau dieser Rundlauf laeuft in tests/test_vf_spec.py gegen die 5 echten HundM-Ketten. Er beweist, dass die Spec das Journal VERLUSTFREI abbildet - sonst waere jeder spec-gebaute Nachbau eine andere Kette als das Original. Der Schreiber ist das Vorbild fuer vfl-spec->journal in Lisp/vf_spec.lsp; beide muessen dieselbe Tokenfolge liefern. Die Grammatik selbst (welche Frage in welcher Reihenfolge) steckt in lib/vf_journal_export.py (Journal.dekodiere) und ist dort kommentiert. Aufruf (erzeugt die Spec-Testdaten aus dem Protokoll der echten Ketten): python lib/vf_spec_export.py tests/testdata/hm_recformat.json \ tests/testdata/hundm05.json """ import argparse import collections import json import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from vf_journal_export import arrays_einzeilig # noqa: E402 PREFIX = "[vf_spec_export]" # ============================================================ # Code-Tabellen: Menue-Antwort <-> Domaenenwert # ============================================================ # Links der Journal-Code, rechts der Wert in der Spec. Die Spec kennt KEINE # Codes - sie sind eine Eigenschaft des Dialogs, nicht der Anlage. Nur die # AS-/ES-Winkel und die Bogen-/Kurvenwinkel stehen im Journal als Wert und # bleiben darum Werte. SEITE = {"1": "links", "2": "rechts"} VARIANTE = {"1": "aussen", "2": "innen"} GEFAELLE = {"1": "hoehe", "2": "winkel"} VERTEILUNG = {"1": "haelfte", "2": "einlauf"} IM_VF = {"1": "horizontal", "2": "vario-kurve", "3": "auf-ab"} ENDPUNKT = {"1": "motorstation", "2": "weiter", "3": "kettenende"} ENDE3 = {"1": "ja-mit-es", "2": "ja-ohne-es", "3": "nein"} ENDE2 = {"1": "ja-mit-es", "2": "nein"} # Glied-Typ -> Menue-Code, getrennt nach Position in der Kette. Der Code # haengt am Frame: die ERSTE Sektion laeuft ohne Frame und bekommt 4 Optionen # (kein GF-Bogen - es gibt noch keine Richtung, an die er anschliessen # koennte), jede spaetere 5. Siehe vf-linienzug-modus. MENU_START = {"Linie-GF": "1", "Linie-VF": "2", "Horizontal-VF": "3", "Linie": "4"} MENU_WEITER = {"GF-Bogen": "1", "Linie-GF": "2", "Linie-VF": "3", "Horizontal-VF": "4", "Linie": "5"} GLIEDER = sorted(set(MENU_WEITER) | {"ES"}) # Glieder, die eine VF-Einheit eroeffnen, und ob ihr erster Koerper # horizontal ist (winkel1 = 0 in vfl-vf-einheit: nur dann laufen die # Separator-/Endpunkt-Fragen VOR der Laengenberechnung). VF_GLIEDER = {"Horizontal-VF": True, "Linie-VF": False, "Linie": False} def _um(tabelle, code, was): if code not in tabelle: raise ValueError("%s: unbekannter Code %r (erlaubt: %s)" % (was, code, sorted(tabelle))) return tabelle[code] def _code(tabelle, wert, was): fuer = dict((v, k) for k, v in tabelle.items()) if wert not in fuer: raise ValueError("%s: unbekannter Wert %r (erlaubt: %s)" % (was, wert, sorted(fuer))) return fuer[wert] def _ja(code, was): return _um({"1": True, "2": False}, code, was) def _ja_code(wert): return "1" if wert else "2" # ============================================================ # Tokens # ============================================================ # Token = (Art, Wert-String) genau wie im Journal auf der XDATA. Arten: # PT Startpunkt "x,y,z" DL Segmentlaenge # REAL Realzahl INT Ganzzahl # STR Menue-/Wertantwort STEP Glied-Marker (Label = Segmenttyp) def tokens_aus_protokoll(eintraege): """Das Frage-Antwort-Protokoll (Schema tests/testdata/hm_recformat.json) zurueck in Journal-Tokens uebersetzen - die Umkehrung von vf_journal_export.Journal.add.""" tok = [] for e in eintraege: typ = e["typ"] if typ == "point_abs": tok.append(("PT", ",".join(_zahl(v) for v in e["wert"]))) elif typ == "point_rel": tok.append(("DL", _zahl(e["dL"]))) if "hz" in e: tok.append(("REAL", _zahl(e["hz"]))) elif typ == "real": tok.append(("REAL", _zahl(e["wert"]))) elif typ == "int": tok.append(("INT", str(int(e["wert"])))) elif typ == "string": tok.append(("STR", str(e["wert"]))) elif typ == "step": tok.append(("STEP", str(e["wert"]))) else: raise ValueError("unbekannter Eintragstyp: %r" % typ) return tok def _zahl(v): """Zahl so schreiben, wie sie im Journal steht (float ohne Exponent).""" if isinstance(v, bool): raise ValueError("Wahrheitswert als Zahl: %r" % v) f = float(v) if f == int(f) and abs(f) < 1e15: return "%.1f" % f return repr(f) def gleich(a, b): """Zwei Tokenfolgen vergleichen - Zahlen numerisch, nicht als Text (0.0 und -0.0, 3 und 3.0 sind dasselbe Journal).""" if len(a) != len(b): return False for (art1, w1), (art2, w2) in zip(a, b): if art1 != art2: return False if art1 in ("REAL", "DL", "INT"): if abs(float(w1) - float(w2)) > 1e-6: return False elif art1 == "PT": p1 = [float(x) for x in w1.split(",")] p2 = [float(x) for x in w2.split(",")] if len(p1) != len(p2) or any(abs(x - y) > 1e-6 for x, y in zip(p1, p2)): return False elif w1 != w2: return False return True def erste_abweichung(a, b): """Index und Beschreibung der ersten Abweichung (fuer Fehlermeldungen).""" for i in range(max(len(a), len(b))): x = a[i] if i < len(a) else None y = b[i] if i < len(b) else None if not gleich([x] if x else [], [y] if y else []): return i, "ist %r, erwartet %r" % (y, x) return None, "" # ============================================================ # Leser: Journal -> Spec # ============================================================ class SpecLeser(object): """Tokens in eine Spec uebersetzen. Folgt derselben Frage-Reihenfolge wie Journal.dekodiere in lib/vf_journal_export.py, legt aber Domaenenwerte ab statt Protokollzeilen.""" def __init__(self, tok): self.tok = list(tok) self.pos = 0 self.erster_step = True self.erster_dl = True # --- Tokenzugriff --- def art(self): return self.tok[self.pos][0] if self.pos < len(self.tok) else None def label(self): return self.tok[self.pos][1] if self.pos < len(self.tok) else None def nimm(self, erwartet): if self.pos >= len(self.tok): raise ValueError("Journal endet vor %s" % erwartet) art, wert = self.tok[self.pos] if art != erwartet: raise ValueError("Token %d: %s erwartet, %s:%s gefunden" % (self.pos, erwartet, art, wert)) self.pos += 1 return wert def real(self): return float(self.nimm("REAL")) def ganz(self): return int(self.nimm("INT")) def opt_winkel_idx(self, ziel): """Die Winkelwahl (vfl-waehle-winkel) steht nur im Journal, wenn mehrere Kandidatenwinkel gueltig waren - sonst faellt die Frage weg.""" if self.art() == "INT": ziel["winkel_idx"] = self.ganz() def linie(self, ziel): """Segmentlaenge, und NUR beim allerersten Segment der Kette die gesnappte Fahrtrichtung (vfl-in-abstand journalisiert sie nur bei freier Richtungswahl).""" ziel["dl"] = float(self.nimm("DL")) if self.erster_dl and self.art() == "REAL": ziel["hz"] = self.real() self.erster_dl = False # --- Ablauf --- def lies(self): spec = collections.OrderedDict() spec["start"] = self.praeambel() spec["sektionen"] = self.sektionen() if self.pos != len(self.tok): raise ValueError("nicht alle Tokens verbraucht (%d von %d)" % (self.pos, len(self.tok))) return spec def praeambel(self): start = collections.OrderedDict() start["punkt"] = [float(v) for v in self.nimm("PT").split(",")] start["hoehe"] = self.real() start["as"] = _ja(self.nimm("STR"), "AS-Element setzen") if start["as"]: start["as_winkel"] = self.nimm("STR") # Wert, kein Code start["as_seite"] = _um(SEITE, self.nimm("STR"), "AS-Seite") return start def sektionen(self): raus = [] while self.pos < len(self.tok): label = self.nimm("STEP") if label == "ES": raus.append(self.es_glied()) continue if label == "Vario-Kurve": raise ValueError("Vario-Kurve-Marker ausserhalb einer " "VF-Einheit (Token %d)" % (self.pos - 1)) sek = collections.OrderedDict() sek["glied"] = label tabelle = MENU_START if self.erster_step else MENU_WEITER code = self.nimm("STR") soll = tabelle.get(label) if soll is None: raise ValueError("Glied %r ist an dieser Stelle nicht " "waehlbar (Token %d)" % (label, self.pos - 1)) if code != soll: raise ValueError("Glied %r erwartet Menue-Code %r, Journal " "hat %r (Token %d)" % (label, soll, code, self.pos - 1)) self.erster_step = False self.glied(label, sek) raus.append(sek) return raus def es_glied(self): sek = collections.OrderedDict() sek["glied"] = "ES" sek["winkel"] = self.nimm("STR") # Wert, kein Code sek["seite"] = _um(SEITE, self.nimm("STR"), "ES-Seite") return sek def glied(self, label, sek): if label == "GF-Bogen": sek["winkel"] = self.ganz() sek["seite"] = _um(SEITE, self.nimm("STR"), "GF-Bogen-Seite") elif label == "Linie-GF": self.linie(sek) sek["gefaelle"] = _um(GEFAELLE, self.nimm("STR"), "Gefaelle") if sek["gefaelle"] == "winkel": sek["winkel"] = self.real() else: sek["hoehe"] = self.real() sek["ende"] = _um(ENDE3, self.nimm("STR"), "Kettenende") elif label == "Linie-VF": self.linie(sek) sek["hoehe"] = self.real() self.opt_winkel_idx(sek) sek["vf"] = self.vf_einheit(auto_ende=False, erst_hor=False) elif label == "Horizontal-VF": self.linie(sek) sek["vf"] = self.vf_einheit(auto_ende=False, erst_hor=True) elif label == "Linie": self.linie(sek) sek["hoehe"] = self.real() self.opt_winkel_idx(sek) # vfl-segment-entscheidung waehlt selbst zwischen reiner # Gefaellestrecke und VF-Einheit. Am Journal erkennbar: eine # VF-Einheit beginnt mit der GF-Verteilungsfrage (STR), eine # Gefaellestrecke geht direkt zum ES-Glied (STEP) oder endet. if self.art() == "STR": sek["segment_typ"] = "VF" sek["vf"] = self.vf_einheit(auto_ende=True, erst_hor=False) else: sek["segment_typ"] = "GF" else: raise ValueError("unbekanntes Glied: %r" % label) def vf_einheit(self, auto_ende, erst_hor): vf = collections.OrderedDict() vf["gf_verteilung"] = _um(VERTEILUNG, self.nimm("STR"), "GF-Verteilung") if erst_hor: # Erster Koerper horizontal: vfl-baue-horizontal-koerper stellt # die Separator-Fragen VOR der Laengenberechnung. Die # Endpunkt-Antwort danach gehoert schon zum naechsten Knoten und # wird darum unten in der Schleife gelesen. vf["sep_vor"] = _ja(self.nimm("STR"), "Separator vor") vf["sep_nach"] = _ja(self.nimm("STR"), "Separator nach") vf["subs"] = [] ziel_ende = False es_gewuenscht = True while True: was = _um(ENDPUNKT, self.nimm("STR"), "Endpunkt des Foerderers") if was == "motorstation": vf["subs"].append(collections.OrderedDict( [("sub", "motorstation")])) break if was == "kettenende": sub = collections.OrderedDict([("sub", "kettenende")]) sub["es"] = _ja(self.nimm("STR"), "ES-Element setzen") self.linie(sub) sub["hoehe"] = self.real() self.opt_winkel_idx(sub) vf["subs"].append(sub) ziel_ende = True es_gewuenscht = sub["es"] break wahl = _um(IM_VF, self.nimm("STR"), "Naechstes in der VF-Einheit") sub = collections.OrderedDict([("sub", wahl)]) if wahl == "horizontal": self.linie(sub) sub["sep_vor"] = _ja(self.nimm("STR"), "Separator vor") sub["sep_nach"] = _ja(self.nimm("STR"), "Separator nach") elif wahl == "vario-kurve": marke = self.nimm("STEP") if marke != "Vario-Kurve": raise ValueError("Vario-Kurve-Marker erwartet, %r " "gefunden" % marke) sub["winkel"] = self.ganz() sub["seite"] = _um(SEITE, self.nimm("STR"), "Vario-Kurve-Seite") sub["variante"] = _um(VARIANTE, self.nimm("STR"), "Vario-Kurve-Variante") else: # auf-ab self.linie(sub) sub["hoehe"] = self.real() self.opt_winkel_idx(sub) vf["subs"].append(sub) # --- Abschluss (vfl-vf-einheit-abschluss) --- # Ohne ES-Wunsch am Zielpunkt endet die Kette dort direkt; bei # auto_ende (Glied "Linie") oder erreichtem Zielpunkt folgt das # ES-Glied ohne Frage. Sonst wird gefragt. if ziel_ende and not es_gewuenscht: vf["ende"] = "zielpunkt-ohne-es" elif auto_ende or ziel_ende: vf["ende"] = "automatisch" else: vf["ende"] = _um(ENDE2, self.nimm("STR"), "Ist das Kettenende") if vf["ende"] != "ja-mit-es": vf["separator"] = _ja(self.nimm("STR"), "Separator an dieser Stelle") return vf def spec_aus_journal(tok): return SpecLeser(tok).lies() # ============================================================ # Schreiber: Spec -> Journal # ============================================================ # Vorbild fuer vfl-spec->journal in Lisp/vf_spec.lsp. Bewusst als reiner # Emitter geschrieben: keine Geometrie, keine Rueckfrage, nur Felder in # Tokens. Was die Spec nicht hergibt, ist ein Spec-Fehler - nichts wird # geraten (siehe Invariante I3 im Fahrplan). class SpecSchreiber(object): def __init__(self, spec): self.spec = spec self.tok = [] self.erster_dl = True self.wo = "Spec" # --- Ausgabe --- def raus(self, art, wert): self.tok.append((art, wert)) def zahl(self, art, wert): self.raus(art, _zahl(wert)) def feld(self, quelle, name, pflicht=True): if name in quelle: return quelle[name] if pflicht: raise ValueError("%s: Feld %r fehlt" % (self.wo, name)) return None def linie(self, quelle): self.zahl("DL", self.feld(quelle, "dl")) if self.erster_dl: # Beim ersten Segment MUSS die Richtung dabeistehen (die Kette hat # noch keine geerbte), bei jedem spaeteren darf sie es NICHT - # sonst verschiebt sich die ganze Queue um einen Eintrag. self.zahl("REAL", self.feld(quelle, "hz")) self.erster_dl = False elif "hz" in quelle: raise ValueError("%s: hz steht nur beim ERSTEN Segment der Kette " "im Journal (vfl-in-abstand)" % self.wo) def winkel_idx(self, quelle): if "winkel_idx" in quelle: self.raus("INT", str(int(quelle["winkel_idx"]))) # --- Ablauf --- def schreib(self): self.praeambel() sektionen = self.spec.get("sektionen") or [] for nr, sek in enumerate(sektionen, 1): self.wo = "Sektion %d (%s)" % (nr, sek.get("glied")) self.sektion(sek, erster=(nr == 1)) return self.tok def praeambel(self): start = self.spec.get("start") if not start: raise ValueError("Spec: Abschnitt \"start\" fehlt") self.wo = "start" punkt = self.feld(start, "punkt") if len(punkt) != 3: raise ValueError("start: punkt braucht x, y, z") self.raus("PT", ",".join(_zahl(v) for v in punkt)) self.zahl("REAL", self.feld(start, "hoehe")) gesetzt = bool(self.feld(start, "as")) self.raus("STR", _ja_code(gesetzt)) if gesetzt: self.raus("STR", str(self.feld(start, "as_winkel"))) self.raus("STR", _code(SEITE, self.feld(start, "as_seite"), "as_seite")) def sektion(self, sek, erster): label = self.feld(sek, "glied") if label == "ES": self.es_glied(sek) return tabelle = MENU_START if erster else MENU_WEITER if label not in tabelle: raise ValueError("%s: Glied %r ist an dieser Stelle nicht " "waehlbar (erlaubt: %s)" % (self.wo, label, sorted(tabelle))) self.raus("STEP", label) self.raus("STR", tabelle[label]) if label == "GF-Bogen": self.raus("INT", str(int(self.feld(sek, "winkel")))) self.raus("STR", _code(SEITE, self.feld(sek, "seite"), "seite")) elif label == "Linie-GF": self.linie(sek) art = self.feld(sek, "gefaelle") self.raus("STR", _code(GEFAELLE, art, "gefaelle")) self.zahl("REAL", self.feld(sek, "winkel" if art == "winkel" else "hoehe")) self.raus("STR", _code(ENDE3, self.feld(sek, "ende"), "ende")) elif label == "Linie-VF": self.linie(sek) self.zahl("REAL", self.feld(sek, "hoehe")) self.winkel_idx(sek) self.vf_einheit(sek, erst_hor=False) elif label == "Horizontal-VF": self.linie(sek) self.vf_einheit(sek, erst_hor=True) elif label == "Linie": self.linie(sek) self.zahl("REAL", self.feld(sek, "hoehe")) self.winkel_idx(sek) if self.feld(sek, "segment_typ") == "VF": self.vf_einheit(sek, erst_hor=False) else: raise ValueError("%s: unbekanntes Glied %r" % (self.wo, label)) def es_glied(self, sek): self.raus("STEP", "ES") self.raus("STR", str(self.feld(sek, "winkel"))) self.raus("STR", _code(SEITE, self.feld(sek, "seite"), "seite")) def vf_einheit(self, sek, erst_hor): vf = self.feld(sek, "vf") self.raus("STR", _code(VERTEILUNG, self.feld(vf, "gf_verteilung"), "gf_verteilung")) if erst_hor: self.raus("STR", _ja_code(self.feld(vf, "sep_vor"))) self.raus("STR", _ja_code(self.feld(vf, "sep_nach"))) subs = vf.get("subs") or [] if not subs: raise ValueError("%s: VF-Einheit ohne subs - sie muss mit " "motorstation oder kettenende enden" % self.wo) for i, sub in enumerate(subs, 1): wahl = self.feld(sub, "sub") letzter = (i == len(subs)) if wahl in ("motorstation", "kettenende") and not letzter: raise ValueError("%s: sub %r beendet die VF-Einheit, es " "folgen aber noch %d weitere" % (self.wo, wahl, len(subs) - i)) if wahl == "motorstation": self.raus("STR", _code(ENDPUNKT, "motorstation", "sub")) elif wahl == "kettenende": self.raus("STR", _code(ENDPUNKT, "kettenende", "sub")) self.raus("STR", _ja_code(self.feld(sub, "es"))) self.linie(sub) self.zahl("REAL", self.feld(sub, "hoehe")) self.winkel_idx(sub) else: if letzter: raise ValueError("%s: VF-Einheit endet mit sub %r - der " "letzte sub muss motorstation oder " "kettenende sein" % (self.wo, wahl)) self.raus("STR", _code(ENDPUNKT, "weiter", "sub")) self.raus("STR", _code(IM_VF, wahl, "sub")) if wahl == "horizontal": self.linie(sub) self.raus("STR", _ja_code(self.feld(sub, "sep_vor"))) self.raus("STR", _ja_code(self.feld(sub, "sep_nach"))) elif wahl == "vario-kurve": self.raus("STEP", "Vario-Kurve") self.raus("INT", str(int(self.feld(sub, "winkel")))) self.raus("STR", _code(SEITE, self.feld(sub, "seite"), "seite")) self.raus("STR", _code(VARIANTE, self.feld(sub, "variante"), "variante")) else: # auf-ab self.linie(sub) self.zahl("REAL", self.feld(sub, "hoehe")) self.winkel_idx(sub) ende = self.feld(vf, "ende") if ende in ("zielpunkt-ohne-es", "automatisch"): return # keine Frage self.raus("STR", _code(ENDE2, ende, "vf.ende")) if ende != "ja-mit-es": self.raus("STR", _ja_code(self.feld(vf, "separator"))) def journal_aus_spec(spec): return SpecSchreiber(spec).schreib() # ============================================================ # Flaches JSON (vom LISP-Parser lesbar) # ============================================================ # ssg-load-json/ssg-parse-json-array liest ZEILENWEISE und FLACH: jedes "{" # beginnt ein Objekt, Verschachtelung ist unmoeglich. Also dieselbe flache # Form wie die Aufzeichnung: ein Kopf-Objekt ("spec_id"), dann Sektionen # ("glied") und deren Sub-Knoten ("sub") in Reihenfolge. def flache_objekte(spec_id, spec, kopf_extra=None): objekte = [] kopf = collections.OrderedDict() kopf["spec_id"] = spec_id kopf["modus"] = spec.get("modus", 1) start = spec["start"] kopf["start_punkt"] = start["punkt"] kopf["start_hoehe"] = start["hoehe"] kopf["as"] = 1 if start["as"] else 0 if start["as"]: kopf["as_winkel"] = start["as_winkel"] kopf["as_seite"] = start["as_seite"] for k, v in (kopf_extra or collections.OrderedDict()).items(): kopf[k] = v objekte.append(kopf) for nr, sek in enumerate(spec["sektionen"], 1): obj = collections.OrderedDict() obj["glied"] = sek["glied"] obj["nr"] = nr for k, v in sek.items(): if k in ("glied", "vf"): continue obj[k] = 1 if v is True else (0 if v is False else v) vf = sek.get("vf") if vf: for k in ("gf_verteilung", "sep_vor", "sep_nach", "ende", "separator"): if k in vf: v = vf[k] obj["vf_" + k if k != "gf_verteilung" else k] = ( 1 if v is True else (0 if v is False else v)) objekte.append(obj) for sub in (vf or {}).get("subs", []): s = collections.OrderedDict() s["sub"] = sub["sub"] for k, v in sub.items(): if k == "sub": continue s[k] = 1 if v is True else (0 if v is False else v) objekte.append(s) return objekte def fremde_objekte(pfad): """Objekte einer bestehenden Ziel-Datei, die KEINE Spec beschreiben. Die Spec-Testdaten werden erzeugt, die Datei ist aber zugleich die Stelle, an der weitere Objekte der Anlage von Hand nachgetragen werden (Kreisel, Foerderer, ...). Alles ohne spec_id/glied/sub gehoert nicht zum Generator und wird beim Neuschreiben uebernommen - sonst waere ein handgeschriebener Eintrag beim naechsten Lauf weg. """ if not pfad or not os.path.isfile(pfad): return [] try: with open(pfad, encoding="utf-8") as fh: daten = json.load(fh) except ValueError: return [] return [o for o in daten if not ("spec_id" in o or "glied" in o or "sub" in o)] def spec_aus_flachen_objekten(objekte): """Umkehrung von flache_objekte - dieselbe Gruppierregel wie hmrec:gruppiere in LISP (Kopf, dann glied/sub in Reihenfolge).""" specs = [] for obj in objekte: if "spec_id" in obj: start = collections.OrderedDict() start["punkt"] = obj["start_punkt"] start["hoehe"] = obj["start_hoehe"] start["as"] = bool(obj.get("as")) if start["as"]: start["as_winkel"] = obj["as_winkel"] start["as_seite"] = obj["as_seite"] specs.append((obj, collections.OrderedDict( [("modus", obj.get("modus", 1)), ("start", start), ("sektionen", [])]))) elif "glied" in obj: sek = collections.OrderedDict([("glied", obj["glied"])]) vf = collections.OrderedDict() for k, v in obj.items(): if k in ("glied", "nr"): continue if k == "gf_verteilung": vf["gf_verteilung"] = v elif k.startswith("vf_"): name = k[3:] vf[name] = bool(v) if name in ("sep_vor", "sep_nach", "separator") else v elif k in ("as", "es", "sep_vor", "sep_nach"): sek[k] = bool(v) else: sek[k] = v if vf: vf["subs"] = [] sek["vf"] = vf specs[-1][1]["sektionen"].append(sek) elif "sub" in obj: sub = collections.OrderedDict([("sub", obj["sub"])]) for k, v in obj.items(): if k == "sub": continue sub[k] = bool(v) if k in ("es", "sep_vor", "sep_nach") else v sek = specs[-1][1]["sektionen"][-1] sek.setdefault("vf", collections.OrderedDict()) sek["vf"].setdefault("subs", []).append(sub) else: # Von Hand ergaenzte Objekte der Anlage (Kreisel u.a.) gehoeren # nicht zur Spec und werden uebersprungen - genauso wie # vsp-gruppieren in Lisp/vf_spec.lsp es tut. Sie stehen in # derselben Datei, damit eine Anlage EINE Beschreibung hat. continue return specs # ============================================================ # CLI # ============================================================ def main(): p = argparse.ArgumentParser( description="Frage-Antwort-Protokoll (hm_recformat.json) in Specs " "uebersetzen und den Rundlauf pruefen.") p.add_argument("protokoll", help="Eingabe (Schema hm_recformat.json)") p.add_argument("ziel", help="Ausgabe (flaches Spec-JSON)") args = p.parse_args() with open(args.protokoll, encoding="utf-8") as fh: daten = json.load(fh) ketten = [] for obj in daten: if "test_id" in obj: ketten.append((obj, [])) elif "typ" in obj and ketten: ketten[-1][1].append(obj) print("%s %d Ketten in %s" % (PREFIX, len(ketten), args.protokoll)) alle = [] fehler = 0 for kopf, eintraege in ketten: tid = kopf["test_id"] tok = tokens_aus_protokoll(eintraege) try: spec = spec_aus_journal(tok) zurueck = journal_aus_spec(spec) except ValueError as e: print("%s %-20s FEHLER: %s" % (PREFIX, tid, e)) fehler += 1 continue if not gleich(tok, zurueck): idx, was = erste_abweichung(tok, zurueck) print("%s %-20s RUNDLAUF WEICHT AB bei Token %s: %s" % (PREFIX, tid, idx, was)) fehler += 1 continue extra = collections.OrderedDict() for feld in ("beschreibung", "sivas_id", "sivas_bezeichnung", "planquadrat", "block", "dxf_handle", "quelle"): if feld in kopf: extra[feld] = kopf[feld] # Die spec_id bleibt die Kettenkennung aus dem Protokoll: dieselbe # Kette soll in beiden Testlaeufen (Spec-Pfad und Aufzeichnungs- # Pfad) denselben Namen tragen, sonst laesst sich das Ergebnis # nicht paaren. alle.extend(flache_objekte(tid, spec, extra)) print("%s %-20s %2d Sektionen, %3d Tokens - Rundlauf ok" % (PREFIX, tid, len(spec["sektionen"]), len(tok))) if fehler: print("%s %d Kette(n) fehlerhaft - nichts geschrieben" % (PREFIX, fehler)) return 1 # Von Hand ergaenzte Objekte der Ziel-Datei ans Ende uebernehmen. fremde = fremde_objekte(args.ziel) if fremde: print("%s %d von Hand ergaenzte(s) Objekt(e) uebernommen: %s" % (PREFIX, len(fremde), ", ".join(str(o.get("id", "?")) for o in fremde))) alle.extend(fremde) text = json.dumps(alle, indent=2, ensure_ascii=False) with open(args.ziel, "w", encoding="utf-8", newline="\r\n") as fh: fh.write(arrays_einzeilig(text)) fh.write("\n") print("%s %d Objekte geschrieben -> %s" % (PREFIX, len(alle), args.ziel)) return 0 if __name__ == "__main__": sys.exit(main())