s.ayadi
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>
663 lines
28 KiB
Python
663 lines
28 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.
|
|
|
|
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
|
|
|
|
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"
|
|
|
|
# Antwort-Klartexte der Menues aus vf_linienzug.lsp (nur zur Annotation).
|
|
JA_NEIN = {"1": "Ja", "2": "Nein"}
|
|
SEITE = {"1": "links", "2": "rechts"}
|
|
VARIANTE = {"1": "aussen", "2": "innen"}
|
|
GF_MODUS = {"1": "ueber Zielhoehe", "2": "ueber Neigungswinkel"}
|
|
ENDPUNKT3 = {"1": "Ja, nur Motorstation", "2": "Nein, weiterbauen",
|
|
"3": "Ja, Motorstation + Kettenende am Zielpunkt"}
|
|
KETTENENDE3 = {"1": "Ja, mit ES-Element", "2": "Ja, ohne ES-Element",
|
|
"3": "Nein, weiterbauen"}
|
|
KETTENENDE2 = {"1": "Ja, Separator + ES-Element", "2": "Nein, weiterbauen"}
|
|
VERTEILUNG = {"1": "halbe Staustrecke am Ausgang", "2": "alles am Einlauf"}
|
|
IM_VF = {"1": "Horizontaler Foerderer", "2": "Vario-Kurve",
|
|
"3": "Auf/Ab-Foerderer"}
|
|
MENU_START = {"1": "Neue Linie: GF", "2": "Neue Linie: Ab/Auf VF",
|
|
"3": "Horizontal-VF", "4": "Neue Linie bis Kettenende"}
|
|
MENU_WEITER = {"1": "GF-Bogen", "2": "Neue Linie: GF",
|
|
"3": "Neue Linie: Ab/Auf VF", "4": "Horizontal-VF",
|
|
"5": "Neue Linie bis Kettenende"}
|
|
|
|
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
|
|
# ============================================================
|
|
|
|
class Journal(object):
|
|
"""Ein Journal (kind:payload je Zeile) in das Frage-Antwort-Protokoll
|
|
uebersetzen. Der Ablauf folgt vf-linienzug-modus (Modus 1):
|
|
|
|
Praeambel: Startpunkt, Starthoehe, AS-Element (Ja/Nein, Winkel, Seite)
|
|
Hauptschleife je Glied: STEP-Marker, Menuewahl, dann die Fragen des
|
|
gewaehlten Segmenttyps
|
|
Innerhalb einer VF-Einheit (Horizontal-VF / Linie): Separator vor/nach,
|
|
"Ist der Endpunkt der Foerderer?", danach Horizontal / Vario-Kurve /
|
|
Auf-Ab bis die Einheit schliesst
|
|
Kettenende: ES-Glied (Winkel, Seite)
|
|
"""
|
|
|
|
def __init__(self, roh):
|
|
self.tok = []
|
|
for zeile in roh.split(TRENNER):
|
|
if zeile:
|
|
art, _, wert = zeile.partition(":")
|
|
self.tok.append((art, wert))
|
|
self.pos = 0
|
|
self.out = []
|
|
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
|
|
|
|
# --- Ausgabe ---
|
|
|
|
def add(self, typ, kommentar, **felder):
|
|
eintrag = {"typ": typ}
|
|
eintrag.update(felder)
|
|
eintrag["kommentar"] = kommentar
|
|
self.out.append(eintrag)
|
|
|
|
def frage_str(self, kommentar_fmt, tabelle=None):
|
|
w = self.nimm("STR")
|
|
self.add("string", kommentar_fmt % (tabelle.get(w, "?") if tabelle
|
|
else w), wert=w)
|
|
return w
|
|
|
|
def frage_real(self, kommentar_fmt):
|
|
w = float(self.nimm("REAL"))
|
|
self.add("real", kommentar_fmt % w, wert=w)
|
|
return w
|
|
|
|
def frage_int(self, kommentar_fmt):
|
|
w = int(self.nimm("INT"))
|
|
self.add("int", kommentar_fmt % w, wert=w)
|
|
return w
|
|
|
|
def marker(self, label):
|
|
self.add("step", STEP_KOMMENTAR, wert=label)
|
|
|
|
def frage_linie(self, was):
|
|
"""Segmentlaenge; die gesnappte Fahrtrichtung steht nur beim ersten
|
|
Segment der Kette im Journal (vfl-in-abstand), danach erbt jedes
|
|
Segment die Richtung des Vorgaengers."""
|
|
dl = float(self.nimm("DL"))
|
|
if self.erster_dl and self.art() == "REAL":
|
|
hz = float(self.nimm("REAL"))
|
|
self.erster_dl = False
|
|
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)
|
|
else:
|
|
self.erster_dl = False
|
|
self.add("point_rel",
|
|
"getpoint %s: dL=%.1f mm (Fahrtrichtung vom "
|
|
"Vorgaenger geerbt)" % (was, dl), dL=dl)
|
|
return dl
|
|
|
|
# --- Ablauf ---
|
|
|
|
def dekodiere(self):
|
|
self.praeambel()
|
|
while self.pos < len(self.tok):
|
|
label = self.nimm("STEP")
|
|
self.marker(label)
|
|
if label == "ES":
|
|
self.glied_es(marker_gelesen=True)
|
|
continue
|
|
if self.erster_step:
|
|
tabelle, kopf = MENU_START, (
|
|
"Naechstes Element am Kettenanfang (1=Neue Linie GF, "
|
|
"2=Neue Linie Ab/Auf VF, 3=Horizontal-VF, 4=Neue Linie "
|
|
"bis Kettenende) -> %s")
|
|
else:
|
|
tabelle, kopf = MENU_WEITER, (
|
|
"Naechstes Element (1=GF-Bogen, 2=Neue Linie GF, "
|
|
"3=Neue Linie Ab/Auf VF, 4=Horizontal-VF, 5=Neue Linie "
|
|
"bis Kettenende) -> %s")
|
|
self.frage_str("getstring " + kopf, tabelle)
|
|
self.erster_step = False
|
|
if label == "Linie-GF":
|
|
self.glied_linie_gf()
|
|
elif label == "Linie-VF":
|
|
self.glied_linie_vf()
|
|
elif label == "Horizontal-VF":
|
|
self.glied_horizontal_vf()
|
|
elif label == "Linie":
|
|
self.glied_linie_kettenende()
|
|
elif label == "GF-Bogen":
|
|
self.frage_int("getint GF-Bogen Winkel (30/60/90 Grad) "
|
|
"-> %d")
|
|
self.frage_str("getstring GF-Bogen Seite (1=links, "
|
|
"2=rechts) -> %s", SEITE)
|
|
else:
|
|
raise ValueError("unbekanntes Glied: %s" % label)
|
|
if self.pos != len(self.tok):
|
|
raise ValueError("nicht alle Tokens verbraucht (%d von %d)"
|
|
% (self.pos, len(self.tok)))
|
|
return self.out
|
|
|
|
def praeambel(self):
|
|
punkt = [float(v) for v in self.nimm("PT").split(",")]
|
|
self.add("point_abs", "getpoint Startpunkt der Kette (absoluter "
|
|
"Weltpunkt, einziger Anker der ganzen Kette)", wert=punkt)
|
|
self.frage_real("getreal Hoehe (Z) Startpunkt -> %.1f mm")
|
|
if self.frage_str("getstring AS-Element setzen? (1=Ja, 2=Nein) "
|
|
"-> %s", JA_NEIN) == "1":
|
|
w = self.nimm("STR")
|
|
self.add("string", "getstring AS-Element Winkel (Wert direkt, "
|
|
"90 oder 30 Grad) -> %s Grad" % w, wert=w)
|
|
self.frage_str("getstring AS-Seite (1=links, 2=rechts) -> %s",
|
|
SEITE)
|
|
|
|
def glied_linie_gf(self):
|
|
self.frage_linie("Endpunkt der GF-Linie")
|
|
if self.frage_str("getstring Gefaelle festlegen (1=ueber Zielhoehe, "
|
|
"2=ueber Neigungswinkel) -> %s", GF_MODUS) == "2":
|
|
self.frage_real("getreal Neigungswinkel -> %.1f Grad")
|
|
else:
|
|
self.frage_real("getreal Zielhoehe (Z) Linienendpunkt "
|
|
"-> %.1f mm")
|
|
if self.frage_str("getstring Ist das Kettenende? (1=Ja mit ES, "
|
|
"2=Ja ohne ES, 3=Nein weiterbauen) -> %s",
|
|
KETTENENDE3) == "1":
|
|
self.glied_es()
|
|
|
|
def glied_horizontal_vf(self):
|
|
"""VF-Einheit mit HORIZONTALEM Anfangskoerper (winkel1=0 in
|
|
vfl-vf-einheit): dort stellt vfl-baue-horizontal-koerper im
|
|
Ziel-Modus die Separator-/Endpunkt-Fragen VOR der
|
|
Laengenberechnung."""
|
|
self.frage_linie("Endpunkt des horizontalen VF-Stuecks")
|
|
self.vf_einheit(auto_ende=False, erster_horizontal=True)
|
|
|
|
def glied_linie_vf(self):
|
|
""""Neue Linie: Ab/Auf VF" - Laenge, Zielhoehe, ggf. Winkelwahl
|
|
(vfl-vf-entscheidung), dann eine VF-Einheit mit GEWINKELTEM
|
|
Anfangskoerper. Zu dem Koerper wird nichts gefragt (vfl-vf-einheit
|
|
baut ihn direkt), die erste Frage der Einheit ist also die
|
|
Endpunkt-Frage der Fortsetzungsschleife."""
|
|
self.frage_linie("Endpunkt des Ab/Auf-VF-Segments")
|
|
self.frage_real("getreal Zielhoehe (Z) Linienendpunkt -> %.1f mm")
|
|
self.frage_winkelwahl()
|
|
self.vf_einheit(auto_ende=False, erster_horizontal=False)
|
|
|
|
def glied_linie_kettenende(self):
|
|
""""Neue Linie bis Kettenende": Laenge + Zielhoehe, danach entscheidet
|
|
vfl-segment-entscheidung selbst. Reine Gefaellestrecke -> ES-Element
|
|
ohne weitere Frage; VF-Einheit -> vf_einheit mit auto-ende."""
|
|
self.frage_linie("Endpunkt der Linie bis Kettenende (Ziel-XY der "
|
|
"Kette)")
|
|
self.frage_real("getreal Zielhoehe (Z) am Kettenende -> %.1f mm")
|
|
self.frage_winkelwahl()
|
|
if self.art() == "STEP" and self.label() == "ES":
|
|
self.glied_es()
|
|
elif self.art() == "STR":
|
|
# vfl-segment-entscheidung liefert fuer typ="VF" immer einen
|
|
# Winkel > 0 (die VF-Winkelstufen beginnen bei 3 Grad), der erste
|
|
# Koerper ist hier also nie horizontal.
|
|
self.vf_einheit(auto_ende=True, erster_horizontal=False)
|
|
|
|
def glied_es(self, marker_gelesen=False):
|
|
if not marker_gelesen:
|
|
label = self.nimm("STEP")
|
|
if label != "ES":
|
|
raise ValueError("ES-Marker erwartet, %s gefunden" % label)
|
|
self.marker(label)
|
|
w = self.nimm("STR")
|
|
self.add("string", "getstring ES-Element Winkel (Wert direkt, 90 "
|
|
"oder 30 Grad) -> %s Grad" % w, wert=w)
|
|
self.frage_str("getstring ES-Seite (1=links, 2=rechts) -> %s", SEITE)
|
|
|
|
def frage_winkelwahl(self):
|
|
"""Winkelwahl (vfl-waehle-winkel) steht nur im Journal, wenn mehrere
|
|
Kandidatenwinkel gueltig waren - bei genau einem faellt die Frage
|
|
weg."""
|
|
if self.art() == "INT":
|
|
self.frage_int("getint Winkelwahl aus den gueltigen "
|
|
"Kandidaten (vfl-waehle-winkel) -> Option %d")
|
|
|
|
def horizontal_koerper(self):
|
|
"""vfl-baue-horizontal-koerper im Ziel-Modus: Separator vor/nach und
|
|
die Endpunkt-Frage stehen VOR der Laengenberechnung. Rueckgabe: die
|
|
Endpunkt-Antwort, die die Fortsetzungsschleife weiterverwendet."""
|
|
self.frage_str("getstring Separator VOR dem horizontalen Stueck? "
|
|
"(1=Ja, 2=Nein) -> %s", JA_NEIN)
|
|
self.frage_str("getstring Separator NACH dem horizontalen Stueck? "
|
|
"(1=Ja, 2=Nein) -> %s", JA_NEIN)
|
|
return self.frage_str("getstring Ist der Endpunkt der Foerderer? "
|
|
"(1=Ja nur Motorstation, 2=Nein weiterbauen, "
|
|
"3=Ja Motorstation + Kettenende) -> %s",
|
|
ENDPUNKT3)
|
|
|
|
def vario_kurve(self):
|
|
label = self.nimm("STEP")
|
|
if label != "Vario-Kurve":
|
|
raise ValueError("Vario-Kurve-Marker erwartet, %s" % label)
|
|
self.marker(label)
|
|
self.frage_int("getint Vario-Kurve Winkel (30/60/90 Grad, Wert "
|
|
"direkt) -> %d Grad")
|
|
self.frage_str("getstring Vario-Kurve Seite (1=links, 2=rechts) "
|
|
"-> %s", SEITE)
|
|
self.frage_str("getstring Vario-Kurve Variante (1=aussen, 2=innen) "
|
|
"-> %s", VARIANTE)
|
|
|
|
def vf_einheit(self, auto_ende, erster_horizontal):
|
|
"""Eine VF-Einheit: GF-Verteilung, erster Koerper, dann die
|
|
Fortsetzungsschleife (vfl-vf-einheit) bis Motorstation/Kettenende,
|
|
danach der Abschluss (vfl-vf-einheit-abschluss).
|
|
|
|
erster_horizontal spiegelt die Verzweigung ueber winkel1 in
|
|
vfl-vf-einheit: nur der horizontale Erstkoerper laeuft durch
|
|
vfl-baue-horizontal-koerper und fragt darum Separator vor/nach und
|
|
die Endpunkt-Frage schon vorab (deren Antwort die Schleife dann
|
|
uebernimmt, statt erneut zu fragen). Ein gewinkelter Erstkoerper
|
|
wird ohne jede Frage gebaut."""
|
|
self.frage_str("getstring GF-Verteilung (1=halbe Staustrecke am "
|
|
"Ausgang, 2=alles am Einlauf) -> %s", VERTEILUNG)
|
|
vorantwort = self.horizontal_koerper() if erster_horizontal else None
|
|
ziel_ende = False
|
|
es_gewuenscht = True
|
|
while True:
|
|
if vorantwort is None:
|
|
antwort = self.frage_str(
|
|
"getstring Ist der Endpunkt der Foerderer? (1=Ja nur "
|
|
"Motorstation, 2=Nein weiterbauen, 3=Ja Motorstation + "
|
|
"Kettenende) -> %s", ENDPUNKT3)
|
|
else:
|
|
antwort, vorantwort = vorantwort, None
|
|
if antwort == "1":
|
|
break
|
|
if antwort == "3":
|
|
es_gewuenscht = self.frage_str(
|
|
"getstring ES-Element setzen? (1=Ja, 2=Nein - Kette "
|
|
"endet direkt am Zielpunkt) -> %s", JA_NEIN) != "2"
|
|
self.frage_linie("Zielpunkt des Kettenendes")
|
|
self.frage_real("getreal Zielhoehe (Z) am Kettenende "
|
|
"-> %.1f mm")
|
|
self.frage_winkelwahl()
|
|
ziel_ende = True
|
|
break
|
|
wahl = self.frage_str("getstring Naechstes in der VF-Einheit "
|
|
"(1=Horizontaler Foerderer, "
|
|
"2=Vario-Kurve, 3=Auf/Ab-Foerderer) "
|
|
"-> %s", IM_VF)
|
|
if wahl == "1":
|
|
self.frage_linie("Endpunkt des horizontalen VF-Stuecks")
|
|
vorantwort = self.horizontal_koerper()
|
|
elif wahl == "2":
|
|
self.vario_kurve()
|
|
else:
|
|
self.frage_linie("Endpunkt des Auf/Ab-VF-Segments")
|
|
self.frage_real("getreal Zielhoehe (Z) am Segmentende "
|
|
"-> %.1f mm")
|
|
self.frage_winkelwahl()
|
|
if ziel_ende and not es_gewuenscht:
|
|
return
|
|
if auto_ende or ziel_ende:
|
|
self.glied_es()
|
|
return
|
|
if self.frage_str("getstring Ist das Kettenende? (1=Ja, Separator + "
|
|
"ES-Element, 2=Nein, weiterbauen) -> %s",
|
|
KETTENENDE2) == "1":
|
|
self.glied_es()
|
|
else:
|
|
self.frage_str("getstring Separator an dieser Stelle einfuegen? "
|
|
"(1=Ja, 2=Nein) -> %s", JA_NEIN)
|
|
|
|
|
|
# ============================================================
|
|
# 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()
|