feat(extract): Beschriftungsmasken, Gitternetz und Entfernen mit Bericht

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tools/extract/leonida/annotations.py

@@ -9,16 +9,35 @@ 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, 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):
@@ -193,3 +212,238 @@ def _parse_grid(value) -> Grid | None:
         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 = []
+    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)
+        owner = owners[cid]
+        owner.stats = result.stats if owner.stats is None else _merged(owner.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

tools/extract/tests/test_annotations_remove.py

@@ -0,0 +1,172 @@
+import numpy as np
+from shapely.geometry import box
+
+from leonida.annotations import (Annotation, AnnotationList, Grid, annotation_mask, candidates, grid_mask,
+                                 mark_generated, remove_annotations)
+from leonida.classify import CLASSES, classify
+
+PARK, ROAD, URBAN = (192, 216, 128), (80, 80, 80), (216, 216, 216)
+WHITE, OUTLINE = (248, 248, 248), (80, 88, 96)
+
+
+def picture(h=200, w=300, colour=URBAN):
+    img = np.zeros((h, w, 3), dtype=np.uint8)
+    img[:] = colour
+    return img
+
+
+def entry(kind, rows, cols, index=1, name="Test"):
+    """Eintrag, dessen Rahmen genau die Pixelzeilen rows und -spalten cols umfasst (halboffen)."""
+    (r0, r1), (c0, c1) = rows, cols
+    return Annotation(index, name, kind, (-1000.0 + c0, 2500.0 - (r1 - 1), -1000.0 + c1 - 1, 2500.0 - r0), None)
+
+
+def letter(img, rows, cols, outline=3):
+    (r0, r1), (c0, c1) = rows, cols
+    img[r0 - outline:r1 + outline, c0 - outline:c1 + outline] = OUTLINE
+    img[r0:r1, c0:c1] = WHITE
+
+
+def test_candidates_follow_the_colour_rules():
+    colours = {"white": WHITE, "urban": URBAN, "red": (248, 112, 112), "speculative": (160, 72, 72),
+               "magenta": (200, 0, 200), "black": (20, 20, 20), "water": (48, 144, 192), "road": ROAD,
+               "park": PARK, "light": (230, 230, 230), "building": (176, 176, 176)}
+    img = np.array([list(colours.values())], dtype=np.uint8)
+    names = list(colours)
+
+    def hits(kind):
+        return {name for name, hit in zip(names, candidates(img, kind)[0]) if hit}
+
+    assert hits("text_white") == {"white"}
+    assert hits("text_red") == {"red"}
+    assert hits("marker") == {"magenta", "black"}
+    assert hits("symbol") == {"white", "magenta", "black"}
+    assert hits("street_name") == {"white", "urban", "light"}
+
+
+def test_white_text_is_masked_together_with_its_outline():
+    img = picture(colour=PARK)
+    img[:, 140:160] = ROAD
+    letter(img, (80, 120), (120, 180))
+    found = annotation_mask(img, entry("text_white", (60, 141), (100, 201)))
+    mask = np.zeros(img.shape[:2], dtype=bool)
+    mask[found.rows, found.cols] = found.mask
+    assert found.applied and found.warning is None
+    assert mask[77:123, 117:183].all()                      # Füllung und Kontur
+    assert not mask[80:120, 107].any()                      # 10 m neben der Kontur ist wieder Karte
+
+
+def test_text_filling_most_of_its_frame_after_growing_is_not_overfull():
+    img = picture()
+    for left in range(60, 240, 30):
+        letter(img, (60, 140), (left, left + 12))           # sechs Buchstabenstriche, Rahmen mit 10 m Luft
+    found = annotation_mask(img, entry("text_white", (47, 154), (47, 244)))
+    assert found.mask.mean() > 0.6
+    assert found.applied and found.warning is None
+
+
+def test_street_names_do_not_warn_about_urban_ground_at_the_frame_edge():
+    img = picture()
+    img[:, 140:160] = ROAD
+    img[80:120, 146:150] = WHITE                            # helle Schrift auf der Fahrbahn
+    found = annotation_mask(img, entry("street_name", (70, 131), (130, 171)))
+    assert found.applied and found.warning is None
+    assert found.mask[20, 17] and not found.mask[20, 2]
+
+
+def test_colour_touching_the_frame_edge_stays_and_is_reported():
+    img = picture()
+    letter(img, (80, 120), (60, 90))
+    letter(img, (80, 120), (180, 240))                      # ragt über den rechten Rahmenrand
+    found = annotation_mask(img, entry("text_white", (60, 141), (40, 211)))
+    assert found.applied
+    assert found.warning == "angeschnitten: Rahmen prüfen"
+    assert found.mask[30, 35]                               # Buchstabe innen
+    assert not found.mask[30, 160]                          # Buchstabe am Rand bleibt
+
+
+def test_letter_counters_are_filled_but_large_enclosed_areas_stay():
+    img = picture(420, 420)
+    img[20:80, 20:80] = WHITE
+    img[41:59, 41:59] = URBAN                               # 18 x 18 m großes „O"-Innere
+    img[150:290, 150:290] = WHITE
+    img[195:245, 195:245] = URBAN                           # 50 x 50 m: echte Fläche
+    small = annotation_mask(img, entry("text_white", (0, 130), (0, 130)))
+    large = annotation_mask(img, entry("text_white", (110, 331), (110, 331)))
+    assert small.applied and large.applied
+    assert small.mask[50, 50]
+    assert not large.mask[220 - 110, 220 - 110]
+
+
+def test_region_masks_exactly_its_polygon():
+    img = picture()
+    triangle = ((-950.0, 2450.0), (-850.0, 2450.0), (-950.0, 2350.0))
+    found = annotation_mask(img, Annotation(1, "Rest", "region", None, triangle))
+    assert found.applied
+    assert abs(int(found.mask.sum()) - 5000) <= 200
+    mask = np.zeros(img.shape[:2], dtype=bool)
+    mask[found.rows, found.cols] = found.mask
+    assert mask[60, 60] and not mask[140, 140]
+
+
+def test_empty_and_overfull_frames_are_skipped_with_a_warning():
+    park = annotation_mask(picture(colour=PARK), entry("text_white", (50, 100), (50, 100)))
+    assert not park.applied and park.warning == "leer: keine Pixel der Art text_white im Rahmen"
+    img = picture()
+    img[52:98, 52:98] = WHITE
+    full = annotation_mask(img, entry("text_white", (50, 100), (50, 100)))
+    assert not full.applied and full.warning.startswith("randvoll: Farbpixel füllen")
+
+
+def test_grid_mask_puts_a_stripe_on_every_line():
+    mask, lines = grid_mask((30, 30), Grid(10, 2))
+    assert lines == 8
+    for index in (0, 9, 10, 19, 20, 29):
+        assert mask[:, index].all() and mask[index, :].all()
+    assert not mask[5, 5] and not mask[15, 25]
+
+
+def test_removal_without_entries_changes_nothing():
+    labels = classify(picture())
+    removal = remove_annotations(picture(), labels, AnnotationList())
+    assert (removal.labels == labels).all() and removal.generated == []
+    assert removal.report() == "Beschriftungen: 0 angewendet, 0 übersprungen · Gitternetz: 0 Linien"
+
+
+def test_removal_rebuilds_a_road_under_white_text():
+    img = picture()
+    img[:, 140:160] = ROAD
+    letter(img, (80, 120), (60, 240))
+    removal = remove_annotations(img, classify(img), AnnotationList([entry("text_white", (60, 141), (40, 261))]))
+    assert (removal.labels[80:120, 141:159] == CLASSES["strasse"]).all()
+    assert not (removal.labels[80:120, 60:137] == CLASSES["strasse"]).any()
+    report = removal.report()
+    assert "#1 Test · A1 · Maske" in report and "Straßen 1 verbunden" in report
+
+
+def test_the_grid_is_closed_before_anything_is_rebuilt():
+    img = picture()
+    img[:, 98:102] = (140, 140, 140)                        # abgedunkelte Gitterlinie über Stadtfläche
+    labels = classify(img)
+    assert (labels[:, 99] == CLASSES["gebaeude"]).all()
+    removal = remove_annotations(img, labels, AnnotationList(grid=Grid(100, 6)))
+    assert (removal.labels == CLASSES["stadt"]).all()
+    assert removal.grid_lines == 7                          # Spalten 0, 100, 200, 300 und Zeilen 0, 100, 200
+
+
+def test_entries_sharing_one_mask_are_reported_together():
+    img = picture()
+    img[:, 140:160] = ROAD
+    letter(img, (80, 120), (60, 150))
+    letter(img, (80, 120), (155, 240))
+    first = entry("text_white", (60, 141), (40, 153), index=1, name="Links")
+    second = entry("text_white", (60, 141), (152, 261), index=2, name="Rechts")
+    report = remove_annotations(img, classify(img), AnnotationList([first, second])).report()
+    assert "#2 Rechts · A1 · Maske" in report and "gemeinsam mit #1" in report
+
+
+def test_buildings_mostly_inside_added_rectangles_are_marked():
+    heights = [{"height": 9.0, "source": "district"}, {"height": 12.0, "source": "district"}]
+    marked = mark_generated([box(1, 1, 9, 9), box(6, 0, 16, 10)], heights, [box(0, 0, 10, 10)])
+    assert marked == 1
+    assert [h["source"] for h in heights] == ["annotation", "district"]