458 lines
19 KiB
Python
458 lines
19 KiB
Python
"""Kartenbeschriftungen der Vorlage entfernen, bevor die Pipeline Geometrie daraus macht.
|
||
|
||
Spec: docs/superpowers/specs/2026-09-13-beschriftungen-design.md
|
||
Koordinaten in annotations.toml sind Welt-x und Welt-y in Metern, genau wie die Minimap sie anzeigt.
|
||
|
||
Wie in der Feinkalibrierung: ein Formfehler in der Datei bricht mit AnnotationError ab, damit kein
|
||
Eintrag still verschwindet; ein formal richtiger Eintrag, der inhaltlich nichts trifft, wird
|
||
gemeldet und übersprungen.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import tomllib
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
from scipy import ndimage as ndi
|
||
from shapely.geometry import Polygon, box
|
||
from shapely.strtree import STRtree
|
||
from skimage.color import rgb2hsv, rgb2lab
|
||
from skimage.draw import polygon as fill_polygon
|
||
from skimage.morphology import disk
|
||
|
||
from .classify import PALETTE
|
||
from .coords import SLICE, chunk_for_world, chunk_name
|
||
from .inpaint import RebuildStats, add_buildings, fill_nearest, is_thin, rebuild
|
||
|
||
KINDS = ("text_white", "text_red", "marker", "symbol", "street_name", "region")
|
||
MARGIN = {"text_white": 7, "text_red": 7, "marker": 3, "symbol": 4, "street_name": 1, "region": 0}
|
||
RED = (248, 112, 112)
|
||
HOLE_MAX_AREA = 400 # m²: größere eingeschlossene Flächen sind echte Karte
|
||
MIN_BOX_SIDE = 2.0 # m
|
||
FULL_SHARE = 0.60 # randvoll ab diesem Anteil der Farbpixel (vor dem Wachsen) am Rahmen
|
||
CUT_MIN_PX = 20 # angeschnitten ab einem verworfenen Kandidaten dieser Größe; nicht bei street_name,
|
||
# dessen Farbregel auch die Stadtfläche neben der Fahrbahn trifft
|
||
GENERATED_SHARE = 0.5 # Anteil der Grundfläche in ergänzten Rechtecken für source = "annotation"
|
||
THIN_WINDOW = 20 # m Rand um eine dünne Maske beim Auffüllen
|
||
GRID_STRIP = 10 # m beiderseits einer Gitterlinie beim Auffüllen
|
||
EIGHT = np.ones((3, 3), dtype=bool)
|
||
|
||
|
||
class AnnotationError(ValueError):
|
||
"""Formfehler in annotations.toml: der Lauf bricht ab."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Annotation:
|
||
index: int # 1-basiert, wie in allen Meldungen
|
||
name: str
|
||
kind: str
|
||
box: tuple[float, float, float, float] | None # x_min, y_min, x_max, y_max
|
||
polygon: tuple[tuple[float, float], ...] | None
|
||
reason: str = ""
|
||
|
||
@property
|
||
def outline(self) -> Polygon:
|
||
return Polygon(self.polygon) if self.polygon else box(*self.box)
|
||
|
||
@property
|
||
def chunk(self) -> str:
|
||
centre = self.outline.centroid
|
||
return chunk_name(*chunk_for_world(centre.x, centre.y)) or "?"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Keep:
|
||
index: int
|
||
box: tuple[float, float, float, float]
|
||
reason: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Grid:
|
||
spacing: float
|
||
width: float
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AnnotationList:
|
||
annotations: list[Annotation] = field(default_factory=list)
|
||
keep: list[Keep] = field(default_factory=list)
|
||
grid: Grid | None = None
|
||
|
||
|
||
# --- Liste lesen und prüfen -------------------------------------------------------------------
|
||
|
||
def load_annotations(path: Path) -> AnnotationList:
|
||
"""Liste aus der Datei; fehlt sie, gibt es keine Einträge und kein Gitternetz."""
|
||
if not path.exists():
|
||
return AnnotationList()
|
||
return parse_annotations(path.read_text())
|
||
|
||
|
||
def parse_annotations(text: str) -> AnnotationList:
|
||
try:
|
||
data = tomllib.loads(text)
|
||
except tomllib.TOMLDecodeError as err:
|
||
raise AnnotationError(f"annotations.toml ist kein gültiges TOML: {err}") from err
|
||
unknown = sorted(set(data) - {"grid", "annotation", "keep"})
|
||
if unknown:
|
||
raise AnnotationError(f"annotations.toml: unbekannter Abschnitt {unknown[0]!r}, "
|
||
"erwartet [grid], [[annotation]] oder [[keep]]")
|
||
for section in ("annotation", "keep"):
|
||
if not isinstance(data.get(section, []), list):
|
||
raise AnnotationError(f"annotations.toml: {section} muss als [[{section}]] geschrieben sein")
|
||
return AnnotationList(
|
||
[_parse_annotation(i, entry) for i, entry in enumerate(data.get("annotation", []), start=1)],
|
||
[_parse_keep(i, entry) for i, entry in enumerate(data.get("keep", []), start=1)],
|
||
_parse_grid(data.get("grid")),
|
||
)
|
||
|
||
|
||
def _number(value) -> bool:
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||
|
||
|
||
def _point(value) -> bool:
|
||
return isinstance(value, list) and len(value) == 2 and all(_number(c) for c in value)
|
||
|
||
|
||
def _inside_slice(x_min, y_min, x_max, y_max) -> bool:
|
||
return x_min >= SLICE["x0"] and x_max <= SLICE["x1"] and y_min >= SLICE["y0"] and y_max <= SLICE["y1"]
|
||
|
||
|
||
def _parse_box(value, fail):
|
||
if not (isinstance(value, list) and len(value) == 2 and all(_point(p) for p in value)):
|
||
fail(f"Feld 'box' muss zwei Ecken [[x, y], [x, y]] enthalten, ist {value!r}")
|
||
(xa, ya), (xb, yb) = value
|
||
x_min, x_max = sorted((float(xa), float(xb)))
|
||
y_min, y_max = sorted((float(ya), float(yb)))
|
||
if x_max - x_min < MIN_BOX_SIDE or y_max - y_min < MIN_BOX_SIDE:
|
||
fail(f"Feld 'box' muss auf beiden Achsen mindestens {MIN_BOX_SIDE:g} m messen, ist {value!r}")
|
||
if not _inside_slice(x_min, y_min, x_max, y_max):
|
||
fail(f"Feld 'box' liegt außerhalb des Ausschnitts, ist {value!r}")
|
||
return x_min, y_min, x_max, y_max
|
||
|
||
|
||
def _parse_polygon(value, fail):
|
||
if not (isinstance(value, list) and len(value) >= 3 and all(_point(p) for p in value)):
|
||
fail(f"Feld 'polygon' muss mindestens 3 Punkte [x, y] enthalten, ist {value!r}")
|
||
shape = Polygon(value)
|
||
if not shape.is_valid or shape.area <= 0:
|
||
fail(f"Feld 'polygon' muss ein Vieleck ohne sich schneidende Kanten sein, ist {value!r}")
|
||
if not _inside_slice(*shape.bounds):
|
||
fail(f"Feld 'polygon' liegt außerhalb des Ausschnitts, ist {value!r}")
|
||
return tuple((float(x), float(y)) for x, y in value)
|
||
|
||
|
||
def _parse_annotation(index: int, entry) -> Annotation:
|
||
def fail(message: str):
|
||
raise AnnotationError(f"Beschriftung {index}: {message}")
|
||
|
||
if not isinstance(entry, dict):
|
||
fail("ist keine Tabelle")
|
||
for key in entry:
|
||
if key not in ("name", "kind", "box", "polygon", "reason"):
|
||
fail(f"Feld {key!r} gibt es nicht")
|
||
name = entry.get("name")
|
||
if not isinstance(name, str) or not name.strip():
|
||
fail("Feld 'name' fehlt oder ist leer")
|
||
kind = entry.get("kind")
|
||
if kind not in KINDS:
|
||
fail(f"unbekanntes kind {kind!r}, erlaubt sind {', '.join(KINDS)}")
|
||
reason = entry.get("reason", "")
|
||
if not isinstance(reason, str):
|
||
fail("reason muss Text sein")
|
||
if kind == "region":
|
||
if ("box" in entry) == ("polygon" in entry):
|
||
fail("region braucht genau eines der Felder 'box' und 'polygon'")
|
||
elif "polygon" in entry:
|
||
fail(f"Feld 'polygon' gibt es nur bei region, nicht bei {kind}")
|
||
elif "box" not in entry:
|
||
fail("Feld 'box' fehlt")
|
||
area = _parse_box(entry["box"], fail) if "box" in entry else None
|
||
outline = _parse_polygon(entry["polygon"], fail) if "polygon" in entry else None
|
||
return Annotation(index, name, kind, area, outline, reason)
|
||
|
||
|
||
def _parse_keep(index: int, entry) -> Keep:
|
||
def fail(message: str):
|
||
raise AnnotationError(f"Ausnahme {index}: {message}")
|
||
|
||
if not isinstance(entry, dict):
|
||
fail("ist keine Tabelle")
|
||
for key in entry:
|
||
if key not in ("box", "reason"):
|
||
fail(f"Feld {key!r} gibt es nicht")
|
||
if "box" not in entry:
|
||
fail("Feld 'box' fehlt")
|
||
reason = entry.get("reason")
|
||
if not isinstance(reason, str) or not reason.strip():
|
||
fail("Feld 'reason' fehlt oder ist leer")
|
||
return Keep(index, _parse_box(entry["box"], fail), reason)
|
||
|
||
|
||
def _parse_grid(value) -> Grid | None:
|
||
if value is None:
|
||
return None
|
||
|
||
def fail(message: str):
|
||
raise AnnotationError(f"Gitternetz: {message}")
|
||
|
||
if not isinstance(value, dict):
|
||
fail("muss als [grid] geschrieben sein")
|
||
for key in value:
|
||
if key not in ("spacing", "width"):
|
||
fail(f"Feld {key!r} gibt es nicht")
|
||
for key in ("spacing", "width"):
|
||
if key not in value:
|
||
fail(f"Feld {key!r} fehlt")
|
||
if not (_number(value[key]) and value[key] > 0):
|
||
fail(f"Feld {key!r} muss eine Zahl größer als 0 sein, ist {value[key]!r}")
|
||
return Grid(float(value["spacing"]), float(value["width"]))
|
||
|
||
|
||
# --- Masken -------------------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class FoundMask:
|
||
rows: slice
|
||
cols: slice
|
||
mask: np.ndarray # bool, Größe des Rahmens
|
||
applied: bool
|
||
warning: str | None = None
|
||
|
||
|
||
def _lab(rgb) -> np.ndarray:
|
||
return rgb2lab(np.array([[rgb]], dtype=np.float32) / 255.0)[0, 0]
|
||
|
||
|
||
def candidates(rgb: np.ndarray, kind: str) -> np.ndarray:
|
||
"""Kandidatenpixel der Farbregel einer Art in einem RGB-Ausschnitt (Tabelle in der Spec)."""
|
||
values = rgb.astype(np.int16)
|
||
low, high = values.min(axis=2), values.max(axis=2)
|
||
white = (low >= 236) & (high - low <= 14)
|
||
if kind == "text_white":
|
||
return white
|
||
if kind == "street_name":
|
||
return (low >= 200) & (high - low <= 20) # helle Schrift; Gebäudegrau (176) bleibt
|
||
lab = rgb2lab(rgb.astype(np.float32) / 255.0)
|
||
if kind == "text_red":
|
||
return np.linalg.norm(lab - _lab(RED), axis=2) < 18
|
||
if kind in ("marker", "symbol"):
|
||
hsv = rgb2hsv(rgb)
|
||
palette = np.stack([_lab(colour) for colour in PALETTE.values()])
|
||
far = np.linalg.norm(lab[:, :, None, :] - palette[None, None], axis=3).min(axis=2) >= 25
|
||
marker = ((hsv[..., 1] >= 0.6) & (hsv[..., 2] >= 0.5) & far) | (high <= 45)
|
||
return marker | white if kind == "symbol" else marker
|
||
raise ValueError(f"Art ohne Farbregel: {kind}")
|
||
|
||
|
||
def _pixel_bounds(annotation: Annotation, shape) -> tuple[int, int, int, int]:
|
||
"""Zeilen und Spalten, deren Pixelmitte im Hüllrechteck liegt; Pixelmitte (r, c) = Welt (x0 + c, y1 − r)."""
|
||
h, w = shape
|
||
x_min, y_min, x_max, y_max = annotation.outline.bounds
|
||
c0, c1 = max(0, math.ceil(x_min - SLICE["x0"])), min(w, math.floor(x_max - SLICE["x0"]) + 1)
|
||
r0, r1 = max(0, math.ceil(SLICE["y1"] - y_max)), min(h, math.floor(SLICE["y1"] - y_min) + 1)
|
||
return r0, r1, c0, c1
|
||
|
||
|
||
def annotation_mask(mosaic: np.ndarray, annotation: Annotation) -> FoundMask:
|
||
"""Maske eines Eintrags in fünf Schritten (Spec, Abschnitt „Erkennung je Art")."""
|
||
h, w = mosaic.shape[:2]
|
||
r0, r1, c0, c1 = _pixel_bounds(annotation, (h, w))
|
||
rows, cols = slice(r0, r1), slice(c0, c1)
|
||
shape = (r1 - r0, c1 - c0)
|
||
if annotation.kind == "region":
|
||
mask = np.zeros(shape, dtype=bool)
|
||
xs, ys = annotation.outline.exterior.coords.xy
|
||
rr, cc = fill_polygon(SLICE["y1"] - np.asarray(ys) - r0, np.asarray(xs) - SLICE["x0"] - c0, shape=shape)
|
||
mask[rr, cc] = True
|
||
if not mask.any():
|
||
return FoundMask(rows, cols, mask, False, "leer: das Vieleck trifft keine Pixelmitte")
|
||
return FoundMask(rows, cols, mask, True)
|
||
|
||
found = candidates(mosaic[rows, cols], annotation.kind)
|
||
comps, count = ndi.label(found, structure=EIGHT)
|
||
border = np.zeros(shape, dtype=bool)
|
||
border[0, :], border[-1, :] = r0 > 0, r1 < h # Kanten auf dem Ausschnittsrand zählen nicht
|
||
border[:, 0] |= c0 > 0
|
||
border[:, -1] |= c1 < w
|
||
touching = np.unique(comps[border & found])
|
||
touching = touching[touching > 0]
|
||
cut = int(max(ndi.sum(found, comps, touching))) if touching.size else 0
|
||
keep = found & ~np.isin(comps, touching)
|
||
if not keep.any():
|
||
return FoundMask(rows, cols, keep, False, f"leer: keine Pixel der Art {annotation.kind} im Rahmen")
|
||
grown = ndi.binary_dilation(keep, structure=disk(MARGIN[annotation.kind]).astype(bool))
|
||
holes = ndi.binary_fill_holes(grown) & ~grown
|
||
hole_ids, hole_count = ndi.label(holes)
|
||
if hole_count:
|
||
sizes = ndi.sum(holes, hole_ids, np.arange(1, hole_count + 1))
|
||
grown |= np.isin(hole_ids, 1 + np.flatnonzero(sizes <= HOLE_MAX_AREA))
|
||
share = float(keep.mean())
|
||
if share > FULL_SHARE:
|
||
return FoundMask(rows, cols, grown, False,
|
||
f"randvoll: Farbpixel füllen {round(share * 100)} % des Rahmens, Rahmen zu eng oder Art falsch")
|
||
cut_off = cut >= CUT_MIN_PX and annotation.kind != "street_name"
|
||
return FoundMask(rows, cols, grown, True, "angeschnitten: Rahmen prüfen" if cut_off else None)
|
||
|
||
|
||
def grid_mask(shape: tuple[int, int], grid: Grid) -> tuple[np.ndarray, int]:
|
||
"""Streifen der Breite grid.width mittig auf jeder Gitterlinie; dazu die Zahl der Linien."""
|
||
h, w = shape
|
||
mask = np.zeros(shape, dtype=bool)
|
||
lines = 0
|
||
for size, axis in ((w, 1), (h, 0)):
|
||
k = 0
|
||
while k * grid.spacing <= size:
|
||
lo = max(0, int(round(k * grid.spacing - grid.width / 2)))
|
||
hi = min(size, int(round(k * grid.spacing + grid.width / 2)))
|
||
if lo < hi:
|
||
if axis == 1:
|
||
mask[:, lo:hi] = True
|
||
else:
|
||
mask[lo:hi, :] = True
|
||
lines += 1
|
||
k += 1
|
||
return mask, lines
|
||
|
||
|
||
def _grid_strips(shape: tuple[int, int], grid: Grid):
|
||
"""Streifen von GRID_STRIP Metern beiderseits jeder Gitterlinie, damit nie das ganze Bild gefüllt wird."""
|
||
h, w = shape
|
||
for size, axis in ((w, 1), (h, 0)):
|
||
k = 0
|
||
while k * grid.spacing <= size:
|
||
lo = max(0, int(round(k * grid.spacing - grid.width / 2 - GRID_STRIP)))
|
||
hi = min(size, int(round(k * grid.spacing + grid.width / 2 + GRID_STRIP)))
|
||
if lo < hi:
|
||
yield (slice(None), slice(lo, hi)) if axis == 1 else (slice(lo, hi), slice(None))
|
||
k += 1
|
||
|
||
|
||
# --- Entfernen, Bericht, Markierung ----------------------------------------------------------
|
||
|
||
@dataclass
|
||
class EntryReport:
|
||
annotation: Annotation
|
||
applied: bool
|
||
warning: str | None = None
|
||
mask_area: int = 0
|
||
stats: RebuildStats | None = None
|
||
shared_with: int | None = None
|
||
|
||
|
||
@dataclass
|
||
class Removal:
|
||
labels: np.ndarray
|
||
generated: list[Polygon]
|
||
entries: list[EntryReport]
|
||
grid_lines: int
|
||
|
||
def report(self) -> str:
|
||
applied = sum(1 for e in self.entries if e.applied)
|
||
lines = [f"Beschriftungen: {applied} angewendet, {len(self.entries) - applied} übersprungen"
|
||
f" · Gitternetz: {self.grid_lines} Linien"]
|
||
for e in self.entries:
|
||
if not e.applied:
|
||
continue
|
||
a = e.annotation
|
||
parts = [f"#{a.index} {a.name}", a.chunk, f"Maske {e.mask_area:,} m²".replace(",", ".")]
|
||
if e.shared_with is not None:
|
||
parts.append(f"gemeinsam mit #{e.shared_with}")
|
||
elif e.stats is not None:
|
||
s = e.stats
|
||
parts.append(f"Straßen {s.connected} verbunden, {s.extended} weitergeführt, {s.dead_ends} Sackgassen")
|
||
parts.append(f"Gebäude {s.closed} geschlossen, {s.added} ergänzt" + (f" ({s.note})" if s.note else ""))
|
||
lines.append(" " + " · ".join(parts))
|
||
warned = [e for e in self.entries if e.warning]
|
||
if warned:
|
||
lines.append("Warnungen:")
|
||
lines += [f" #{e.annotation.index} {e.annotation.name} · {e.annotation.chunk} · {e.warning}" for e in warned]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def remove_annotations(mosaic: np.ndarray, labels: np.ndarray, annotations: AnnotationList) -> Removal:
|
||
"""Entfernt alle Einträge und das Gitternetz aus einer Kopie des Klassenbilds."""
|
||
labels = labels.copy()
|
||
union = np.zeros(labels.shape, dtype=bool)
|
||
entries, found = [], []
|
||
for annotation in annotations.annotations:
|
||
result = annotation_mask(mosaic, annotation)
|
||
entry = EntryReport(annotation, result.applied, result.warning)
|
||
if result.applied:
|
||
union[result.rows, result.cols] |= result.mask
|
||
entry.mask_area = int(result.mask.sum())
|
||
entries.append(entry)
|
||
found.append(result)
|
||
|
||
grid_lines = 0
|
||
if annotations.grid is not None:
|
||
grid, grid_lines = grid_mask(labels.shape, annotations.grid)
|
||
blocked = grid | union
|
||
for strip in _grid_strips(labels.shape, annotations.grid):
|
||
fill_nearest(labels[strip], (grid & ~union)[strip], blocked[strip])
|
||
|
||
components, _ = ndi.label(union, structure=EIGHT)
|
||
owners: dict[int, EntryReport] = {} # Maske -> erster Eintrag in Dateireihenfolge
|
||
touched = []
|
||
for entry, result in zip(entries, found):
|
||
ids = np.unique(components[result.rows, result.cols][result.mask]) if entry.applied else np.array([], int)
|
||
ids = ids[ids > 0]
|
||
for cid in ids:
|
||
owners.setdefault(int(cid), entry)
|
||
touched.append(ids)
|
||
|
||
generated, rebuilt = [], []
|
||
h, w = labels.shape
|
||
for cid, sl in enumerate(ndi.find_objects(components), start=1):
|
||
if sl is None:
|
||
continue
|
||
piece = components[sl] == cid
|
||
if is_thin(piece):
|
||
win = (slice(max(0, sl[0].start - THIN_WINDOW), min(h, sl[0].stop + THIN_WINDOW)),
|
||
slice(max(0, sl[1].start - THIN_WINDOW), min(w, sl[1].stop + THIN_WINDOW)))
|
||
fill_nearest(labels[win], components[win] == cid, union[win])
|
||
continue
|
||
mask = np.zeros(labels.shape, dtype=bool)
|
||
mask[sl] = piece
|
||
result = rebuild(labels, mask, union, add=False)
|
||
owner = owners[cid]
|
||
owner.stats = result.stats if owner.stats is None else _merged(owner.stats, result.stats)
|
||
rebuilt.append((cid, sl))
|
||
# Ergänzt wird erst, wenn alle Masken aufgebaut sind: sonst zählten Buchstabenkonturen noch nicht
|
||
# aufgebauter Masken als Straße und teilten Blöcke, und das Ergebnis hinge von der Reihenfolge ab.
|
||
for cid, sl in rebuilt:
|
||
mask = np.zeros(labels.shape, dtype=bool)
|
||
mask[sl] = components[sl] == cid
|
||
result = add_buildings(labels, mask, union)
|
||
owners[cid].stats = _merged(owners[cid].stats, result.stats)
|
||
generated += [Polygon([(SLICE["x0"] + c, SLICE["y1"] - r) for r, c in corners])
|
||
for corners in result.buildings]
|
||
for entry, ids in zip(entries, touched):
|
||
if entry.applied and entry.stats is None and ids.size and owners[int(ids[0])] is not entry:
|
||
entry.shared_with = owners[int(ids[0])].annotation.index
|
||
return Removal(labels, generated, entries, grid_lines)
|
||
|
||
|
||
def _merged(a: RebuildStats, b: RebuildStats) -> RebuildStats:
|
||
return RebuildStats(a.connected + b.connected, a.extended + b.extended, a.dead_ends + b.dead_ends,
|
||
a.closed + b.closed, a.added + b.added, "; ".join(dict.fromkeys(n for n in (a.note, b.note) if n)))
|
||
|
||
|
||
def mark_generated(polygons: list[Polygon], heights: list[dict], generated: list[Polygon]) -> int:
|
||
"""Setzt source = "annotation" bei Gebäuden, die zu mindestens der Hälfte in ergänzten Rechtecken liegen."""
|
||
if not generated:
|
||
return 0
|
||
tree = STRtree(generated)
|
||
marked = 0
|
||
for polygon, height in zip(polygons, heights):
|
||
covered = sum(polygon.intersection(generated[k]).area for k in tree.query(polygon))
|
||
if covered >= GENERATED_SHARE * polygon.area:
|
||
height["source"] = "annotation"
|
||
marked += 1
|
||
return marked
|