lib/vf_journal_export.py (Journal.dekodiere) und lib/vf_spec_export.py (SpecLeser) implementierten denselben Tokenwalk (Praeambel, Sektions- schleife, Glied-Dispatch, VF-Einheit-Unterschleife) sowie dieselben Code-Tabellen (SEITE/VARIANTE/VERTEILUNG/IM_VF/ENDPUNKT/ENDE2/ENDE3/ GEFAELLE/GF_LAENGE_MODUS, MENU_START/MENU_START_AKTUELL/MENU_WEITER, Grammatik) unabhaengig voneinander - Aenderungen an der Grammatik (wie im vorigen Commit) mussten von Hand synchron gehalten werden und sind bereits einmal auseinandergelaufen. Neuer gemeinsamer Parser-Kern (Kern/lies_journal) plus Sink-Schnittstelle (SpecSink in vf_spec_export.py, ProtokollSink in vf_journal_export.py) ersetzt die doppelte Implementierung; Code-Tabellen/Grammatik-Objekte leben jetzt nur noch in lib/vf_linienzug_grammatik.py. Journal behaelt nur noch die reine XDATA/DXF-Zerlegung, SpecSchreiber (Spec -> Journal, Gegenstueck zu Lisp/vf_spec.lsp) bleibt unveraendert in vf_spec_export.py. Zusaetzlich: pruefe_flachen_freitext schlaegt jetzt hart fehl, wenn generierter Freitext (quelle/beschreibung) ein rohes Anfuehrungszeichen enthaelt - der Lisp-seitige Flach-JSON-Leser (ssg-cfg-parse-value) kennt keine Escapes und wuerde daran die Zeichenkette (und damit die gesamte restliche Datei) falsch zerlegen. Byte-identischer Output gegen die Baseline geprueft (beide CLIs, echte Eingaben); pytest tests/test_vf_spec.py + test_vf_journal_grammatik.py unveraendert: 2 bekannte, vorbestehende Fehlschlaege, 38 bestanden. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
606 lines
25 KiB
Python
606 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
vf_journal_export.py - Liest die SSG_VF_EDIT-Eingabe-Journale der VF_n-Bloecke
|
|
aus einer Zeichnung und schreibt sie als Frage-Antwort-Protokoll (JSON) im
|
|
Schema von tests/testdata/linienzug_tests.json.
|
|
|
|
Hintergrund: Der Linienzug-Modus (Lisp/vf_linienzug.lsp) ist nicht scriptbar -
|
|
er fragt jede Eingabe einzeln ab (getpoint/getstring/getint/getreal) und
|
|
zeichnet sie in einem Journal auf, das nach dem Bau als XDATA (App
|
|
SSG_VF_EDIT, Marker "linienzug") am fertigen VF_n-Block landet. Dieses Journal
|
|
ist die vollstaendige, wiederabspielbare Bau-Anleitung der Kette
|
|
(vfl-journal-replay-start). Dieses Skript dreht die Serialisierung aus
|
|
vfl-entry->string wieder zurueck und annotiert jeden Wert mit der Frage, die
|
|
er beantwortet - abgeleitet aus dem Frageablauf von vf-linienzug-modus.
|
|
|
|
Den Frageablauf selbst laeuft der GEMEINSAME Parser-Kern
|
|
(lib/vf_linienzug_grammatik.py), den sich dieses Skript mit
|
|
lib/vf_spec_export.py teilt: dort steht die Grammatik EINMAL, hier nur noch
|
|
die ProtokollSink mit den Annotationstexten. Frueher war der Ablauf in beiden
|
|
Dateien von Hand nachgebaut und lief regelmaessig auseinander.
|
|
|
|
Bewusst OHNE ezdxf: gebraucht werden nur INSERT-Kopf (Blockname, Einfuegepunkt,
|
|
Drehung) und die 1000er-Gruppen der XDATA - das ist reine Gruppencode-Paar-
|
|
Lesung und funktioniert auch bei ASCII-DXF im 100-MB-Bereich.
|
|
|
|
Zwei Eigenheiten der Quelldaten, die hier aufgeloest werden:
|
|
|
|
1. Das Newline zwischen zwei Journal-Eintraegen steht im ASCII-DXF als
|
|
Steuerzeichen-Notation "^J" (zwei Zeichen) - ein echtes 0x0A wuerde die
|
|
zeilenweise DXF-Struktur zerreissen. Getrennt wird darum an "^J".
|
|
2. Kopierte Ketten teilen ihr Journal: die vier VF_46-Bloecke in
|
|
data/polylines.dxf sind Kopien EINER gebauten Kette und tragen deshalb
|
|
Wort fuer Wort dasselbe Journal - inklusive des Startpunkts der zuerst
|
|
gebauten Kette. Alle uebrigen Eintraege sind relativ (Distanz, Winkel,
|
|
Zielhoehen), nur dieser eine Punkt ist absolut. Er wird darum je Instanz
|
|
auf deren echten Blockeinfuegepunkt gezogen (--kein-startpunkt-fix
|
|
schaltet das ab), damit jeder Eintrag fuer sich abspielbar ist.
|
|
|
|
Aufruf:
|
|
python vf_journal_export.py <dxf> <output.json> [--csv <sivas_export.csv>]
|
|
[--kein-startpunkt-fix]
|
|
|
|
Beispiel (erzeugt tests/testdata/hm_recformat.json):
|
|
python lib/vf_journal_export.py data/polylines.dxf \
|
|
tests/testdata/hm_recformat.json --csv results/HundM_export.csv
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
# Grammatik (Code-Tabellen, Menue-Zuordnung) UND Frageablauf kommen aus dem
|
|
# gemeinsamen Modul - hier steht nur noch die Annotation dazu. SEITE/VARIANTE
|
|
# braucht zusaetzlich die Kurzbeschreibung (kurzbeschreibung): sie liest die
|
|
# fertigen Protokolleintraege nach und hat dort nur den rohen Menue-Code.
|
|
from vf_linienzug_grammatik import ( # noqa: E402
|
|
GRAMMATIK_AUFZEICHNUNG, SEITE, Sink, VARIANTE, lies_journal)
|
|
|
|
PREFIX = "[vf_journal_export]"
|
|
|
|
XDATA_APP = "SSG_VF_EDIT"
|
|
MARKER = "linienzug"
|
|
|
|
# Steuerzeichen-Notation des Eintrags-Trenners im ASCII-DXF (siehe Kopf).
|
|
TRENNER = "^J"
|
|
|
|
RUND = 6
|
|
|
|
# Anlagenkuerzel in den test_ids (VF_<kuerzel>_LZ_01 ...).
|
|
TEST_ID_ANLAGE = "hundm05"
|
|
|
|
STEP_KOMMENTAR = ("Glied-Marker (vfl-journal-mark) - beim Replay "
|
|
"uebersprungen, Label = gewaehlter Segmenttyp")
|
|
|
|
|
|
# ============================================================
|
|
# DXF lesen
|
|
# ============================================================
|
|
|
|
def gruppen(dxf_pfad):
|
|
"""Alle (Gruppencode, Wert)-Paare der Datei."""
|
|
with open(dxf_pfad, "r", encoding="utf-8", errors="replace") as fh:
|
|
zeilen = fh.read().splitlines()
|
|
return [(zeilen[i].strip(), zeilen[i + 1])
|
|
for i in range(0, len(zeilen) - 1, 2)]
|
|
|
|
|
|
def lese_vf_inserts(dxf_pfad):
|
|
"""INSERTs mit SSG_VF_EDIT-XDATA einsammeln.
|
|
|
|
Rueckgabe: Liste von Dicts (handle, block, punkt, rotation, marker,
|
|
journal_roh). Das Journal steckt in den 1000er-Gruppen NACH dem Marker
|
|
und ist auf 250 Zeichen je Gruppe gestueckelt (DXF-Limit 255).
|
|
"""
|
|
paare = gruppen(dxf_pfad)
|
|
treffer = []
|
|
i = 0
|
|
while i < len(paare):
|
|
code, wert = paare[i]
|
|
if code != "0" or wert != "INSERT":
|
|
i += 1
|
|
continue
|
|
e = {"handle": None, "block": None, "x": 0.0, "y": 0.0, "z": 0.0,
|
|
"rotation": 0.0, "marker": None, "chunks": []}
|
|
in_app = False
|
|
j = i + 1
|
|
while j < len(paare) and paare[j][0] != "0":
|
|
c, v = paare[j]
|
|
if c == "5" and e["handle"] is None:
|
|
e["handle"] = v
|
|
elif c == "2" and e["block"] is None:
|
|
e["block"] = v
|
|
elif c == "10":
|
|
e["x"] = float(v)
|
|
elif c == "20":
|
|
e["y"] = float(v)
|
|
elif c == "30":
|
|
e["z"] = float(v)
|
|
elif c == "50":
|
|
e["rotation"] = float(v)
|
|
elif c == "1001":
|
|
in_app = (v == XDATA_APP)
|
|
elif c == "1000" and in_app:
|
|
if e["marker"] is None:
|
|
e["marker"] = v
|
|
else:
|
|
e["chunks"].append(v)
|
|
j += 1
|
|
if e["marker"] is not None:
|
|
e["punkt"] = [e["x"], e["y"], e["z"]]
|
|
e["journal_roh"] = "".join(e["chunks"])
|
|
del e["chunks"]
|
|
treffer.append(e)
|
|
i = j
|
|
return treffer
|
|
|
|
|
|
# ============================================================
|
|
# Journal dekodieren
|
|
# ============================================================
|
|
# Den Frageablauf laeuft der gemeinsame Kern in lib/vf_linienzug_grammatik.py
|
|
# (dieselbe Routine, die lib/vf_spec_export.py fuer die Spec benutzt). Hier
|
|
# steht nur noch die ANNOTATION: zu jeder Frage die Konsolen-Vorlage aus
|
|
# vf_linienzug.lsp und - wo die Domaenen-Kurzform nicht schon der
|
|
# Antworttext ist - der Klartext der Antwort.
|
|
|
|
# Prompt-Vorlagen je Frage-Schluessel (siehe Sink-Doku im Grammatik-Modul).
|
|
# Ein Schluessel, nicht der Feldname: dasselbe Feld ("hoehe") beantwortet an
|
|
# verschiedenen Stellen verschiedene Fragen.
|
|
|
|
_REAL_FRAGEN = {
|
|
"start_hoehe": "getreal Hoehe (Z) Startpunkt -> %.1f mm",
|
|
"gf_winkel": "getreal Neigungswinkel -> %.1f Grad",
|
|
"gf_hoehe": "getreal Zielhoehe (Z) Linienendpunkt -> %.1f mm",
|
|
"vf_zielhoehe": "getreal Zielhoehe (Z) Linienendpunkt -> %.1f mm",
|
|
"kettenende_hoehe": "getreal Zielhoehe (Z) am Kettenende -> %.1f mm",
|
|
"aufab_hoehe": "getreal Zielhoehe (Z) am Segmentende -> %.1f mm",
|
|
"eingang_hoehe": "getreal Ziel-Hoehe fuer den GF1-Einlauf der "
|
|
"VF-Einheit (nur bei komplett neu gebautem Eingang) "
|
|
"-> %.1f mm",
|
|
}
|
|
|
|
_INT_FRAGEN = {
|
|
"gf_bogen_winkel": "getint GF-Bogen Winkel (30/60/90 Grad) -> %d",
|
|
"vario_winkel": "getint Vario-Kurve Winkel (30/60/90 Grad, Wert "
|
|
"direkt) -> %d Grad",
|
|
"winkel_idx": "getint Winkelwahl aus den gueltigen Kandidaten "
|
|
"(vfl-waehle-winkel) -> Option %d",
|
|
}
|
|
|
|
# STR-Antworten, deren Inhalt schon der Wert ist (kein Code).
|
|
_ROH_FRAGEN = {
|
|
"as_winkel": "getstring AS-Element Winkel (Wert direkt, 90 oder 30 "
|
|
"Grad) -> %s Grad",
|
|
"es_winkel": "getstring ES-Element Winkel (Wert direkt, 90 oder 30 "
|
|
"Grad) -> %s Grad",
|
|
}
|
|
|
|
_JA_NEIN_FRAGEN = {
|
|
"as": "getstring AS-Element setzen? (1=Ja, 2=Nein) -> %s",
|
|
"sep_vor": "getstring Separator VOR dem horizontalen Stueck? (1=Ja, "
|
|
"2=Nein) -> %s",
|
|
"sep_nach": "getstring Separator NACH dem horizontalen Stueck? (1=Ja, "
|
|
"2=Nein) -> %s",
|
|
"es": "getstring ES-Element setzen? (1=Ja, 2=Nein - Kette endet direkt "
|
|
"am Zielpunkt) -> %s",
|
|
"separator": "getstring Separator an dieser Stelle einfuegen? (1=Ja, "
|
|
"2=Nein) -> %s",
|
|
}
|
|
|
|
# Die Menue-Antwort auf "Naechstes Element" ist der Glied-Name; sein
|
|
# Klartext ist derselbe am Kettenanfang wie spaeter, nur die aufgezaehlten
|
|
# Optionen im Fragetext unterscheiden sich. Darum EINE Antworttabelle fuer
|
|
# beide Fragen (frueher zwei code-indizierte Tabellen nebeneinander).
|
|
_GLIED_TEXT = {"GF-Bogen": "GF-Bogen", "Linie-GF": "Neue Linie: GF",
|
|
"Linie-VF": "Neue Linie: Ab/Auf VF",
|
|
"Horizontal-VF": "Horizontal-VF",
|
|
"Linie": "Neue Linie bis Kettenende"}
|
|
|
|
# STR-Antworten ueber eine Code-Tabelle: (Vorlage, Klartext je Domaenenwert).
|
|
# Steht statt der Klartexte None, ist der Domaenenwert selbst schon der
|
|
# Antworttext ("links"/"rechts", "aussen"/"innen").
|
|
#
|
|
# ACHTUNG: die Fragetexte zaehlen die Menue-Optionen der AUFZEICHNUNG auf
|
|
# (4 Optionen am Kettenanfang). Wird das Protokoll je gegen die aktuelle
|
|
# Grammatik erzeugt (GRAMMATIK_AKTUELL, 6 Optionen), muss "menu_start" hier
|
|
# mitgezogen werden - die Antworten selbst bleiben richtig.
|
|
_CODE_FRAGEN = {
|
|
"as_seite": ("getstring AS-Seite (1=links, 2=rechts) -> %s", None),
|
|
"es_seite": ("getstring ES-Seite (1=links, 2=rechts) -> %s", None),
|
|
"gf_bogen_seite": ("getstring GF-Bogen Seite (1=links, 2=rechts) -> %s",
|
|
None),
|
|
"vario_seite": ("getstring Vario-Kurve Seite (1=links, 2=rechts) -> %s",
|
|
None),
|
|
"vario_variante": ("getstring Vario-Kurve Variante (1=aussen, 2=innen) "
|
|
"-> %s", None),
|
|
"gefaelle": ("getstring Gefaelle festlegen (1=ueber Zielhoehe, 2=ueber "
|
|
"Neigungswinkel) -> %s",
|
|
{"hoehe": "ueber Zielhoehe",
|
|
"winkel": "ueber Neigungswinkel"}),
|
|
"ende3": ("getstring Ist das Kettenende? (1=Ja mit ES, 2=Ja ohne ES, "
|
|
"3=Nein weiterbauen) -> %s",
|
|
{"ja-mit-es": "Ja, mit ES-Element",
|
|
"ja-ohne-es": "Ja, ohne ES-Element",
|
|
"nein": "Nein, weiterbauen"}),
|
|
"ende2": ("getstring Ist das Kettenende? (1=Ja, Separator + ES-Element, "
|
|
"2=Nein, weiterbauen) -> %s",
|
|
{"ja-mit-es": "Ja, Separator + ES-Element",
|
|
"nein": "Nein, weiterbauen"}),
|
|
"gf_verteilung": ("getstring GF-Verteilung (1=halbe Staustrecke am "
|
|
"Ausgang, 2=alles am Einlauf) -> %s",
|
|
{"haelfte": "halbe Staustrecke am Ausgang",
|
|
"einlauf": "alles am Einlauf"}),
|
|
"endpunkt": ("getstring Ist der Endpunkt der Foerderer? (1=Ja nur "
|
|
"Motorstation, 2=Nein weiterbauen, 3=Ja Motorstation + "
|
|
"Kettenende) -> %s",
|
|
{"motorstation": "Ja, nur Motorstation",
|
|
"weiter": "Nein, weiterbauen",
|
|
"kettenende": "Ja, Motorstation + Kettenende am "
|
|
"Zielpunkt"}),
|
|
"im_vf": ("getstring Naechstes in der VF-Einheit (1=Horizontaler "
|
|
"Foerderer, 2=Vario-Kurve, 3=Auf/Ab-Foerderer) -> %s",
|
|
{"horizontal": "Horizontaler Foerderer",
|
|
"vario-kurve": "Vario-Kurve",
|
|
"auf-ab": "Auf/Ab-Foerderer"}),
|
|
"menu_start": ("getstring Naechstes Element am Kettenanfang (1=Neue "
|
|
"Linie GF, 2=Neue Linie Ab/Auf VF, 3=Horizontal-VF, "
|
|
"4=Neue Linie bis Kettenende) -> %s", _GLIED_TEXT),
|
|
"menu_weiter": ("getstring Naechstes Element (1=GF-Bogen, 2=Neue Linie "
|
|
"GF, 3=Neue Linie Ab/Auf VF, 4=Horizontal-VF, 5=Neue "
|
|
"Linie bis Kettenende) -> %s", _GLIED_TEXT),
|
|
}
|
|
|
|
# Kontext der Laengeneingabe je Frage-Schluessel (die "%s"-Stelle im
|
|
# getpoint-Kommentar).
|
|
_LINIE_TEXT = {
|
|
"gf_linie": "Endpunkt der GF-Linie",
|
|
"vf_linie": "Endpunkt des Ab/Auf-VF-Segments",
|
|
"horizontal_vf": "Endpunkt des horizontalen VF-Stuecks",
|
|
"horizontal_sub": "Endpunkt des horizontalen VF-Stuecks",
|
|
"aufab_sub": "Endpunkt des Auf/Ab-VF-Segments",
|
|
"linie_kettenende": "Endpunkt der Linie bis Kettenende (Ziel-XY der "
|
|
"Kette)",
|
|
"kettenende_ziel": "Zielpunkt des Kettenendes",
|
|
}
|
|
|
|
|
|
def _vorlage(tabelle, frage):
|
|
if frage not in tabelle:
|
|
raise ValueError("keine Annotation fuer die Frage %r hinterlegt "
|
|
"(lib/vf_journal_export.py)" % frage)
|
|
return tabelle[frage]
|
|
|
|
|
|
class ProtokollSink(Sink):
|
|
"""Schreibt aus den Meldungen des Parser-Kerns das annotierte
|
|
Frage-Antwort-Protokoll (Schema tests/testdata/hm_recformat.json):
|
|
je verbrauchtem Wert ein Eintrag {"typ", Wertfeld(er), "kommentar"}.
|
|
|
|
Die Reihenfolge der Schluessel im Eintrag ist fest (typ, Wert,
|
|
kommentar) - das erzeugte JSON wird von Hand gelesen und mit frueheren
|
|
Laeufen verglichen.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.out = []
|
|
|
|
def ergebnis(self):
|
|
return self.out
|
|
|
|
def add(self, typ, kommentar, **felder):
|
|
eintrag = {"typ": typ}
|
|
eintrag.update(felder)
|
|
eintrag["kommentar"] = kommentar
|
|
self.out.append(eintrag)
|
|
|
|
# --- Struktur: nur die Glied-Marker stehen im Protokoll ---
|
|
|
|
def sektion_start(self, label):
|
|
self.add("step", STEP_KOMMENTAR, wert=label)
|
|
|
|
def marke(self, label):
|
|
self.add("step", STEP_KOMMENTAR, wert=label)
|
|
|
|
def glied_gewaehlt(self, label, code, frage):
|
|
vorlage, texte = _vorlage(_CODE_FRAGEN, frage)
|
|
self.add("string", vorlage % texte[label], wert=code)
|
|
|
|
# --- Werte ---
|
|
|
|
def punkt_abs(self, punkt):
|
|
self.add("point_abs", "getpoint Startpunkt der Kette (absoluter "
|
|
"Weltpunkt, einziger Anker der ganzen Kette)", wert=punkt)
|
|
|
|
def linie(self, dl, hz, frage):
|
|
was = _vorlage(_LINIE_TEXT, frage)
|
|
if hz is None:
|
|
self.add("point_rel",
|
|
"getpoint %s: dL=%.1f mm (Fahrtrichtung vom "
|
|
"Vorgaenger geerbt)" % (was, dl), dL=dl)
|
|
else:
|
|
self.add("point_rel",
|
|
"getpoint %s: dL=%.1f mm, gesnappte Fahrtrichtung "
|
|
"hz=%.0f Grad (nur beim ersten Segment der Kette "
|
|
"journalisiert)" % (was, dl, hz), dL=dl, hz=hz)
|
|
|
|
def real_feld(self, name, wert, frage):
|
|
self.add("real", _vorlage(_REAL_FRAGEN, frage) % wert, wert=wert)
|
|
|
|
def int_feld(self, name, wert, frage):
|
|
self.add("int", _vorlage(_INT_FRAGEN, frage) % wert, wert=wert)
|
|
|
|
def roh_feld(self, name, wert, frage):
|
|
self.add("string", _vorlage(_ROH_FRAGEN, frage) % wert, wert=wert)
|
|
|
|
def ja_nein(self, name, code, wert, frage):
|
|
self.add("string",
|
|
_vorlage(_JA_NEIN_FRAGEN, frage) % ("Ja" if wert else "Nein"),
|
|
wert=code)
|
|
|
|
def code_feld(self, name, code, wert, frage):
|
|
self._code(code, wert, frage)
|
|
|
|
def wahl(self, name, code, wert, frage):
|
|
self._code(code, wert, frage)
|
|
|
|
def _code(self, code, wert, frage):
|
|
"""Eine ueber eine Code-Tabelle beantwortete Frage. Der Domaenenwert
|
|
ist entweder selbst schon der Antworttext (links/rechts,
|
|
aussen/innen) oder wird ueber die Klartexttabelle der Frage
|
|
uebersetzt; im Protokoll steht als "wert" der rohe Journal-Code, so
|
|
wie er in der XDATA stand."""
|
|
vorlage, texte = _vorlage(_CODE_FRAGEN, frage)
|
|
self.add("string", vorlage % (texte[wert] if texte else wert),
|
|
wert=code)
|
|
|
|
# implizit() schreibt nichts: segment_typ und das automatische VF-Ende
|
|
# ergeben sich aus dem Ablauf, es wurde dafuer kein Token gelesen.
|
|
|
|
|
|
def tokens_aus_journal(roh):
|
|
"""Die Rohform der XDATA ("ART:wert", getrennt durch TRENNER) in
|
|
Tokenpaare zerlegen."""
|
|
tok = []
|
|
for zeile in roh.split(TRENNER):
|
|
if zeile:
|
|
art, _, wert = zeile.partition(":")
|
|
tok.append((art, wert))
|
|
return tok
|
|
|
|
|
|
class Journal(object):
|
|
"""Ein Journal (kind:payload je Zeile) in das Frage-Antwort-Protokoll
|
|
uebersetzen.
|
|
|
|
Der Ablauf selbst steckt im gemeinsamen Parser-Kern
|
|
(lib/vf_linienzug_grammatik.py) - diese Klasse haelt nur noch die
|
|
Rohform-Zerlegung und die Sink zusammen.
|
|
"""
|
|
|
|
def __init__(self, roh, grammatik=GRAMMATIK_AUFZEICHNUNG):
|
|
self.tok = tokens_aus_journal(roh)
|
|
self.grammatik = grammatik
|
|
|
|
def dekodiere(self):
|
|
return lies_journal(self.tok, ProtokollSink(), self.grammatik)
|
|
|
|
|
|
# ============================================================
|
|
# Sivas-Export als Quervergleich
|
|
# ============================================================
|
|
|
|
def lese_strecken(csv_pfad):
|
|
"""Die "ILS 2.0 Strecke"-Zeilen eines Sivas-Exports (EXPORTSIVAS).
|
|
|
|
Dient nur der Zuordnung/Dokumentation: die Merkmale (DeltaL, DeltaH,
|
|
L_GF_m, L_VF_m, ...) sind die Attribute des fertigen VF_n-Blocks und
|
|
damit der Erwartungswert zu einem nachgebauten Foerderer.
|
|
"""
|
|
strecken = []
|
|
with open(csv_pfad, encoding="utf-8") as fh:
|
|
for zeile in csv.DictReader(fh, delimiter=";"):
|
|
if zeile.get("TeileArt") != "ILS 2.0 Strecke":
|
|
continue
|
|
if not zeile.get("Nachbarn"):
|
|
continue # unverbunden - gehoert zu keinem Kreisel
|
|
strecken.append({
|
|
"id": zeile["TeileId"],
|
|
"bezeichnung": zeile["Bezeichnung"],
|
|
"planquadrat": zeile["Planquadrat"],
|
|
"x": float(zeile["Position"].split(",")[0]),
|
|
"merkmale": json.loads(zeile["Merkmale"]),
|
|
})
|
|
return strecken
|
|
|
|
|
|
def erwartung_hinweis(strecke):
|
|
m = strecke["merkmale"]
|
|
felder = ["DeltaL_mm", "DeltaH_mm", "Hoehe_Von_mm", "Hoehe_Bis_mm",
|
|
"L_GF_m", "L_VF_m", "VF_Winkel", "Anzahl_Separator"]
|
|
werte = ", ".join("%s=%s" % (f, m[f]) for f in felder if f in m)
|
|
return ("Sivas-Export (EXPORTSIVAS), Strecke %s: %s"
|
|
% (strecke["id"], werte))
|
|
|
|
|
|
# ============================================================
|
|
# Kurzbeschreibung der Kette
|
|
# ============================================================
|
|
|
|
def kurzbeschreibung(eintraege):
|
|
"""Kette in einer Zeile: die Glied-Marker mit ihren Kernwerten."""
|
|
teile = []
|
|
for i, e in enumerate(eintraege):
|
|
if e["typ"] != "step":
|
|
continue
|
|
folge = eintraege[i + 1:i + 8]
|
|
strings = [x["wert"] for x in folge if x["typ"] == "string"]
|
|
ints = [x["wert"] for x in folge if x["typ"] == "int"]
|
|
reals = [x["wert"] for x in folge if x["typ"] == "real"]
|
|
laengen = [x["dL"] for x in folge if x["typ"] == "point_rel"]
|
|
label = e["wert"]
|
|
if label == "Vario-Kurve":
|
|
teile.append("Vario-Kurve %d %s %s"
|
|
% (ints[0], SEITE.get(strings[0], "?"),
|
|
VARIANTE.get(strings[1], "?")))
|
|
elif label == "ES":
|
|
teile.append("ES %s %s" % (strings[0], SEITE.get(strings[1], "?")))
|
|
elif label == "Linie-GF":
|
|
teile.append("GF-Linie %.0f mm / %.0f Grad"
|
|
% (laengen[0], reals[0]))
|
|
elif label == "Horizontal-VF":
|
|
teile.append("Horizontal-VF %.0f mm" % laengen[0])
|
|
elif label == "Linie":
|
|
teile.append("Linie bis Kettenende %.0f mm auf Z%.0f"
|
|
% (laengen[0], reals[0]))
|
|
elif label == "GF-Bogen":
|
|
teile.append("GF-Bogen %d %s" % (ints[0],
|
|
SEITE.get(strings[0], "?")))
|
|
return " -> ".join(teile)
|
|
|
|
|
|
# ============================================================
|
|
# Testfaelle bauen
|
|
# ============================================================
|
|
|
|
def baue_testfaelle(inserts, strecken, startpunkt_fix, quelle_datei,
|
|
ziel_datei):
|
|
"""FLACHE Testfall-Liste: je Kette ein Kopf-Objekt (traegt "test_id"),
|
|
danach ihre Eingaben (tragen "typ") - Struktur wie linienzug_tests.json.
|
|
|
|
Bewusst NICHT verschachtelt ("eingaben": [...]): ssg-load-json /
|
|
ssg-parse-json-array (Lisp/ssg_core.lsp) liest die Datei ZEILENWEISE und
|
|
behandelt jedes "{" als Objektanfang. Ein verschachteltes Array wuerde die
|
|
Kopf-Objekte verschlucken. Der Treiber gruppiert wieder ueber "test_id"
|
|
(siehe hmrec:gruppiere in tests/test_hm_recformat.lsp).
|
|
"""
|
|
# Anlagenkuerzel der test_id: FEST, nicht aus dem Dateinamen
|
|
# abgeleitet. Die id benennt die ANLAGE, nicht die Datei - und sie
|
|
# muss stabil bleiben, weil dieselbe Kette in beiden Testlaeufen
|
|
# (Aufzeichnungsformat und Spec) denselben Namen tragen soll:
|
|
# tests/testdata/hundm05.json (die Spec) wird aus dieser Datei
|
|
# erzeugt und uebernimmt die ids. Frueher kam das Kuerzel aus dem
|
|
# Namen der Ziel-JSON; nach der Umbenennung in hm_recformat.json
|
|
# haette eine Regenerierung die ids stillschweigend geaendert.
|
|
praefix = TEST_ID_ANLAGE
|
|
faelle = []
|
|
for nr, e in enumerate(sorted(inserts, key=lambda i: i["x"]), start=1):
|
|
eintraege = Journal(e["journal_roh"]).dekodiere()
|
|
journal_start = list(eintraege[0]["wert"])
|
|
punkt = [round(v, RUND) for v in e["punkt"]]
|
|
if startpunkt_fix:
|
|
eintraege[0]["wert"] = punkt
|
|
eintraege[1]["wert"] = punkt[2]
|
|
eintraege[1]["kommentar"] = ("getreal Hoehe (Z) Startpunkt "
|
|
"-> %.1f mm" % punkt[2])
|
|
else:
|
|
punkt = [round(v, RUND) for v in journal_start]
|
|
strecke = None
|
|
if strecken:
|
|
strecke = min(strecken, key=lambda s: abs(s["x"] - e["x"]))
|
|
kopf = {
|
|
"test_id": "VF_%s_LZ_%02d" % (praefix, nr),
|
|
# "anlagetyp", nicht "typ": "typ" markiert im flachen Array einen
|
|
# EINGABE-Eintrag (point_abs/real/string/int/step) - ein Kopf mit
|
|
# "typ" waere davon nicht zu unterscheiden.
|
|
"anlagetyp": "linienzug",
|
|
"modus": 1,
|
|
"anzahl_eingaben": len(eintraege),
|
|
"block": e["block"],
|
|
"dxf_handle": e["handle"],
|
|
"startpunkt_mm": punkt,
|
|
"rotation_grad": round(e["rotation"], 4),
|
|
"beschreibung": "Linienzug (Modus 1), Start Z%.0f: %s"
|
|
% (punkt[2], kurzbeschreibung(eintraege)),
|
|
}
|
|
if strecke:
|
|
kopf["sivas_id"] = strecke["id"]
|
|
kopf["sivas_bezeichnung"] = strecke["bezeichnung"]
|
|
kopf["planquadrat"] = strecke["planquadrat"]
|
|
kopf["erwartung_hinweis"] = erwartung_hinweis(strecke)
|
|
# Ohne Anfuehrungszeichen im Text: ssg-cfg-parse-value schneidet einen
|
|
# String am ersten/letzten Zeichen ab und kennt keine Escapes.
|
|
kopf["quelle"] = ("INSERT %s (Block %s) in %s, XDATA-App %s, Marker "
|
|
"%s; im Journal aufgezeichneter Kettenstart: [%s]"
|
|
% (e["handle"], e["block"],
|
|
os.path.basename(quelle_datei), XDATA_APP,
|
|
e["marker"],
|
|
", ".join("%.6f" % v for v in journal_start)))
|
|
faelle.append(kopf)
|
|
faelle.extend(eintraege)
|
|
return faelle
|
|
|
|
|
|
def arrays_einzeilig(text):
|
|
"""Rein numerische JSON-Arrays auf eine Zeile ziehen (wie
|
|
dxf_scan_components.py) - ssg-parse-json-array (Lisp/ssg_core.lsp) liest
|
|
zeilenweise und kann ein umgebrochenes Zahlen-Array nicht lesen."""
|
|
def zusammen(m):
|
|
return "[" + " ".join(m.group(1).split()) + "]"
|
|
return re.sub(r"\[([\s\d.,+\-eE]*?)\]", zusammen, text)
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(
|
|
description="Liest die SSG_VF_EDIT-Eingabe-Journale der VF_n-Bloecke "
|
|
"einer Zeichnung und schreibt sie als Frage-Antwort-"
|
|
"Protokoll (Schema linienzug_tests.json).")
|
|
p.add_argument("dxf", help="Quellzeichnung (ASCII-DXF)")
|
|
p.add_argument("output", help="Ziel-JSON")
|
|
p.add_argument("--csv", help="Sivas-Export (EXPORTSIVAS) der gleichen "
|
|
"Zeichnung - liefert Strecken-ID und "
|
|
"Erwartungswerte je Kette")
|
|
p.add_argument("--kein-startpunkt-fix", action="store_true",
|
|
help="Kettenstart NICHT auf den Blockeinfuegepunkt "
|
|
"ziehen (Journal unveraendert uebernehmen)")
|
|
args = p.parse_args()
|
|
|
|
if not os.path.isfile(args.dxf):
|
|
print("%s FEHLER: DXF nicht gefunden: %s" % (PREFIX, args.dxf))
|
|
sys.exit(1)
|
|
|
|
mb = os.path.getsize(args.dxf) / (1024.0 * 1024.0)
|
|
print("%s lese %s (%.1f MB) ..." % (PREFIX, args.dxf, mb))
|
|
inserts = lese_vf_inserts(args.dxf)
|
|
print("%s %d INSERT(s) mit %s-XDATA gefunden"
|
|
% (PREFIX, len(inserts), XDATA_APP))
|
|
inserts = [e for e in inserts if e["marker"].startswith(MARKER)]
|
|
if not inserts:
|
|
print("%s FEHLER: kein Linienzug-Journal (Marker \"%s\") in der "
|
|
"Zeichnung" % (PREFIX, MARKER))
|
|
sys.exit(1)
|
|
|
|
strecken = lese_strecken(args.csv) if args.csv else []
|
|
if args.csv:
|
|
print("%s %d verbundene Strecken im Sivas-Export"
|
|
% (PREFIX, len(strecken)))
|
|
|
|
faelle = baue_testfaelle(inserts, strecken,
|
|
not args.kein_startpunkt_fix, args.dxf,
|
|
args.output)
|
|
koepfe = [f for f in faelle if "test_id" in f]
|
|
for f in koepfe:
|
|
print("%s %s %s (%s) %d Eingaben"
|
|
% (PREFIX, f["test_id"], f["block"],
|
|
f.get("sivas_bezeichnung", "ohne Sivas-Zuordnung"),
|
|
f["anzahl_eingaben"]))
|
|
|
|
ziel_dir = os.path.dirname(os.path.abspath(args.output))
|
|
if ziel_dir and not os.path.isdir(ziel_dir):
|
|
os.makedirs(ziel_dir)
|
|
with open(args.output, "w", encoding="utf-8") as fh:
|
|
fh.write(arrays_einzeilig(json.dumps(faelle, ensure_ascii=False,
|
|
indent=2)))
|
|
fh.write("\n")
|
|
print("%s %d Ketten / %d JSON-Objekte geschrieben -> %s"
|
|
% (PREFIX, len(koepfe), len(faelle), args.output))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|