[FEAT] VF-Linienzug aus Eingabedaten bauen (Stufe 1: Spec -> Journal -> Replay)
Stufe 1 des Fahrplans in doc/TODO-plan-vf-interactive.md, fertig fuer
Modus 1. Eine SPEC beschreibt eine Kette in Domaenenwerten ("links",
"winkel", "aussen", ja/nein) statt in Menue-Codes; vsp-journal uebersetzt sie
in ein Eingabe-Journal, vsp-bau-aus-spec spielt es per
vfl-journal-replay-start durch den UNVERAENDERTEN vf-linienzug-modus. Der
fertige Block traegt damit dasselbe SSG_VF_EDIT-Journal wie eine handgebaute
Kette und bleibt per Doppelklick editier- und 2D/3D-konvertierbar - kein
Journal-Konsument muss etwas von Specs wissen.
CAD-Lauf VF_SPEC_BAU: 5 OK, 0 Fehler. Je Kette status "executed",
prompts 0 (keine Live-Eingabe erreicht), eingaben_offen 0, keine
Spec-Fehler, keine Bau-Meldungen, glieder_ist == glieder_soll,
Einfuegepunkt gleich dem Kettenstart der Spec.
Neu:
- Lisp/vf_spec.lsp: Code-Tabellen, Emitter vsp-journal, Validierung
vsp-pruefen, flacher JSON-Loader, Headless-Rahmen vsp-headless-an/-aus,
Ergebnis-Record, vsp-bau-aus-spec/-liste, c:VF_SPEC_BAU. Von vf_core.lsp
NACH vf_linienzug.lsp geladen - damit sind beide Ladewege abgedeckt (Menue
und der .scr-Loader), das MNL bleibt unveraendert.
- lib/vf_spec_export.py: derselbe Emitter in Python PLUS die Umkehrung
(spec_aus_journal), erzeugt die Spec-Testdaten.
- tests/testdata/vf_spec_hundm05.json: die 5 echten Ketten als Spec,
erzeugt statt handgeschrieben.
- tests/test_vf_spec.lsp (C:TEST_VF_SPEC, in alltests.json): 7 Faelle je
Glied-Typ, 7 Validierungsfaelle, Gruppieren, und die 5 echten Ketten gegen
ihr aufgezeichnetes Journal.
- tests/test_vf_spec.py: 36 Tests (Rundlauf, Spec-Daten, Validierung,
Erwartungswert-Abgleich, Ergebnisse von VF_SPEC_BAU).
Drei Regeln, die der Uebersetzer besitzen muss - jede wuerde sonst die ganze
Replay-Queue verschieben:
- Der Menue-Code haengt am Frame: erste Sektion 4 Optionen (kein GF-Bogen,
es gibt noch keine Richtung), jede spaetere 5.
- hz steht genau EINMAL im Journal, beim ersten Segment (vfl-in-abstand
journalisiert die Richtung nur bei freier Richtungswahl).
- Nur ein HORIZONTALER Erstkoerper einer VF-Einheit stellt die Separator-/
Endpunktfragen vorab (winkel1=0 laeuft durch vfl-baue-horizontal-koerper);
ein gewinkelter wird ohne jede Frage gebaut.
Wie die Richtigkeit belegt ist - drei Implementierungen derselben Grammatik,
die einander pruefen:
- Rundlauf gegen echte Daten: journal_aus_spec(spec_aus_journal(tok)) == tok
fuer alle 5 HundM-Ketten, tokenweise numerisch verglichen. Damit ist
bewiesen, dass die Spec das Journal verlustfrei abbildet.
- Der Weg, den LISP nimmt (flaches JSON -> Spec -> Journal), ergibt separat
geprueft dieselben Journale wie das aufgezeichnete Protokoll.
- Der unabhaengige Dekoder aus lib/vf_journal_export.py verdaut alle 7
synthetischen Emitter-Ausgaben restlos, auch die zwei Zweige, die in den
echten Daten fehlen (Linie-VF, gewinkelter Erstkoerper).
- Die Erwartungswerte im LISP-Test kommen aus dem geprueften Python-Emitter;
TestLispErwartungen liest sie zurueck und vergleicht erneut. Ohne das
koennte ein Uebertragungsfehler einen falschen Erwartungswert
festschreiben und der CAD-Lauf waere gruen, obwohl der Emitter falsch
liegt.
Abweichungen von der Planung (Begruendungen in doc/TODO-plan-vf-interactive.md
Abschnitt 4.6): vsp-pruefen ruft den Emitter statt eine zweite Regelmenge zu
pflegen; das vollstaendige Journal steht nicht im Record (statt dessen
glieder_soll/glieder_ist als Desync-Detektor); Zusatzfeld segment_typ am
Glied "Linie", weil dort vfl-segment-entscheidung selbst zwischen GF und VF
waehlt; vf_ende-Werte "automatisch"/"zielpunkt-ohne-es" fuer die Faelle ohne
Frage; Praefix vsp- statt vfs- (das gehoert vf_standard.lsp).
Zusaetzlich:
- tests/test_vf_headless_statisch.py prueft jetzt auch, dass
Lisp/vf_spec.lsp UEBERHAUPT keine Eingabe-/Dialogfunktion aufruft - beim
Daten-Pfad ist die Null die Vorgabe.
- tests/testdata/mubea.json: die drei einzeln eingefuegten S-LP-Separatoren
entfernt (Fortsetzung von 699744e - die Separatoren stecken jetzt in den
Staustreckenbloecken).
Noch offen: die Uebersetzer fuer Modus 2 und 3. Modus 2 braucht laut
Fahrplan erst 2-3 echte "linienzug2"-Journale als Referenz -
tests/testdata/hundm05.json enthaelt ausschliesslich Modus-1-Ketten.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,742 @@
|
||||
#!/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/hundm05.json \
|
||||
tests/testdata/vf_spec_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/hundm05.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 Form wie
|
||||
# tests/testdata/hundm05.json - 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
|
||||
hundm05: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 (hundm05.json) in Specs "
|
||||
"uebersetzen und den Rundlauf pruefen.")
|
||||
p.add_argument("protokoll", help="Eingabe (Schema hundm05.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]
|
||||
alle.extend(flache_objekte(tid.replace("hundm05_LZ", "spec_LZ"),
|
||||
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())
|
||||
Reference in New Issue
Block a user