dxfmakros/lib/dxf_vf_abbild.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
dxf_vf_abbild.py - Abbild der VF_/GF_-Wrapperbloecke einer Testzeichnung:
Attribute und Bausteinliste je Kette.

Zweck: Regressionsnetz fuer Umbauten am Bau-Ablauf. Nach jedem Schritt der
Funktionstrennung (doc/TODO-plan-vf-interactive.md, Stufe 2) muss die
gebaute Geometrie unveraendert sein - "unveraendert" heisst hier: gleiche
Attribute (inklusive der reihenfolgeabhaengigen Komma-Listen L_VF_m/L_GF_m/
GF_WINKEL, die in der Zeichnung unsichtbar, im Sivas-Export aber falsch
waeren) UND gleiche Bausteinzusammensetzung des Blocks (faengt eine falsche
Blockvariante, die kein Attribut verraet).

Die Ketten werden ueber ihren Einfuegepunkt der Spec zugeordnet, nicht ueber
den Blocknamen: die laufende VF-Nummer beginnt in jeder neuen Zeichnung
wieder bei 1, der Kettenstart ist dagegen der stabile Anker.

Liest ASCII-DXF streamend (Gruppencode-Paare; ein ezdxf-DOM waere fuer die
~90 MB Testzeichnungen zu gross). Binaere DXF kann dieses Skript NICHT -
dafuer gibt es lib/dxf_scan_components.py.

Aufruf:
    python dxf_vf_abbild.py <dxf> [--spec tests/testdata/hundm05.json]
                            [--json abbild.json]
                            [--referenz tests/reference/hundm05_attribute.json]
                            [--schreibe-referenz]

Ohne --referenz wird das Abbild nur ausgegeben. Mit --referenz wird
verglichen (Exit-Code 1 bei Abweichung); mit --schreibe-referenz wird die
Referenz aus der Zeichnung neu geschrieben - nur nach geprueftem Lauf!
"""

import argparse
import collections
import io
import json
import os
import sys

PREFIX = "[dxf_vf_abbild]"

# Attribute, die von Lauf zu Lauf legitim wechseln: die laufende Nummer und
# der daraus gebildete Name. Sie sagen nichts ueber die Geometrie.
FLUECHTIG = ("ID", "Bezeichnung")

RUND = 3


def gruppen(pfad):
    """Alle (Gruppencode, Wert)-Paare der Datei."""
    with io.open(pfad, "r", encoding="utf-8", errors="replace") as fh:
        while True:
            code = fh.readline()
            if not code:
                return
            wert = fh.readline()
            if not wert:
                return
            code = code.strip()
            if not (code.isdigit() or (code.startswith("-")
                                       and code[1:].isdigit())):
                continue
            yield int(code), wert.rstrip("\n").rstrip("\r")


def lese_bloecke(pfad, praefixe):
    """(modelspace_inserts, blockinhalte) aus einer ASCII-DXF.

    modelspace_inserts: INSERTs der ENTITIES-Sektion mit passendem Praefix,
    je mit Einfuegepunkt, Handle und den zugehoerigen ATTRIB-Werten.
    blockinhalte: {Blockname: [(Baustein, x, y, z, rotation), ...]} aus der
    BLOCKS-Sektion.
    """
    sektion = None
    block_name = None
    ent = None
    ms = []
    blockinhalt = collections.defaultdict(list)
    letzter_insert = None
    erwarte_sektion = False
    erwarte_block_name = False

    for code, wert in gruppen(pfad):
        if code == 0:
            if ent is not None:
                _entity_ablegen(ent, sektion, block_name, ms, blockinhalt,
                                praefixe)
                if ent["typ"] == "INSERT" and not block_name:
                    letzter_insert = ent
                elif ent["typ"] == "ATTRIB" and letzter_insert is not None:
                    letzter_insert.setdefault("attribs", {})[
                        ent.get("tag", "?")] = ent.get("wert", "")
            ent = None
            if wert == "SECTION":
                erwarte_sektion = True
            elif wert == "ENDSEC":
                sektion = None
            elif wert == "BLOCK":
                erwarte_block_name = True
                ent = {"typ": "BLOCK"}
            elif wert == "ENDBLK":
                block_name = None
            elif wert in ("INSERT", "ATTRIB"):
                ent = {"typ": wert}
            continue

        if erwarte_sektion and code == 2:
            sektion = wert
            erwarte_sektion = False
            continue
        if ent is None:
            continue
        if ent["typ"] == "BLOCK" and code == 2 and erwarte_block_name:
            block_name = wert
            erwarte_block_name = False
            ent = None
            continue

        if code == 2:
            # Bei ATTRIB ist Gruppencode 2 der TAG, bei INSERT der Blockname.
            if ent["typ"] == "ATTRIB":
                ent["tag"] = wert
            else:
                ent["name"] = wert
        elif code == 1 and ent["typ"] == "ATTRIB":
            ent["wert"] = wert
        elif code == 10:
            ent["x"] = float(wert)
        elif code == 20:
            ent["y"] = float(wert)
        elif code == 30:
            ent["z"] = float(wert)
        elif code == 50:
            ent["rot"] = float(wert)
        elif code == 5:
            ent["handle"] = wert

    if ent is not None:
        _entity_ablegen(ent, sektion, block_name, ms, blockinhalt, praefixe)
    return ms, blockinhalt


def _entity_ablegen(ent, sektion, block_name, ms, blockinhalt, praefixe):
    if ent["typ"] != "INSERT":
        return
    name = ent.get("name", "")
    if block_name:
        blockinhalt[block_name].append(
            (name, round(ent.get("x", 0.0), RUND), round(ent.get("y", 0.0), RUND),
             round(ent.get("z", 0.0), RUND), round(ent.get("rot", 0.0), RUND)))
    elif sektion == "ENTITIES" and any(name.startswith(p) for p in praefixe):
        ms.append(ent)


def spec_startpunkte(spec_datei):
    """{spec_id: [x, y, z]} aus einer Spec-Datei (flaches JSON)."""
    if not spec_datei or not os.path.isfile(spec_datei):
        return {}
    with io.open(spec_datei, encoding="utf-8") as fh:
        daten = json.load(fh)
    return dict((o["spec_id"], o["start_punkt"]) for o in daten
                if "spec_id" in o)


def abbild(dxf, spec_datei=None, praefixe=("VF_", "GF_")):
    ms, blocks = lese_bloecke(dxf, praefixe)
    starts = spec_startpunkte(spec_datei)
    raus = collections.OrderedDict()
    for e in sorted(ms, key=lambda x: x.get("x", 0.0)):
        name = e.get("name")
        punkt = [e.get("x"), e.get("y"), e.get("z")]
        kette = _kette_zuordnen(punkt, starts) or name
        inhalt = blocks.get(name, [])
        raus[kette] = collections.OrderedDict([
            ("block", name),
            ("insert", [round(v, RUND) if v is not None else None
                        for v in punkt]),
            ("bausteine", collections.OrderedDict(
                sorted(collections.Counter(b[0] for b in inhalt).items()))),
            ("bausteine_gesamt", len(inhalt)),
            ("attribs", collections.OrderedDict(
                (k, v) for k, v in sorted(e.get("attribs", {}).items())
                if k not in FLUECHTIG)),
        ])
    return raus


def _kette_zuordnen(punkt, starts, toleranz=1.0):
    """Kette ueber den Einfuegepunkt der Spec zuordnen - der Blockname
    taugt nicht: die VF-Nummer beginnt in jeder Zeichnung neu bei 1."""
    for sid, p in starts.items():
        if all(abs(a - b) <= toleranz for a, b in zip(punkt[:3], p[:3])):
            return sid
    return None


def vergleiche(ist, soll):
    """Liste von Abweichungstexten (leer = gleich)."""
    fehler = []
    for kette in sorted(set(ist) | set(soll)):
        if kette not in ist:
            fehler.append("%s: in der Zeichnung nicht gebaut" % kette)
            continue
        if kette not in soll:
            fehler.append("%s: nicht in der Referenz (neue Kette? Referenz "
                          "mit --schreibe-referenz auffrischen)" % kette)
            continue
        a, b = ist[kette], soll[kette]
        for feld in ("bausteine", "bausteine_gesamt"):
            if a[feld] != b[feld]:
                fehler.append("%s: %s ist %s, Referenz %s"
                              % (kette, feld, a[feld], b[feld]))
        for tag in sorted(set(a["attribs"]) | set(b["attribs"])):
            wa, wb = a["attribs"].get(tag), b["attribs"].get(tag)
            if wa != wb:
                fehler.append("%s: Attribut %s ist %r, Referenz %r"
                              % (kette, tag, wa, wb))
        for achse, (wa, wb) in enumerate(zip(a["insert"], b["insert"])):
            if wa is None or wb is None or abs(wa - wb) > 1.0:
                fehler.append("%s: Einfuegepunkt Achse %d ist %s, Referenz %s"
                              % (kette, achse, wa, wb))
    return fehler


def main():
    p = argparse.ArgumentParser(description=__doc__.split("\n")[1])
    p.add_argument("dxf", help="Testzeichnung (ASCII-DXF)")
    p.add_argument("--spec", help="Spec-JSON zur Ketten-Zuordnung")
    p.add_argument("--json", help="Abbild hierhin schreiben")
    p.add_argument("--referenz", help="Referenz-Abbild zum Vergleich")
    p.add_argument("--schreibe-referenz", action="store_true",
                   help="Referenz aus dieser Zeichnung neu schreiben")
    p.add_argument("--praefix", default="VF_,GF_",
                   help="Blocknamen-Praefixe, komma-getrennt")
    args = p.parse_args()

    if not os.path.isfile(args.dxf):
        print("%s FEHLER: %s nicht gefunden" % (PREFIX, args.dxf))
        return 1
    groesse = os.path.getsize(args.dxf) / (1024.0 * 1024.0)
    print("%s lese %s (%.1f MB) ..." % (PREFIX, args.dxf, groesse))

    ist = abbild(args.dxf, args.spec, tuple(args.praefix.split(",")))
    print("%s %d Kette(n): %s" % (PREFIX, len(ist), ", ".join(ist)))

    if args.json:
        _schreibe(args.json, ist)
        print("%s Abbild -> %s" % (PREFIX, args.json))

    if args.schreibe_referenz:
        if not args.referenz:
            print("%s FEHLER: --schreibe-referenz braucht --referenz" % PREFIX)
            return 1
        _schreibe(args.referenz, ist)
        print("%s Referenz neu geschrieben -> %s" % (PREFIX, args.referenz))
        return 0

    if args.referenz:
        if not os.path.isfile(args.referenz):
            print("%s FEHLER: Referenz %s fehlt" % (PREFIX, args.referenz))
            return 1
        with io.open(args.referenz, encoding="utf-8") as fh:
            soll = json.load(fh)
        fehler = vergleiche(ist, soll)
        if fehler:
            print("%s %d ABWEICHUNG(EN):" % (PREFIX, len(fehler)))
            for f in fehler:
                print("   - %s" % f)
            return 1
        print("%s Geometrie unveraendert (Attribute und Bausteine je Kette "
              "gleich der Referenz)" % PREFIX)
    return 0


def _schreibe(pfad, daten):
    verz = os.path.dirname(os.path.abspath(pfad))
    if verz and not os.path.isdir(verz):
        os.makedirs(verz)
    with io.open(pfad, "w", encoding="utf-8", newline="\r\n") as fh:
        fh.write(json.dumps(daten, indent=2, ensure_ascii=False))
        fh.write("\n")


if __name__ == "__main__":
    sys.exit(main())