basemap/werkzeug/schema-bauen.py
Der Schema-Erzeuger. Er verändert Attribute: er setzt die Negativliste als exclude_when in alle OSM-Quelleinträge und mischt sie in bestehende hinein.
gepflegt223 Zeilen8386 Bytenach 4.6(b) geschuldet
SHA-256 32021e71aba00af9de3b4952b66e370fd04985fcdc03a7ff79e1b66372908092
"""Setzt die Tag-Negativliste in jeden OSM-Quelleintrag des uebernommenen Schemas.
Aufruf ueber schema-bauen.sh. Reine Textarbeit, ohne YAML-Bibliothek: das
uebernommene Schema lebt von Ankern und Verweisen, und jede Bibliothek loest
die beim Zurueckschreiben auf. Das Ergebnis waere ein Schema, das sich nicht
mehr mit dem Original vergleichen laesst -- und der Vergleich ist nach ADR-0002
die Grundlage dafuer, Abweichungen bewusst nachzuhalten.
Die Regel ist eng: nur Zeilen, die exakt QUELLZEILE_OSM lauten, beginnen einen
betroffenen Eintrag. Attributbezogene exclude_when (tiefer eingerueckt) werden
nicht angefasst.
"""
import os
import re
import sys
# Einrueckungen des uebernommenen Schemas. Sie sind Vertrag, nicht Geschmack:
# die Negativliste wird per Text eingesetzt, und eine falsche Tiefe erzeugt
# stilles Unheil statt eines Fehlers.
TIEFE_LISTE = 2 # Schluessel unter "negativliste:" in tag-negativliste.yml
TIEFE_EINTRAG = 4 # Schluessel eines Feature-Eintrags, z.B. " exclude_when:"
TIEFE_REGEL = 6 # Schluessel innerhalb eines exclude_when
QUELLZEILE_OSM = ' - source: osm'
EXCLUDE_ZEILE = ' ' * TIEFE_EINTRAG + 'exclude_when:'
MARKE = '# XAPE-Tag-Negativliste -- erzeugt aus tag-negativliste.yml, nicht von Hand aendern'
# Zeilen des uebernommenen Schemas, die ersetzt werden. Jede traegt ihren
# Ersatz und wird gezaehlt: bleibt eine aus, ist das Schema stromaufwaerts
# umbenannt worden, und eine Zusicherung aus ABWEICHUNGEN.md gilt still nicht
# mehr.
ERSETZUNGEN = [
(
'examples: shortbread.spec.yml',
'examples: xape.spec.yml',
# Ohne den Verweis laeuft "planetiler verify" ins Leere. Die Beispiele
# stammen unveraendert vom uebernommenen Schema -- laufen sie gegen
# xape.yml durch, ist belegt, dass die Negativliste ausser marina und
# harbour nichts anfasst.
),
(
'schema_name: ',
'schema_name: XAPE Basemap',
# Der Name landet in den Metadaten des Archivs und damit in dem, was
# ausgeliefert wird. Ein Erzeugnis, das die Objektarten des Vorbilds
# absichtlich nicht enthaelt, darf dessen Namen nicht fuehren.
),
(
'schema_description: ',
'schema_description: Abgeleitet von Shortbread (Planetiler v0.10.2). '
'Die von XAPE selbst gefuehrten Objektarten sind kategorisch '
'ausgeschlossen, siehe tag-negativliste.yml.',
),
]
def negativliste_lesen(pfad):
"""Die Schluessel unter 'negativliste:' als Zeilen in TIEFE_LISTE."""
zeilen = []
im_block = False
for roh in open(pfad, encoding='utf-8'):
zeile = roh.rstrip('\n')
if not im_block:
if zeile.strip() == 'negativliste:':
im_block = True
continue
if zeile.strip() == '' or zeile.lstrip().startswith('#'):
continue
if not zeile.startswith(' ' * TIEFE_LISTE):
break
zeilen.append(zeile)
if not zeilen:
sys.exit(f'keine Eintraege unter negativliste: in {pfad}')
return zeilen
def regelschluessel(zeilen, tiefe):
"""Die Schluessel der obersten Ebene eines Blocks, zur Kollisionspruefung."""
return {
z.split(':', 1)[0].strip().strip('"\'')
for z in zeilen
if z.startswith(' ' * tiefe)
and not z.startswith(' ' * (tiefe + 1))
and ':' in z
}
def eintragsende(zeilen, start):
"""Index hinter dem Feature-Eintrag, der bei start beginnt."""
i = start + 1
while i < len(zeilen):
if re.match(r'^ {0,2}- ', zeilen[i]) or re.match(r'^\S', zeilen[i]):
break
i += 1
return i
def kopfzeile_ersetzen(zeile, zaehler):
"""Eine der ERSETZUNGEN anwenden, falls die Zeile passt."""
for praefix, ersatz in ((e[0], e[1]) for e in ERSETZUNGEN):
if zeile.strip() == praefix.strip() or (
praefix.endswith(' ') and zeile.startswith(praefix)
):
zaehler[praefix] = zaehler.get(praefix, 0) + 1
return ersatz
return zeile
def schema_umschreiben(quelle, regelblock):
"""Negativliste in jeden OSM-Quelleintrag setzen. Gibt (Zeilen, Zahlen) zurueck."""
aus = []
getroffen = 0
gemischt = 0
zaehler = {}
listen_keys = regelschluessel(regelblock, TIEFE_REGEL)
i = 0
while i < len(quelle):
zeile = kopfzeile_ersetzen(quelle[i], zaehler)
aus.append(zeile)
if quelle[i] != QUELLZEILE_OSM:
i += 1
continue
getroffen += 1
ende = eintragsende(quelle, i)
vorhanden = next(
(k for k in range(i + 1, ende) if quelle[k] == EXCLUDE_ZEILE), None
)
if vorhanden is None:
aus.append(EXCLUDE_ZEILE)
aus.extend(regelblock)
i += 1
continue
# Ein Mapping ist in diesem Schema bereits ein Oder ueber seine
# Schluessel, also wird hineingemischt statt danebengesetzt -- ein
# zweiter Schluessel exclude_when waere ein doppelter YAML-Schluessel.
gemischt += 1
aus.extend(quelle[i + 1:vorhanden + 1])
aus.extend(regelblock)
for k in range(vorhanden + 1, ende):
if quelle[k].startswith(' ' * TIEFE_REGEL):
key = quelle[k][TIEFE_REGEL:].split(':', 1)[0].strip().strip('"\'')
if key in listen_keys:
sys.exit(
f'Schluessel-Kollision in Zeile {k + 1}: "{key}" steht im '
f'uebernommenen exclude_when und in der Negativliste. Von '
f'Hand aufloesen -- ein Mischen wuerde eine der beiden '
f'Regeln still verlieren.'
)
aus.append(quelle[k])
i = ende
return aus, getroffen, gemischt, zaehler
def beispiele_zusammenfuehren(verzeichnis, ziel):
"""Uebernommene Beispiele plus eigene Proben in eine Datei."""
teile = [
'# ERZEUGT -- nicht von Hand aendern.',
'# shortbread.upstream.spec.yml (unveraendert) + xape-negativliste.spec.yml',
'examples:',
]
for name in ('shortbread.upstream.spec.yml', 'xape-negativliste.spec.yml'):
pfad = os.path.join(verzeichnis, name)
gesehen = False
for roh in open(pfad, encoding='utf-8'):
z = roh.rstrip('\n')
if not gesehen:
gesehen = z.strip() == 'examples:'
continue
teile.append(z)
if not gesehen:
sys.exit(f'kein examples: in {pfad}')
open(ziel, 'w', encoding='utf-8').write('\n'.join(teile) + '\n')
def main():
oben_pfad, listen_pfad, ziel_pfad = sys.argv[1:4]
regelblock = [' ' * TIEFE_REGEL + z for z in
[MARKE] + [z[TIEFE_LISTE:] for z in negativliste_lesen(listen_pfad)]]
quelle = open(oben_pfad, encoding='utf-8').read().split('\n')
aus, getroffen, gemischt, zaehler = schema_umschreiben(quelle, regelblock)
if getroffen == 0:
sys.exit('kein einziger OSM-Quelleintrag gefunden -- Schema-Aufbau geaendert?')
for praefix, _, *_ in ERSETZUNGEN:
if zaehler.get(praefix, 0) != 1:
sys.exit(
f'"{praefix.strip()}" {zaehler.get(praefix, 0)} mal gefunden, '
f'erwartet genau einmal -- das uebernommene Schema hat den '
f'Schluessel umbenannt, und eine Zusicherung aus ABWEICHUNGEN.md '
f'gilt damit still nicht mehr.'
)
kopf = [
'# ERZEUGT -- nicht von Hand aendern.',
'# Quelle: schema/shortbread.upstream.yml (unveraendert uebernommen)',
'# plus: tag-negativliste.yml',
'# Erzeuger: werkzeug/schema-bauen.sh',
f'# Die Negativliste steht an {getroffen} OSM-Quelleintraegen, davon {gemischt} '
'in ein vorhandenes exclude_when gemischt.',
'#',
'# Abweichungen vom uebernommenen Schema stehen in ABWEICHUNGEN.md.',
'',
]
open(ziel_pfad, 'w', encoding='utf-8').write('\n'.join(kopf + aus))
verzeichnis = os.path.dirname(ziel_pfad)
ziel_spec = os.path.join(verzeichnis, 'xape.spec.yml')
beispiele_zusammenfuehren(verzeichnis, ziel_spec)
print(f'{ziel_spec}: Beispiele zusammengefuehrt')
print(f'{ziel_pfad}: {getroffen} Quelleintraege belegt, {gemischt} gemischt')
if __name__ == '__main__':
main()
basemap/werkzeug/kachel-pruefung.py
Liest die fertigen Kacheln ohne Bibliothek und ohne das Bauwerkzeug aus und prüft Bestand, Wache und Wirkung. Verändert nichts.
gepflegt766 Zeilen31028 Bytefreiwillig, nicht geschuldet
SHA-256 30befaca69af4cf9839dfb8ac18bc46502213d0c9caa67debf5f7d53c488af63
"""Nachweis, dass der Ausschluss kategorisch erfolgt ist.
Liest ein PMTiles-Archiv aus und belegt vier Dinge:
1. BESTAND: `pier`, `dock`, `ferry_terminal` und `lighthouse` sind vorhanden.
Ein Ausschluss, der die halbe Karte mitnimmt, ist kein enger Ausschluss --
und eine Pruefung, die nur nach dem Verbotenen sucht, ist auch dann gruen,
wenn das Archiv leer ist.
2. GEGENPROBE: zu jeder geforderten Objektart laeuft im selben Kacheldurchgang
ein Gegenstueck mit, das per Konstruktion nicht vorkommen kann. Dieselbe
Regel, die oben zaehlt, muss jedes davon als FEHLT melden. Ohne diesen
Schritt waere Punkt 1 eine Eigenschaft des Quelltextes, die niemand am
einzelnen Lauf nachlesen kann -- mit ihm steht in jedem Protokoll, dass
schon eine einzelne fehlende Art den Rueckgabewert 1 erzwingt.
3. WACHE: kein Merkmal traegt den Wert `marina`.
4. WIRKUNG (nur mit --vergleich): ein Bau ohne Negativliste enthaelt MEHR
Merkmale als dieser. Ist die Zahl gleich, hat die Liste nichts bewirkt.
Zugleich darf sie je Pflicht-Objektart hoechstens SCHRANKE_ANTEIL davon
mitnehmen -- eine Liste, die eine Objektart ausduennt statt sie zu lassen,
ist nicht eng gefasst, auch wenn ein Rest uebrigbleibt.
WARUM PUNKT 3 NOETIG IST -- GEMESSEN AM 15.08.2026
Punkt 2 allein beweist nichts. Das uebernommene Schema kopiert `leisure` nicht in
die Kacheln: ein Hafenbecken erscheint dort als `kind=water`, ein Marina-Gelaende
als `kind=<landuse>`. Der Wert `marina` kommt in der Ausgabe deshalb NIE vor --
mit Negativliste wie ohne. Nachgemessen an zwei vollstaendigen Baulaeufen: beide
null Treffer, aber 7.405.127 gegen 7.405.650 Merkmale. Die Liste entfernt 523
Merkmale, und nur diese Differenz belegt ihre Wirkung.
Punkt 2 bleibt trotzdem stehen, als Wache: nimmt eine kuenftige Schemafassung den
Wert doch in die Ausgabe auf, faellt es hier auf.
WARUM DAS FORMAT SELBST GELESEN WIRD
Der Nachweis ist nach ADR-0002 das Beweismittel dafuer, dass der Ausschluss
kategorisch war. Ein Beweismittel, das mit demselben Werkzeug arbeitet, das die
Kacheln erzeugt hat, prueft die Baukette gegen sich selbst. Deshalb liest dieses
Programm PMTiles und Vector Tiles ohne Bibliothek und ohne das Bauwerkzeug --
nur Standardbibliothek.
Aufruf:
python3 kachel-pruefung.py <archiv.pmtiles> [--protokoll <datei.md>]
[--vergleich <archiv-ohne-negativliste.pmtiles>]
[--werkzeug <text>] [--quellenstand <text>]
python3 kachel-pruefung.py <archiv.pmtiles> --gegenprobe
python3 kachel-pruefung.py --help
--gegenprobe fuehrt denselben Lauf mit einer um die Gegenstuecke ERWEITERTEN
Pflichtliste: die echten Objektarten sind im Archiv, die Gegenstuecke fehlen per
Konstruktion. Rueckgabewert 1 ist dort das erwartete Ergebnis und belegt, dass
schon eine einzelne fehlende Art die Pruefung durchfallen laesst, waehrend alle
uebrigen vorhanden sind. Ein 0 hiesse, die Bestandspruefung schlaegt nicht an.
Werkzeugfassung und Quellenstand werden aus den Metadaten des geprueften Archivs
GELESEN, nicht behauptet. --werkzeug und --quellenstand ueberschreiben sie fuer
Archive, die sie nicht tragen; fehlt beides, faellt der Nachweis durch. Ein
Protokoll ohne Herkunft belegt nichts.
Das Protokoll traegt Datum, Werkzeugfassung, Quellenstand und Befund und wird
auch dann geschrieben, wenn der Nachweis durchfaellt -- gerade dann.
Rueckgabe 0, wenn der Nachweis traegt, sonst 1.
"""
import gzip
import json
import struct
import sys
from datetime import datetime
# --- Was gelten muss --------------------------------------------------------
# Kommt kein einziges Mal vor. Paare aus (Attributschluessel, Wert).
VERBOTEN = [
('leisure', 'marina'),
('kind', 'marina'),
]
# Muss vorkommen, sonst ist der Ausschluss nicht eng gefasst, sondern breit.
# Je Eintrag: Anzeigename, Ebene, optional (Schluessel, Wert).
PFLICHT = [
('pier (Linien)', 'pier_lines', None),
('pier (Flaechen)', 'pier_polygons', None),
('dock', 'water_polygons', ('kind', 'dock')),
('ferry_terminal', 'public_transport', ('kind', 'ferry_terminal')),
('lighthouse', 'pois', ('man_made', 'lighthouse')),
]
# Ein Wert, den kein Objekt tragen kann -- er ist hier erfunden und kommt in
# keiner Quelle vor. Er ist der Kern der Gegenprobe: ein Pflichteintrag, der
# garantiert fehlt.
GEGENPROBE_WERT = 'xape-gegenprobe-kommt-im-bestand-nicht-vor'
# Obergrenze fuer den Verlust je Pflicht-Objektart im Wirkungsvergleich. Die
# Bestandspruefung fragt nur, OB eine Art noch da ist; sie ist damit blind
# dagegen, dass die Negativliste neun von zehn Molen mitnimmt.
#
# GEMESSEN AM 16.08.2026, xape-osm gegen den Lauf mit geleerter Liste:
# ferry_terminal 649 statt 655 Verlust 6 0,92 % <- groesster Wert
# pier (Flaechen) 2593 statt 2596 Verlust 3 0,12 %
# pier (Linien), dock, lighthouse Verlust 0 0,00 %
#
# Fuenf Prozent lassen diese Randfaelle mit Abstand durch und schlagen an, lange
# bevor eine Objektart ausgeduennt ist. Ein Prozent waere zu knapp: der groesste
# gemessene Wert lieferte dann bei jeder kleinen Bestandsaenderung einen
# Fehlalarm, und ein Fehlalarm hier loescht das Archiv.
SCHRANKE_ANTEIL = 0.05
def gegenprobe_zu(eintraege):
"""Zu jedem Pflichteintrag ein Gegenstueck, das nicht vorkommen kann.
Dieselbe Ebene, derselbe Attributschluessel -- nur ein Wert, den es nicht
gibt. Damit laeuft die Gegenprobe durch genau denselben Zaehl- und
Bewertungsweg wie die echte Pruefung und nicht durch einen nachgebauten.
"""
return [
(f'{name} (Gegenprobe)', ebene,
((paar[0] if paar else 'kind'), GEGENPROBE_WERT))
for name, ebene, paar in eintraege
]
GEGENPROBE = gegenprobe_zu(PFLICHT)
# --- PMTiles v3 -------------------------------------------------------------
KOPF_LAENGE = 127
def _varint(daten, i):
wert = 0
verschiebung = 0
while True:
b = daten[i]
i += 1
wert |= (b & 0x7F) << verschiebung
if not b & 0x80:
return wert, i
verschiebung += 7
def kopf_lesen(daten):
if daten[:7] != b'PMTiles':
sys.exit('kein PMTiles-Archiv')
if daten[7] != 3:
sys.exit(f'PMTiles-Fassung {daten[7]}, erwartet 3')
f = struct.unpack_from('<QQQQQQQQQQQ', daten, 8)
return {
'wurzel_offset': f[0], 'wurzel_laenge': f[1],
'meta_offset': f[2], 'meta_laenge': f[3],
'blatt_offset': f[4], 'blatt_laenge': f[5],
'kachel_offset': f[6], 'kachel_laenge': f[7],
'innere_kompression': daten[97],
'kachel_kompression': daten[98],
'min_zoom': daten[100], 'max_zoom': daten[101],
}
def _entpacken(roh, art):
# 1 = keine, 2 = gzip. Andere Werte kommen in diesem Bau nicht vor.
return gzip.decompress(roh) if art == 2 else roh
def metadaten_lesen(daten, kopf):
"""Der JSON-Block, den PMTiles im Kopf adressiert.
Hier stehen Werkzeugfassung und Quellenstand -- im geprueften Erzeugnis
selbst. Sie von dort zu lesen statt sie als Argument entgegenzunehmen ist
der ganze Punkt: eine uebergebene Fassungsnummer ist eine Behauptung ueber
das Archiv, eine gelesene ist eine Aussage des Archivs.
"""
roh = daten[kopf['meta_offset']:kopf['meta_offset'] + kopf['meta_laenge']]
try:
return json.loads(_entpacken(roh, kopf['innere_kompression']))
except (OSError, ValueError):
# Ein Archiv ohne lesbaren Metadatenblock ist kein Grund zum Abbruch --
# es fehlt dann die Herkunft, und darueber entscheidet main().
return {}
def werkzeug_aus_metadaten(meta):
"""`Planetiler v0.10.2, tile-join v2.79.0` -- aus dem Archiv gelesen."""
teile = []
if meta.get('planetiler:version'):
teile.append(f"Planetiler v{meta['planetiler:version']}")
if meta.get('generator'):
# tile-join schreibt hier seine eigene Fassung samt Programmnamen.
teile.append(str(meta['generator']))
return ', '.join(teile)
def quellenstand_aus_metadaten(meta):
"""Der OSM-Stand, aus dem die Kacheln entstanden sind.
Die Sequenznummer der Geofabrik-Replikation ist die eine Angabe, an der
sich ein Lauf wiederholen laesst. Fehlt sie, gibt es keinen Quellenstand --
dann liefert diese Funktion einen leeren Text und kein Ungefaehr.
"""
seq = meta.get('planetiler:osm:osmosisreplicationseq')
if seq in (None, ''):
return ''
text = f'OSM-Replikation, Sequenznummer {seq}'
zeit = meta.get('planetiler:osm:osmosisreplicationtime')
if zeit:
text += f', Stand {zeit}'
url = meta.get('planetiler:osm:osmosisreplicationurl')
if url:
text += f' ({url})'
return text
def verzeichnis_lesen(roh):
"""Ein PMTiles-Verzeichnis in Eintraege (kachel_id, offset, laenge, lauf)."""
n, i = _varint(roh, 0)
ids, laeufe, laengen, offsets = [], [], [], []
letzte = 0
for _ in range(n):
d, i = _varint(roh, i)
letzte += d
ids.append(letzte)
for _ in range(n):
v, i = _varint(roh, i)
laeufe.append(v)
for _ in range(n):
v, i = _varint(roh, i)
laengen.append(v)
for k in range(n):
v, i = _varint(roh, i)
# 0 heisst: unmittelbar hinter dem vorigen Eintrag.
offsets.append(offsets[k - 1] + laengen[k - 1] if v == 0 and k > 0 else v - 1)
return list(zip(ids, offsets, laengen, laeufe))
def kachel_id_zu_zxy(kachel_id):
"""PMTiles-Kachelnummer in Zoom, Spalte, Zeile (Hilbert-Kurve)."""
z = 0
basis = 0
while True:
anzahl = (1 << z) * (1 << z)
if kachel_id < basis + anzahl:
break
basis += anzahl
z += 1
rest = kachel_id - basis
n = 1 << z
x = y = 0
s = 1
t = rest
while s < n:
rx = 1 & (t // 2)
ry = 1 & (t ^ rx)
if ry == 0:
if rx == 1:
x, y = s - 1 - x, s - 1 - y
x, y = y, x
x += s * rx
y += s * ry
t //= 4
s *= 2
return z, x, y
def alle_kacheln(daten, kopf):
"""(kachel_id, roher Kachelinhalt) fuer jeden eindeutigen Inhalt."""
wurzel = _entpacken(
daten[kopf['wurzel_offset']:kopf['wurzel_offset'] + kopf['wurzel_laenge']],
kopf['innere_kompression'])
offen = verzeichnis_lesen(wurzel)
gesehen = set()
while offen:
kachel_id, offset, laenge, lauf = offen.pop()
if lauf == 0:
blatt = _entpacken(
daten[kopf['blatt_offset'] + offset:
kopf['blatt_offset'] + offset + laenge],
kopf['innere_kompression'])
offen.extend(verzeichnis_lesen(blatt))
continue
if offset in gesehen:
continue
gesehen.add(offset)
roh = daten[kopf['kachel_offset'] + offset:
kopf['kachel_offset'] + offset + laenge]
yield kachel_id, _entpacken(roh, kopf['kachel_kompression'])
# --- Mapbox Vector Tile -----------------------------------------------------
def _felder(daten, ende=None):
"""(feldnummer, drahttyp, wert) je Feld. Wert ist bytes oder int."""
i = 0
ende = len(daten) if ende is None else ende
while i < ende:
kopf, i = _varint(daten, i)
feld, draht = kopf >> 3, kopf & 7
if draht == 0:
wert, i = _varint(daten, i)
elif draht == 2:
laenge, i = _varint(daten, i)
wert = daten[i:i + laenge]
i += laenge
elif draht == 5:
wert = daten[i:i + 4]
i += 4
elif draht == 1:
wert = daten[i:i + 8]
i += 8
else:
sys.exit(f'unbekannter Drahttyp {draht}')
yield feld, draht, wert
def _wert_lesen(roh):
for feld, _draht, wert in _felder(roh):
if feld == 1:
return wert.decode('utf-8', 'replace')
if feld in (4, 5):
return str(wert)
if feld == 6:
return str((wert >> 1) ^ -(wert & 1))
if feld == 7:
return 'true' if wert else 'false'
return ''
def kachel_lesen(roh):
"""{ebene: [ {schluessel: wert}, ... ]} fuer eine Kachel."""
ebenen = {}
for feld, _draht, wert in _felder(roh):
if feld != 3:
continue
name, schluessel, werte, merkmale = '', [], [], []
for f2, _d2, w2 in _felder(wert):
if f2 == 1:
name = w2.decode('utf-8', 'replace')
elif f2 == 3:
schluessel.append(w2.decode('utf-8', 'replace'))
elif f2 == 4:
werte.append(_wert_lesen(w2))
elif f2 == 2:
merkmale.append(w2)
ausgabe = []
for m in merkmale:
paare = {}
for f3, draht3, w3 in _felder(m):
if f3 != 2:
continue
# tags ist gepackt (Drahttyp 2) oder einzeln (Drahttyp 0).
zahlen = []
if draht3 == 2:
j = 0
while j < len(w3):
v, j = _varint(w3, j)
zahlen.append(v)
else:
zahlen.append(w3)
for a in range(0, len(zahlen) - 1, 2):
ki, vi = zahlen[a], zahlen[a + 1]
if ki < len(schluessel) and vi < len(werte):
paare[schluessel[ki]] = werte[vi]
ausgabe.append(paare)
ebenen.setdefault(name, []).extend(ausgabe)
return ebenen
# --- Die Pruefung -----------------------------------------------------------
def fehlende_arten(zaehler, eintraege):
"""Welche der geforderten Objektarten kein einziges Mal vorkommen.
Diese eine Regel entscheidet ueber den Rueckgabewert -- und sie entscheidet
ueber Pflichtliste und Gegenprobe gleichermassen. Genau deshalb belegt die
Gegenprobe etwas: sie laeuft nicht durch einen zweiten, nachgebauten Weg.
"""
return [name for name, _, _ in eintraege if zaehler.get(name, 0) == 0]
def pruefen(pfad, pflicht=PFLICHT, gegenprobe=GEGENPROBE):
daten = open(pfad, 'rb').read()
kopf = kopf_lesen(daten)
verstoesse = []
gefunden = {name: 0 for name, _, _ in list(pflicht) + list(gegenprobe)}
je_ebene = {}
kacheln = 0
merkmale = 0
# Nach Ebene vorsortiert: sonst laeuft je Merkmal die ganze Liste durch,
# und die Gegenprobe verdoppelte die Arbeit ueber Millionen Merkmale.
regeln_je_ebene = {}
for name, ebene, paar in list(pflicht) + list(gegenprobe):
regeln_je_ebene.setdefault(ebene, []).append((name, paar))
for kachel_id, roh in alle_kacheln(daten, kopf):
kacheln += 1
for ebene, eintraege in kachel_lesen(roh).items():
merkmale += len(eintraege)
je_ebene[ebene] = je_ebene.get(ebene, 0) + len(eintraege)
regeln = regeln_je_ebene.get(ebene, ())
for paare in eintraege:
for schluessel, wert in VERBOTEN:
if paare.get(schluessel) == wert:
z, x, y = kachel_id_zu_zxy(kachel_id)
verstoesse.append({
'objektart': f'{schluessel}={wert}',
'ebene': ebene,
'fundstelle': f'{z}/{x}/{y}',
'merkmal': paare,
})
for name, paar in regeln:
if paar is None or paare.get(paar[0]) == paar[1]:
gefunden[name] += 1
fehlend = fehlende_arten(gefunden, pflicht)
# Die Gegenprobe traegt, wenn JEDES Gegenstueck als fehlend erkannt wurde.
# Ein getroffenes Gegenstueck waere kein harmloser Zufall, sondern hiesse:
# der erfundene Wert steht im Bestand, und die Gegenprobe belegt nichts.
gegen_fehlend = fehlende_arten(gefunden, gegenprobe)
gegen_getroffen = [n for n, _, _ in gegenprobe if n not in gegen_fehlend]
return {
'archiv': pfad,
'kacheln': kacheln,
'merkmale': merkmale,
'zoom': f"{kopf['min_zoom']}-{kopf['max_zoom']}",
'metadaten': metadaten_lesen(daten, kopf),
'pflicht': list(pflicht),
'gegenprobe': list(gegenprobe),
'verstoesse': verstoesse,
'gefunden': gefunden,
'je_ebene': je_ebene,
'fehlend': fehlend,
'gegen_fehlend': gegen_fehlend,
'gegen_getroffen': gegen_getroffen,
'traegt': not verstoesse and not fehlend and not gegen_getroffen,
}
def wirkung_pruefen(mit, ohne):
"""Belegt, dass die Negativliste ueberhaupt etwas entfernt.
`ohne` ist ein Bau derselben Quelle mit geleerter Negativliste. Er muss mehr
Merkmale enthalten -- sonst hat die Liste nichts bewirkt, und der
Entlastungsbeweis nach ADR-0002 traegt nicht.
"""
# Zwei Archive sind nur vergleichbar, wenn sie dieselben Ebenen fuehren.
# Sonst wird das zusammengefuehrte Gesamtarchiv gegen einen reinen OSM-Lauf
# gestellt, und die fehlenden Landpolygone sehen aus wie eine Wirkung der
# Negativliste. Genau das ist beim Bauen dieses Werkzeugs passiert.
nur_mit = set(mit['je_ebene']) - set(ohne['je_ebene'])
nur_ohne = set(ohne['je_ebene']) - set(mit['je_ebene'])
if nur_mit or nur_ohne:
return {
'entfernt': 0,
'unterschiede': [],
'verlust': [],
'zu_weit': [],
'unvergleichbar': sorted(nur_mit | nur_ohne),
'traegt': False,
}
unterschiede = []
for ebene in sorted(set(mit['je_ebene']) | set(ohne['je_ebene'])):
a = mit['je_ebene'].get(ebene, 0)
b = ohne['je_ebene'].get(ebene, 0)
if a != b:
unterschiede.append((ebene, a, b, b - a))
entfernt = ohne['merkmale'] - mit['merkmale']
# Die andere Richtung: was hat die Liste von den Arten mitgenommen, die
# BLEIBEN sollen? Die Bestandspruefung fragt nur, ob noch eine Mole da ist.
# Erst dieser Vergleich sagt, ob der Ausschluss eng gefasst ist.
verlust = []
for name, _, _ in mit['pflicht']:
a = mit['gefunden'].get(name, 0)
b = ohne['gefunden'].get(name, 0)
anteil = (b - a) / b if b else 0.0
verlust.append((name, a, b, b - a, anteil))
zu_weit = [v for v in verlust if v[4] > SCHRANKE_ANTEIL]
return {
'entfernt': entfernt,
'unterschiede': unterschiede,
'verlust': verlust,
'zu_weit': zu_weit,
'unvergleichbar': [],
'traegt': entfernt > 0 and not zu_weit,
}
def protokoll_schreiben(pfad, ergebnis, angaben):
z = ergebnis['verstoesse']
zeilen = [
'# Pruefprotokoll der Tag-Negativliste',
'',
f"**Archiv:** `{ergebnis['archiv']}` ",
f"**Geprueft am:** {angaben['datum']} ",
f"**Befund:** {'BESTANDEN' if ergebnis['traegt'] else 'FEHLGESCHLAGEN'} ",
f"**Werkzeug:** {angaben['werkzeug']} "
f"_({angaben['werkzeug_herkunft']})_ ",
f"**Quellenstand:** {angaben['quellenstand']} "
f"_({angaben['quellenstand_herkunft']})_ ",
f"**Geprueft:** {ergebnis['kacheln']} Kacheln, "
f"{ergebnis['merkmale']} Merkmale, Zoom {ergebnis['zoom']}",
'',
'## Ausgeschlossen — muss null Mal vorkommen',
'',
]
if z:
zeilen.append('| Objektart | Ebene | Fundstelle |')
zeilen.append('| --- | --- | --- |')
for v in z[:50]:
zeilen.append(f"| `{v['objektart']}` | `{v['ebene']}` | `{v['fundstelle']}` |")
if len(z) > 50:
zeilen.append(f'| … | | {len(z) - 50} weitere |')
else:
for schluessel, wert in VERBOTEN:
zeilen.append(f'- `{schluessel}={wert}` — **0 Treffer**')
zeilen += ['', '## Erhalten — muss vorkommen', '']
for name, _, _ in ergebnis['pflicht']:
anzahl = ergebnis['gefunden'][name]
marke = '**0 — FEHLT**' if anzahl == 0 else f'{anzahl}'
zeilen.append(f'- {name}: {marke}')
if ergebnis['gegenprobe']:
zeilen += [
'',
'## Gegenprobe — schlaegt die Bestandspruefung an, wenn eine Art fehlt',
'',
'Zu jeder geforderten Objektart oben lief im selben Kacheldurchgang',
'ein Gegenstueck mit: dieselbe Ebene, derselbe Attributschluessel,',
f'aber der Wert `{GEGENPROBE_WERT}`,',
'den es im Bestand nicht gibt. Bewertet hat es dieselbe Regel, die',
'auch die echten Zaehlungen bewertet — kein nachgebauter zweiter Weg.',
'',
]
for name, _, _ in ergebnis['gegenprobe']:
anzahl = ergebnis['gefunden'][name]
marke = ('0 — als FEHLT erkannt' if anzahl == 0
else f'**{anzahl} — GETROFFEN, die Gegenprobe belegt nichts**')
zeilen.append(f'- {name}: {marke}')
erkannt = len(ergebnis['gegen_fehlend'])
zeilen += [
'',
f'{erkannt} von {len(ergebnis["gegenprobe"])} Gegenstuecken wurden als '
'fehlend erkannt, und jedes',
'einzelne davon haette diesen Lauf mit Rueckgabewert 1 beendet. Damit',
'ist an diesem Lauf abgelesen und nicht behauptet: fehlt eine der',
'geforderten Objektarten, faellt die Pruefung durch. Der Befund oben',
'ist eine Messung.',
]
w = ergebnis.get('wirkung')
if w:
zeilen += [
'',
'## Wirkung der Negativliste',
'',
f"Gegen einen Bau derselben Quelle mit **geleerter** Negativliste "
f"(`{ergebnis['vergleich']}`):",
'',
f"**{w['entfernt']} Merkmale entfernt.**"
+ ('' if w['traegt'] else ' — **das ist zu wenig, die Liste wirkt nicht**'),
'',
]
if w['unterschiede']:
zeilen.append('| Ebene | mit Liste | ohne Liste | Differenz |')
zeilen.append('| --- | --- | --- | --- |')
for ebene, a, b, d in w['unterschiede']:
zeilen.append(f'| `{ebene}` | {a} | {b} | {d:+d} |')
if w['verlust']:
zeilen += [
'',
'### Und was sie von den geforderten Arten mitgenommen hat',
'',
f'Obergrenze: **{SCHRANKE_ANTEIL * 100:g} %** je Objektart. Die',
'Bestandspruefung fragt nur, OB eine Art noch da ist — eine',
'Negativliste, die neun von zehn Molen mitnimmt und eine',
'uebriglaesst, bestuende sie. Erst diese Spalte sagt, ob der',
'Ausschluss eng gefasst ist.',
'',
'| Objektart | mit Liste | ohne Liste | Verlust | Anteil |',
'| --- | --- | --- | --- | --- |',
]
for name, a, b, d, anteil in w['verlust']:
marke = ' **— ueber der Schranke**' if anteil > SCHRANKE_ANTEIL else ''
zeilen.append(
f'| {name} | {a} | {b} | {d} | {anteil:.2%}{marke} |')
zeilen += [
'',
'Diese Differenz ist der eigentliche Wirkungsnachweis. Ein Scan nach dem',
'Wert `marina` findet in **beiden** Archiven nichts, weil das uebernommene',
'Schema `leisure` nicht in die Kacheln kopiert — ein Hafenbecken steht dort',
'als `kind=water`. Der Ausschluss ist an der Zahl der Merkmale ablesbar,',
'nicht am Namen der Objektart.',
]
zeilen += [
'',
'Erzeugt von `werkzeug/kachel-pruefung.py`. Das Programm liest PMTiles und',
'Vector Tiles ohne Bibliothek und ohne das Bauwerkzeug — ein Beweismittel,',
'das mit dem erzeugenden Werkzeug arbeitet, prueft die Baukette gegen sich',
'selbst.',
'',
]
open(pfad, 'w', encoding='utf-8').write('\n'.join(zeilen))
def argumente_lesen(argumente):
"""Streng, nicht kulant: ein unbekannter Schalter ist ein Abbruch.
Die frueherere Fassung ging die Liste mit `if a == '--protokoll'` durch und
ueberging alles Uebrige stillschweigend -- ein `--protokol` mit einem l hat
dann kein Protokoll geschrieben und nichts gesagt.
"""
if not argumente:
print(__doc__, file=sys.stderr)
sys.exit(1)
if argumente[0] in ('--help', '-h'):
print(__doc__)
sys.exit(0)
if argumente[0].startswith('-'):
sys.exit(f'erst der Archivpfad, dann die Schalter -- {argumente[0]}')
werte = {
'--protokoll': None,
'--vergleich': None,
'--werkzeug': None,
'--quellenstand': None,
}
gegenprobe_modus = False
i = 1
while i < len(argumente):
a = argumente[i]
if a in werte:
if i + 1 >= len(argumente):
sys.exit(f'{a} braucht einen Wert')
werte[a] = argumente[i + 1]
i += 2
elif a == '--gegenprobe':
gegenprobe_modus = True
i += 1
else:
sys.exit(f'unbekanntes Argument: {a}\n{__doc__}')
return argumente[0], werte, gegenprobe_modus
def herkunft_bestimmen(ergebnis, angabe_werkzeug, angabe_quellenstand):
"""Datum, Werkzeugfassung und Quellenstand fuer das Protokoll.
Gelesen wird aus dem geprueften Archiv; die Schalter ueberschreiben das fuer
Archive, die die Angabe nicht tragen. Was auf keinem der beiden Wege
zusammenkommt, wird nicht erfunden -- der Nachweis faellt dann durch, denn
ein Protokoll ohne Werkzeugfassung und Quellenstand belegt keinen Lauf.
"""
meta = ergebnis['metadaten']
aus_archiv = 'aus den Metadaten des Archivs gelesen'
werkzeug = angabe_werkzeug or werkzeug_aus_metadaten(meta)
quellenstand = angabe_quellenstand or quellenstand_aus_metadaten(meta)
return {
'datum': datetime.now().astimezone().isoformat(timespec='seconds'),
'werkzeug': werkzeug or 'FEHLT',
'werkzeug_herkunft': 'uebergeben' if angabe_werkzeug else aus_archiv,
'quellenstand': quellenstand or 'FEHLT',
'quellenstand_herkunft': (
'uebergeben' if angabe_quellenstand else aus_archiv),
'vollstaendig': bool(werkzeug) and bool(quellenstand),
}
def main():
archiv, werte, gegenprobe_modus = argumente_lesen(sys.argv[1:])
protokoll = werte['--protokoll']
vergleich = werte['--vergleich']
if gegenprobe_modus:
print(
'GEGENPROBE Die Pflichtliste ist um ihre Gegenstuecke ERWEITERT: die '
'geforderten Objektarten stehen weiter darin und sind im Archiv, je '
'ein Gegenstueck daneben fehlt per Konstruktion. Rueckgabewert 1 ist '
'das erwartete Ergebnis -- er belegt, dass schon eine einzelne '
'fehlende Art genuegt, waehrend alle uebrigen vorhanden sind.',
file=sys.stderr)
ergebnis = pruefen(archiv, pflicht=PFLICHT + GEGENPROBE, gegenprobe=[])
else:
ergebnis = pruefen(archiv)
angaben = herkunft_bestimmen(
ergebnis, werte['--werkzeug'], werte['--quellenstand'])
if not angaben['vollstaendig']:
print(
f'OHNE HERKUNFT Werkzeugfassung: {angaben["werkzeug"]}, '
f'Quellenstand: {angaben["quellenstand"]}. Das Archiv nennt sie '
f'nicht, und uebergeben wurde nichts -- ein Protokoll ohne beides '
f'belegt keinen Lauf. Mit --werkzeug bzw. --quellenstand nachreichen.',
file=sys.stderr)
ergebnis['traegt'] = False
wirkung = None
if vergleich:
wirkung = wirkung_pruefen(ergebnis, pruefen(vergleich))
ergebnis['wirkung'] = wirkung
ergebnis['vergleich'] = vergleich
if wirkung['unvergleichbar']:
print(
f'UNVERGLEICHBAR Die Archive fuehren verschiedene Ebenen: '
f'{", ".join(wirkung["unvergleichbar"])}. Beide Seiten muessen aus '
f'derselben Baustufe stammen -- das Gesamtarchiv gegen einen reinen '
f'OSM-Lauf zu stellen misst die fehlenden Landpolygone, nicht die '
f'Negativliste.', file=sys.stderr)
ergebnis['traegt'] = False
elif wirkung['entfernt'] <= 0:
print(
f'WIRKUNGSLOS {vergleich} enthaelt {wirkung["entfernt"]} Merkmale mehr. '
f'Die Negativliste entfernt nichts -- der Entlastungsbeweis nach '
f'ADR-0002 traegt so nicht.', file=sys.stderr)
ergebnis['traegt'] = False
elif wirkung['zu_weit']:
for name, a, b, d, anteil in wirkung['zu_weit']:
print(
f'ZU WEIT {name}: {d} von {b} Merkmalen fehlen mit Liste '
f'({anteil:.2%}, Schranke {SCHRANKE_ANTEIL * 100:g} %). Die Art ist '
f'zwar noch da, aber ausgeduennt -- der Ausschluss ist dann '
f'nicht mehr eng gefasst.', file=sys.stderr)
ergebnis['traegt'] = False
for n in ergebnis['gegen_getroffen']:
print(f'GEGENPROBE GETROFFEN {n} hat Treffer, obwohl der Wert erfunden '
f'ist -- die Gegenprobe belegt so nichts.', file=sys.stderr)
for v in ergebnis['verstoesse'][:20]:
print(f"VERSTOSS {v['objektart']} Ebene {v['ebene']} Kachel {v['fundstelle']} "
f"{json.dumps(v['merkmal'], ensure_ascii=False)}", file=sys.stderr)
if len(ergebnis['verstoesse']) > 20:
print(f"… und {len(ergebnis['verstoesse']) - 20} weitere", file=sys.stderr)
for n in ergebnis['fehlend']:
print(f'FEHLT {n} kommt im Archiv nicht vor -- der Ausschluss ist zu weit',
file=sys.stderr)
print(f"{ergebnis['kacheln']} Kacheln, {ergebnis['merkmale']} Merkmale, "
f"Zoom {ergebnis['zoom']}")
print(f"Werkzeug: {angaben['werkzeug']} ({angaben['werkzeug_herkunft']})")
print(f"Quellenstand: {angaben['quellenstand']} "
f"({angaben['quellenstand_herkunft']})")
for name, _, _ in ergebnis['pflicht']:
print(f" {name}: {ergebnis['gefunden'][name]}")
if ergebnis['gegenprobe']:
print(f"Gegenprobe: {len(ergebnis['gegen_fehlend'])} von "
f"{len(ergebnis['gegenprobe'])} erfundenen Objektarten als fehlend "
f"erkannt -- jede einzelne haette den Lauf mit 1 beendet")
if wirkung:
print(f"Wirkung der Negativliste: {wirkung['entfernt']} Merkmale entfernt")
for ebene, a, b, d in wirkung['unterschiede']:
print(f" {ebene}: {a} statt {b} ({d:+d})")
for name, a, b, d, anteil in wirkung['verlust']:
print(f" {name}: {a} statt {b} (Verlust {d}, {anteil:.2%})")
if protokoll:
protokoll_schreiben(protokoll, ergebnis, angaben)
print(f'Protokoll: {protokoll}')
if not ergebnis['traegt']:
sys.exit(1)
print('BESTANDEN')
if __name__ == '__main__':
main()