5902343352
Nach dem gruenen VF_SPEC_BAU-Lauf bauten zwei Tests dieselben 5 Ketten. Der Spec-Pfad ist jetzt der routinemaessige Bau-Test, der Aufzeichnungs-Pfad der abgeschaltete Regressionstest fuer das Journal-Format. Umbenennungen: - tests/testdata/vf_spec_hundm05.json -> tests/testdata/hundm05.json Die Spec ist das Eingabeformat und die Datei, in der weitere Ketten der Anlage nachgetragen werden. spec_ids jetzt VF_hundm05_LZ_* (wie im Aufzeichnungs-Protokoll), damit sich die Ergebnisse beider Laeufe paaren lassen. - tests/testdata/hundm05.json -> tests/testdata/hm_recformat.json tests/test_hundm05.lsp -> tests/test_hm_recformat.lsp (TEST_HM_RECFORMAT, Praefix hmrec:, Export hm_recformat:export-results) tests/test_hundm05.py -> tests/test_hm_recformat.py In alltests.json auf "disabled": true. Neu: tests/test_hundm05.lsp (TEST_HUNDM05) baut die Spec - OHNE eigene Bau-Logik, es ruft vsp-bau-datei aus Lisp/vf_spec.lsp. Damit gibt es genau einen Code-Pfad, der aus einer Spec Geometrie erzeugt; c:VF_SPEC_BAU nutzt denselben. tests/test_hundm05.py prueft das Ergebnis (die 9 Tests, die vorher in test_vf_spec.py standen); test_vf_spec.py ist jetzt reiner Uebersetzer-Test ohne CAD. Warum das Aufzeichnungsformat bleibt (Details in doc/TODO-plan-vf-interactive.md Abschnitt 4.7): - hm_recformat.json ist die EINZIGE eingecheckte Kopie der Originaldaten (data/polylines.dxf ist mit 124 MB per .gitignore ausgeschlossen). Die Spec ist daraus abgeleitet; ohne die Aufzeichnung faellt die rechte Seite des Rundlauf-Beweises weg. - Es prueft eine ANDERE Invariante: das Journal kommt roh aus der XDATA einer Kundenzeichnung. TEST_HM_RECFORMAT ist damit der einzige Test dafuer, dass ein BESTEHENDER VF_n-Block weiter abspielbar ist - also dass Doppelklick-Edit und 2D/3D-Konvertierung an Altbestand funktionieren. Ein spec-gebautes Journal kann das nicht zeigen, es kommt aus dem Uebersetzer. - Es dokumentiert die Frage-Reihenfolge (kommentar je Eintrag); die Spec verbirgt den Dialog, das ist ihr Zweck. Weitere Anpassungen: - Ergebnisdatei heisst nach der Spec-Datei (<basisname>_results.json), damit Befehl und Testrunner in dieselbe Datei schreiben. Verzeichnis: tests/output (Override DXFM_VF_SPEC_OUT); NICHT DXFM_RESULTS - das sind die Sivas-/ CSV-Exporte, dort landete die Datei ausserhalb des Testbaums. - Neu *vsp-dim-override* fuer TEST_HUNDM05_2D/_3D: wird je Kette angewandt, weil der Abbruch-Handler in vf-linienzug-modus *ssg-ils-dim* zurueck setzt. - Das Anlagenkuerzel der test_id in lib/vf_journal_export.py kam aus dem Namen der Ziel-JSON. Nach der Umbenennung haette eine Regenerierung die ids stillschweigend auf VF_hm_recformat_LZ_* geaendert und die Paarung der beiden Laeufe zerlegt - jetzt feste Konstante TEST_ID_ANLAGE. - conftest: hundm05_* Fixtures -> hmrec_* (die Spec-Fixtures stehen in tests/test_hundm05.py). Verifiziert: 98 pytest-Tests gruen (die 9 Ergebnis-Tests warten auf einen neuen TEST_HUNDM05-Lauf), alle .lsp lint-sauber, Spec-Regenerierung idempotent, TEST_VF_SPEC in BricsCAD 28 PASS / 0 FAIL. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
746 lines
29 KiB
Python
746 lines
29 KiB
Python
#!/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 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:
|
|
raise ValueError("Objekt ohne spec_id/glied/sub: %r" % obj)
|
|
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
|
|
|
|
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())
|