# Copyright (C) 2026 Wilhelmus Dingenus de Pooter
# SPDX-License-Identifier: GPL-3.0-or-later
"""
enc_tile_bake.py
----------------
QGIS-FREE bake of staged S-57 / Inland-ENC cells into the Chart Viewer's two
vector-tile planes (wet skin, dry+everything). Runs entirely on the native
iso8211 engine + raw GDAL/OGR — it creates NO ``QgsVectorLayer`` — so it is safe
to drive from a background (non-GUI) QThread at scan time (see
[[qgis-worker-shutdown]]: building a QgsVectorLayer off the GUI thread crashes).

Pipeline (all off-GUI-thread safe):
  decode_s57_buckets(cells)            # native decode → WKT+attrs buckets
    → split WET / DRY (+ fold AtoN)    # pure-Python
    → raw OGR GeoPackage per plane     # osgeo.ogr, dedupes case-colliding fields
    → gdal.VectorTranslate → MBTiles   # MVT, one pass per plane (multi-layer)

The viewer later loads the two MBTiles and styles them with enc_tile_styles
(symbology + labels generated from our portrayal engine). SOUNDG is deliberately
NOT tiled — it stays a native adaptive layer.
"""

import json
import os
import re

# Wet skin = the seabed/depth portrayal S-102 replaces. Everything else (land,
# coastline, navaids, structures…) is the dry+everything plane.
WET_CLASSES = {"DEPARE", "DRGARE", "DEPCNT", "UNSARE", "M_QUAL", "M_SREL"}
_SOUNDG = "SOUNDG"

# ── per-cell decode cache (JSON, schema-aware) ───────────────────────────────
# The bake caches its native per-cell decode so the Chart Viewer can open
# without re-decoding. The cache schema is fixed and produced only by this
# module:  {cell_path: {"buckets": {(klass, kind, geomtype): {"rows":
# [(fid, wkt, attrs)…], "attrs": set()}, …}, "soundings": [(x, y, z)…]}}.
# It is serialised as JSON (never pickle — deserialising pickle would execute
# arbitrary code if a cache file were tampered with); tuple keys join on a
# unit separator and tuples/sets are restored on read, so the round-trip is
# exact for this schema.
_KEY_SEP = "\x1f"


def write_decode_cache(path, per_cell):
    """Write the per-cell decode cache as JSON. Returns ``path``."""
    slim = {}
    for cell_path, cell in (per_cell or {}).items():
        buckets = {}
        for key, b in (cell.get("buckets") or {}).items():
            skey = _KEY_SEP.join(key) if isinstance(key, tuple) else str(key)
            buckets[skey] = {"rows": b.get("rows") or [],
                             "attrs": sorted(b.get("attrs") or ())}
        out_cell = dict(cell)
        out_cell["buckets"] = buckets
        out_cell["soundings"] = cell.get("soundings") or []
        slim[cell_path] = out_cell
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(slim, fh, ensure_ascii=False, separators=(",", ":"))
    return path


def read_decode_cache(path):
    """Read a JSON decode cache back into the native schema (tuple bucket keys,
    tuple rows/soundings, attrs as sets). Refuses legacy pickle caches."""
    if str(path).lower().endswith(".pkl"):
        raise ValueError("tile cache was baked by an older plugin version — "
                         "re-scan the source to rebuild it")
    with open(path, "r", encoding="utf-8") as fh:
        raw = json.load(fh)
    per_cell = {}
    for cell_path, cell in (raw or {}).items():
        buckets = {}
        for skey, b in (cell.get("buckets") or {}).items():
            key = tuple(skey.split(_KEY_SEP)) if _KEY_SEP in skey else skey
            buckets[key] = {"rows": [tuple(r) for r in b.get("rows") or []],
                            "attrs": set(b.get("attrs") or ())}
        out_cell = dict(cell)
        out_cell["buckets"] = buckets
        out_cell["soundings"] = [tuple(s) for s in cell.get("soundings") or []]
        per_cell[cell_path] = out_cell
    return per_cell


def _log(msg, *_a):
    try:
        from qgis.core import QgsMessageLog, Qgis
        QgsMessageLog.logMessage(str(msg), "ENCore GIS", Qgis.Info)
    except Exception:                                   # noqa: BLE001
        pass


def _wkt_pos(wkt, dp=6):
    """Rounded (x, y) key from a POINT wkt — the QGIS-free equivalent of
    enc_loader._aton_pos_key, used to pair AtoN topmarks/lights to their buoy."""
    m = re.search(r"\(\s*([-\d.eE]+)\s+([-\d.eE]+)", wkt or "")
    if not m:
        return None
    try:
        return (round(float(m.group(1)), dp), round(float(m.group(2)), dp))
    except ValueError:
        return None


def _aton_compose(buckets, log):
    """Fold the AtoN family (buoy/beacon bodies + co-located topmarks + buoy-
    lights) into one ATON layer of (wkt, attrs) rows, reusing the shared engine
    core. Pops the consumed body/TOPMAR buckets and drops claimed buoy-lights
    from LIGHTS. Returns the list of composed rows (or [])."""
    try:
        from .enc_loader import (
            compose_aton_rows, match_buoy_lights, _parse_lnam_refs, _norm_lnam,
            _ATON_BUOY_FIELDS, _ATON_TOPMARK_FIELDS, _ATON_TT_RENAME,
            _ATON_LIGHT_FIELDS, _ATON_LT_RENAME, _ATON_IDENT_FIELDS,
            _ATON_TOPMARK_CLASS, _ATON_LIGHT_CLASS, _ATON_BODY_CLASSES_UPPER)
    except (ImportError, ValueError):
        from enc_loader import (
            compose_aton_rows, match_buoy_lights, _parse_lnam_refs, _norm_lnam,
            _ATON_BUOY_FIELDS, _ATON_TOPMARK_FIELDS, _ATON_TT_RENAME,
            _ATON_LIGHT_FIELDS, _ATON_LT_RENAME, _ATON_IDENT_FIELDS,
            _ATON_TOPMARK_CLASS, _ATON_LIGHT_CLASS, _ATON_BODY_CLASSES_UPPER)

    body_keys = [k for k in buckets
                 if k[0] in _ATON_BODY_CLASSES_UPPER and k[1] == "point"]
    tm_keys = [k for k in buckets if k[0] == _ATON_TOPMARK_CLASS and k[1] == "point"]
    light_keys = [k for k in buckets if k[0] == _ATON_LIGHT_CLASS and k[1] == "point"]
    if not body_keys and not tm_keys:
        return []

    def G(named, key):
        v = named.get(key)
        if v in (None, ""):
            # Inland ENC encodes some acronyms LOWERCASE (e.g. catlam=attr 17011,
            # marsys=17009) — match case-insensitively, parity with the project
            # loader's _aton_attr. Without this the Inland-only SIGNI recreation
            # marks (CATLAM 11/12/15/16) lose CATLAM and mis-portray in the tiles.
            lk = key.lower()
            for kk, vv in named.items():
                if kk.lower() == lk:
                    v = vv
                    break
        return None if v in (None, "") else v

    buoy_rows, refs_by_pos = [], {}
    for k in body_keys:
        for _rcid, wkt, named in buckets[k]["rows"]:
            pos = _wkt_pos(wkt)
            attrs = {"S57CLASS": k[0]}
            for bf in _ATON_BUOY_FIELDS:
                attrs[bf] = G(named, bf)
            for idf in _ATON_IDENT_FIELDS:
                attrs[idf] = G(named, idf)
            buoy_rows.append({"geom": wkt, "attrs": attrs, "pos": pos})
            if pos is not None:
                refs = _parse_lnam_refs(G(named, "LNAM_REFS"))
                if refs:
                    refs_by_pos.setdefault(pos, set()).update(refs)

    topmark_rows = []
    for k in tm_keys:
        for _rcid, wkt, named in buckets[k]["rows"]:
            tt = {_ATON_TT_RENAME[tf]: G(named, tf) for tf in _ATON_TOPMARK_FIELDS}
            ident = {idf: G(named, idf) for idf in _ATON_IDENT_FIELDS}
            topmark_rows.append(
                {"geom": wkt, "pos": _wkt_pos(wkt), "tt": tt,
                 "standalone_attrs": {"S57CLASS": _ATON_TOPMARK_CLASS, **tt, **ident}})

    lights = []
    for k in light_keys:
        for _rcid, wkt, named in buckets[k]["rows"]:
            if G(named, "SECTR1") is not None or G(named, "SECTR2") is not None:
                continue
            lnam = _norm_lnam(G(named, "LNAM"))
            pos = _wkt_pos(wkt)
            if not lnam or pos is None:
                continue
            lt = {_ATON_LT_RENAME[lf]: G(named, lf) for lf in _ATON_LIGHT_FIELDS}
            lights.append((lnam, pos, lt))

    claimed, fold = match_buoy_lights(refs_by_pos, lights)
    rows, stats = compose_aton_rows(buoy_rows, topmark_rows, fold)
    log(f"Tile bake: AtoN — {len(buoy_rows)} body, {stats['paired']} w/topmark, "
        f"{stats['lit']} lit, {stats['standalone']} lone topmark")
    for k in light_keys:
        kept = [r for r in buckets[k]["rows"] if _norm_lnam(G(r[2], "LNAM")) not in claimed]
        if kept:
            buckets[k]["rows"] = kept
        else:
            del buckets[k]
    for k in body_keys + tm_keys:
        buckets.pop(k, None)
    return rows


def _src_name(acr):
    return re.sub(r"\W+", "_", acr).strip("_").upper()[:30] or "X"


def _cell_catcov1_wkt(bcs):
    """Union of a cell's M_COVR CATCOV=1 area geometries → WKT (EPSG:4326), or
    None. CATCOV is read case-insensitively (inland cells lowercase it)."""
    try:
        from .enc_coverage import catcov1_union_wkt
    except (ImportError, ValueError):
        from enc_coverage import catcov1_union_wkt
    rows = []
    for key, b in bcs.items():
        if str(key[0]).upper() != "M_COVR":
            continue
        for _rcid, wkt, named in b["rows"]:
            cat = named.get("CATCOV")
            if cat in (None, ""):
                cat = next((v for k, v in named.items()
                            if k.lower() == "catcov"), None)
            rows.append((wkt, cat))
    return catcov1_union_wkt(rows) if rows else None


def _apply_coverage_priority(per_cell, log):
    """Coverage-priority clip (ECDIS 'larger scale overrides'): each cell keeps
    its OWN CATCOV=1 coverage MINUS the union of any HIGHER usage-band (more
    detailed) cell coverage that overlaps it — so the larger-scale chart wins its
    footprint and the smaller-scale one fills only the gaps. Removes the cross-
    band double-draw (buoys, soundings, depth areas). Mutates ``per_cell`` in
    place; M_COVR coverage definitions are left intact. GDAL/GEOS geometry only."""
    try:
        from . import enc_coverage
        from .enc_scanner import usage_band
    except (ImportError, ValueError):
        import enc_coverage
        from enc_scanner import usage_band

    paths = list(per_cell.keys())
    cov, usage, env = {}, {}, {}
    for p in paths:
        cov[p] = _cell_catcov1_wkt(per_cell[p]["buckets"])
        usage[p] = usage_band(os.path.splitext(os.path.basename(p))[0])
        env[p] = enc_coverage._envelope(cov[p]) if cov[p] else None

    def _disjoint(a, b):
        return a[1] < b[0] or b[1] < a[0] or a[3] < b[2] or b[3] < a[2]

    clipped_cells = removed = 0
    for p in paths:
        own, ub, oe = cov[p], usage[p], env[p]
        if not own or oe is None:
            continue
        higher = [cov[q] for q in paths
                  if q != p and usage[q] > ub and cov[q] and env[q] is not None
                  and not _disjoint(oe, env[q])]
        if not higher:
            continue
        mask = enc_coverage.difference_wkt(own, enc_coverage.union_wkt(higher))
        mask_empty = not mask                           # fully superseded by detail
        for key, b in per_cell[p]["buckets"].items():
            if str(key[0]).upper() == "M_COVR":
                continue
            n0 = len(b["rows"])
            b["rows"] = ([] if mask_empty
                         else enc_coverage.clip_rows(b["rows"], mask, key[1]))
            removed += n0 - len(b["rows"])
        snd = per_cell[p].get("soundings") or []
        if snd:
            kept = [] if mask_empty else enc_coverage.clip_points_xyz(snd, mask)
            removed += len(snd) - len(kept)
            per_cell[p]["soundings"] = kept
        clipped_cells += 1
    log(f"Tile bake: coverage-priority clip — {clipped_cells} cell(s) clipped by "
        f"higher-band coverage, {removed} feature(s) removed")


def _read_cscl(path):
    """The cell's compilation scale (DSPM ``CSCL`` — the denominator of its
    natural display scale), read natively from the ISO 8211 bytes. Baked onto
    every feature so the tile styles can declutter OVERSCALE detail (e.g. minor
    depth contours) per-cell: a 1:12 000 harbour cell drops its fine contours
    sooner than a 1:90 000 approach cell. Returns an int or None; never raises."""
    try:
        try:
            from .iso8211.reader import Iso8211Reader
        except (ImportError, ValueError):
            from iso8211.reader import Iso8211Reader
        for rec in Iso8211Reader.open(path).records():
            dspm = rec.first("DSPM")
            if dspm is not None:
                v = dspm.get("CSCL")
                if v not in (None, ""):
                    try:
                        return int(float(str(v).strip()))
                    except (ValueError, TypeError):
                        return None
                break                                   # DSPM found, no CSCL
    except Exception:                                   # noqa: BLE001
        return None
    return None


# The dry plane is styled by replaying the native renderers (enc_tile_styles.
# styles_from_renderer). Classes portrayed by the S-52 engine use a
# CATEGORIZED renderer keyed on a SYNTHETIC ``s52key`` field (= repr of the
# feature's drawing instructions; see enc_s52_portrayal.apply_s52_engine) — a
# field the raw decode does NOT carry, so without it every categorised tile
# filter matches nothing (land, built-up areas… render invisible). We recompute
# that exact key here: ``portray()`` is pure-Python (no QGIS), and the key is a
# repr of colour TOKENS (LANDA/CHBLK…), so it is palette- and contour-independent
# — matching the native categories the viewer extracts (palette "Day", the same
# fixed ctx enc_chartview._s52_engine_apply uses).
_S52_PORTRAY = None
_S52_KEY_PALETTE = "Day"
_S52_KEY_CTX = {"shallow": 2.0, "safety": 6.0, "deep": 10.0}


def _s52_portray_fn():
    global _S52_PORTRAY
    if _S52_PORTRAY is None:
        try:
            from .enc_s52_portrayal import portray
        except (ImportError, ValueError):
            try:
                from enc_s52_portrayal import portray
            except Exception:                           # noqa: BLE001
                return False                            # engine unavailable
        _S52_PORTRAY = portray
    return _S52_PORTRAY


def _s52key_for(acr, kind, named, cache):
    """The categorised-renderer key the native S-52 engine assigns this feature
    (``repr`` of its drawing instructions), or None if the engine is
    unavailable. Mirrors apply_s52_engine: the key is repr(instrs) even when the
    class has no lookup (repr([]) = '[]'), so a class the engine did not portray
    just gets a harmless '[]' the dry styles ignore (they filter on real attrs /
    are single-symbol). Cached by (acr, kind, attrs) — features repeat heavily."""
    portray = _s52_portray_fn()
    if portray is False:
        return None
    attrs = {k: v for k, v in named.items() if v not in (None, "")}
    ck = (acr, kind, tuple(sorted((k, str(v)) for k, v in attrs.items())))
    if ck in cache:
        return cache[ck]
    try:
        key = repr(portray(acr, attrs, kind, _S52_KEY_PALETTE, _S52_KEY_CTX))
    except Exception:                                   # noqa: BLE001
        key = None
    cache[ck] = key
    return key


def _ogr_geom_type(kind):
    # wkbUnknown accepts single OR multi geometries without the GPKG
    # "POLYGON inserted into MULTIPOLYGON layer" warnings; MVT layers hold mixed
    # geometry anyway and the tile style's geometryType selects at draw time.
    from osgeo import ogr
    return {"point": ogr.wkbPoint}.get(kind, ogr.wkbUnknown)


def _write_gpkg(layer_specs, gpkg, log):
    """Write [(name, kind, field_names, rows[(wkt, attrs)])] to a GeoPackage via
    raw OGR (no QGIS). Field names are de-duplicated case-insensitively (GPKG /
    SQLite columns are case-insensitive — e.g. RESARE carries RESTRN AND restrn,
    which otherwise aborts the table)."""
    from osgeo import ogr, osr
    drv = ogr.GetDriverByName("GPKG")
    if os.path.exists(gpkg):
        os.remove(gpkg)
    ds = drv.CreateDataSource(gpkg)
    srs = osr.SpatialReference()
    srs.ImportFromEPSG(4326)
    n_written = 0
    for name, kind, field_names, rows in layer_specs:
        if not rows:
            continue
        # One failing source-layer must NOT abort the whole plane — log and skip
        # it, so the bake still produces (partial) tiles rather than dropping to
        # the native fallback.
        try:
            lyr = ds.CreateLayer(name, srs, _ogr_geom_type(kind))
            fields, seen = [], set()
            for fn in field_names:
                low = fn.lower()
                if low in seen:                         # case-insensitive collision
                    continue
                seen.add(low)
                fields.append(fn)
                lyr.CreateField(ogr.FieldDefn(fn, ogr.OFTString))
            defn = lyr.GetLayerDefn()
            lyr.StartTransaction()
            for wkt, attrs in rows:
                g = ogr.CreateGeometryFromWkt(wkt)
                if g is None:
                    continue
                ft = ogr.Feature(defn)
                ft.SetGeometry(g)
                for fn in fields:
                    v = attrs.get(fn)
                    if v not in (None, ""):
                        ft.SetField(fn, str(v))
                lyr.CreateFeature(ft)
                ft = None
            lyr.CommitTransaction()
            n_written += 1
        except Exception as e:                          # noqa: BLE001
            log(f"Tile bake: source-layer {name!r} skipped ({e})")
    ds = None
    return n_written


def _field_union(rows):
    """Ordered union of attribute keys across (wkt, attrs) rows."""
    fields = []
    for _wkt, attrs in rows:
        for k in attrs:
            if k not in fields:
                fields.append(k)
    return fields


def _write_native_gpkg(aton_rows, light_rows, soundings, gpkg, log):
    """Write the NATIVE overlay layers to a persistent GeoPackage (NOT tiled):
    the AtoN composite, the LIGHTS layer, and SOUNDG (as points with a ``depth``
    field). The viewer portrays these with the original engine QMLs and they are
    the pick index for those classes; the GPKG travels with a saved tileset so a
    set stays self-contained. Returns the path, or None if nothing was written."""
    specs = []
    if aton_rows:
        keys = []
        for r in aton_rows:
            for kf in r["attrs"]:
                if kf not in keys:
                    keys.append(kf)
        specs.append(("ATON", "point", keys,
                      [(r["geom"], r["attrs"]) for r in aton_rows]))
    if light_rows:
        specs.append(("LIGHTS", "point", _field_union(light_rows), light_rows))
    if soundings:
        specs.append(("SOUNDG", "point", ["depth"],
                      [(f"POINT({x} {y})", {"depth": str(z)})
                       for x, y, z in soundings]))
    if not specs:
        return None
    n = _write_gpkg(specs, gpkg, log)
    log(f"Tile bake: native overlay → {os.path.basename(gpkg)} ({n} layer(s))")
    return gpkg if n else None


def _specs_from_buckets(buckets, aton_rows, s52key=False):
    """Turn buckets (+ the composed AtoN rows) into _write_gpkg layer specs.

    Buckets are MERGED by source-layer name: a class with more than one geometry
    kind (e.g. BUISGL as line AND area) yields several buckets that all map to the
    same source-layer, so they must become ONE MVT source-layer (mixed geometry)
    — otherwise the second CreateLayer collides and aborts the bake.

    ``s52key=True`` (the dry plane) also bakes the synthetic ``s52key`` field so
    the categorised S-52 tile styles match (see _s52key_for)."""
    by_src = {}                 # src -> {"kinds": set, "fields": set, "rows": []}
    portray_cache = {}
    for (acr, kind, _memtype), b in buckets.items():
        e = by_src.setdefault(_src_name(acr),
                              {"kinds": set(), "fields": set(), "rows": []})
        e["kinds"].add(kind)
        e["fields"].update(b["attrs"])
        for rcid, wkt, named in b["rows"]:
            attrs = {"rcid": str(rcid), **named}
            if s52key:
                k = _s52key_for(acr, kind, named, portray_cache)
                if k is not None:
                    attrs["s52key"] = k
            e["rows"].append((wkt, attrs))
        if s52key:
            e["fields"].add("s52key")
    specs = []
    for src, e in by_src.items():
        # One kind → that OGR type; several → 'mixed' (→ wkbUnknown, holds any).
        kind = next(iter(e["kinds"])) if len(e["kinds"]) == 1 else "mixed"
        specs.append((src, kind, ["rcid"] + sorted(e["fields"]), e["rows"]))
    if aton_rows:
        keys = []
        for r in aton_rows:
            for kf in r["attrs"]:
                if kf not in keys:
                    keys.append(kf)
        specs.append(("ATON", "point", keys,
                      [(r["geom"], r["attrs"]) for r in aton_rows]))
    return specs


def bake_planes(paths, out_dir, log=None, minzoom=6, maxzoom=14,
                should_stop=None, progress=None):
    """Bake staged cells → wet.mbtiles + dry.mbtiles in ``out_dir``. QGIS-free;
    safe on a background thread. ``should_stop`` is an optional callable polled
    for cooperative cancel; ``progress`` an optional ``fn(done, total, msg)``
    for a progress bar. Returns {"wet": path|None, "dry": path|None, "decode":…}."""
    log = log or _log
    from osgeo import gdal
    gdal.UseExceptions()
    try:
        from .enc_chartview import decode_s57_buckets
    except (ImportError, ValueError):
        from enc_chartview import decode_s57_buckets

    try:
        from .s101.qgis_loader import decode_s101_buckets
    except (ImportError, ValueError):
        try:
            from s101.qgis_loader import decode_s101_buckets
        except Exception:                               # noqa: BLE001
            decode_s101_buckets = None

    def _is_s101(path):
        return (decode_s101_buckets is not None
                and os.path.basename(path)[:3].upper() == "101")

    def _decode(path):
        """Route a cell to the S-101 or S-57/IENC decoder (same bucket shape).
        S-101 cells follow the fixed '101…' FNC naming (S-101 Part 10a)."""
        if _is_s101(path):
            return decode_s101_buckets([path], log, should_stop)
        return decode_s57_buckets([path], log, should_stop)

    def _cscl(path):
        """Compilation/optimum-display scale for the per-cell overscale declutter:
        S-57 DSPM CSCL, or S-101 DataCoverage optimumDisplayScale."""
        if _is_s101(path):
            try:
                from .s101.qgis_loader import s101_compilation_scale
            except (ImportError, ValueError):
                from s101.qgis_loader import s101_compilation_scale
            return s101_compilation_scale(path)
        return _read_cscl(path)

    def _stop():
        return should_stop is not None and should_stop()

    paths = list(paths)
    n_cells = max(len(paths), 1)

    # Percentage-weighted progress (0-100), so the LENGTHY tiling phase gets a
    # fair, animated share of the bar instead of being crammed into the last two
    # steps: decode 0-60 %, wet tiles 60-80 %, dry tiles 80-100 %. The tiling
    # phases animate via GDAL's own VectorTranslate callback.
    def _p(pct, msg):
        if progress is not None:
            try:
                progress(int(max(0.0, min(100.0, pct))), 100, msg)
            except Exception:                           # noqa: BLE001
                pass

    # Decode PER CELL so the viewer can build each cell off the cache (keeping
    # its chart-by-chart progressive UX), then MERGE into combined buckets for
    # the (whole-coverage) tile planes.
    per_cell = {}
    buckets, soundings = {}, []
    for i, p in enumerate(paths):
        if _stop():
            return {"wet": None, "dry": None, "decode": None}
        _p(60.0 * i / n_cells, f"Reading cell {i + 1}/{len(paths)}…")
        bcs, snd = _decode(p)
        # Tag every feature with the cell's compilation scale so the tiles can
        # declutter overscale detail per-cell (see _cscl / enc_tile_styles).
        cscl = _cscl(p)
        if cscl:
            cs = str(cscl)
            for v in bcs.values():
                v["attrs"].add("cscl")
                for _rcid, _wkt, named in v["rows"]:
                    named["cscl"] = cs
        per_cell[p] = {"buckets": bcs, "soundings": snd}

    # Coverage-priority clip BEFORE merging: the larger-scale (higher usage band)
    # chart wins its footprint, the smaller-scale one fills the gaps — so nested
    # cells (U4/U5/U6, whose CATCOV=1 DO overlap) don't double-draw buoys /
    # soundings / depth areas. Applied to the per-cell cache too, so the viewer's
    # native per-cell build matches the tiles.
    if not _stop():
        try:
            _apply_coverage_priority(per_cell, log)
        except Exception as e:                          # noqa: BLE001
            log(f"Tile bake: coverage-priority clip skipped ({e})")

    # Merge the (clipped) per-cell buckets into the whole-coverage tile planes.
    for p in paths:
        for k, v in per_cell[p]["buckets"].items():
            cb = buckets.setdefault(k, {"rows": [], "attrs": set()})
            cb["rows"].extend(v["rows"])
            cb["attrs"].update(v["attrs"])
        soundings.extend(per_cell[p].get("soundings") or [])
    # Cache the per-cell decode (pre-AtoN-fold) so the viewer skips the expensive
    # decode. QGIS-free JSON (wkt strings + dicts; see write_decode_cache).
    decode_cache = None
    try:
        os.makedirs(out_dir, exist_ok=True)
        decode_cache = write_decode_cache(
            os.path.join(out_dir, "decode.json"), per_cell)
    except Exception as e:                              # noqa: BLE001
        log(f"Tile bake: decode cache failed: {e}")
        decode_cache = None
    aton_rows = _aton_compose(buckets, log)                # mutates buckets

    # NAVAIDS are NOT tiled — the floating AtoN composite and the LIGHTS layer
    # render NATIVELY over the tiles (sector-light arcs are QML geometry
    # generators that MVT clips; the AtoN keeps its folded topmark/light + SCAMIN
    # + the pick parent/child links). We pull LIGHTS out of the tile buckets here
    # (AtoN was already folded out by _aton_compose) and write both — plus SOUNDG
    # — to a native GeoPackage that travels with the set when it is saved.
    light_rows = []
    for k in [k for k in buckets if k[0] == "LIGHTS" and k[1] == "point"]:
        light_rows.extend((wkt, {"rcid": str(rcid), **named})
                          for rcid, wkt, named in buckets[k]["rows"])
        del buckets[k]

    # NOTMRK: annotate the same-position stack fields (nmk_i/nmk_n) into the
    # bucket rows so they bake into the dry tiles — the CEVNI displacement
    # stack (enc_notmrk) reads them from the tile attributes.
    try:
        try:
            from .enc_notmrk import annotate_buckets
        except (ImportError, ValueError):
            from enc_notmrk import annotate_buckets
        annotate_buckets(buckets)
    except Exception as e:                              # noqa: BLE001
        log(f"Tile bake: NOTMRK stack annotation failed ({e})")

    wet, dry = {}, {}
    for key, b in buckets.items():
        (wet if key[0] in WET_CLASSES else dry)[key] = b

    os.makedirs(out_dir, exist_ok=True)
    result = {"wet": None, "dry": None, "decode": decode_cache, "native": None}
    # Native overlay GeoPackage (AtoN composite · LIGHTS · SOUNDG) — one point
    # layer each, portrayed by the viewer with the original engine QMLs.
    try:
        result["native"] = _write_native_gpkg(
            aton_rows, light_rows, soundings,
            os.path.join(out_dir, "native.gpkg"), log)
    except Exception as e:                              # noqa: BLE001
        log(f"Tile bake: native overlay GPKG failed ({e})")
    for j, (plane, bks, extra) in enumerate(
            (("wet", wet, []), ("dry", dry, []))):
        if _stop():
            break
        base = 60 + 20 * j                              # wet 60-80 %, dry 80-100 %
        _p(base, f"Baking {plane}-skin tiles…")
        # One plane failing must NOT abort the other (or drop to native) — the
        # worker only reports failure if BOTH planes die.
        try:
            specs = _specs_from_buckets(bks, extra, s52key=(plane == "dry"))
            if not specs:
                continue
            gpkg = os.path.join(out_dir, f"{plane}.gpkg")
            mb = os.path.join(out_dir, f"{plane}.mbtiles")
            n = _write_gpkg(specs, gpkg, log)
            if not n:
                continue
            if os.path.exists(mb):
                os.remove(mb)

            def _cb(complete, _msg, _u, _base=base, _plane=plane):
                _p(_base + 20.0 * complete,
                   f"Baking {_plane}-skin tiles… {int(complete * 100)}%")
                return 0 if _stop() else 1              # 0 = cancel the translate

            gdal.VectorTranslate(mb, gpkg, format="MBTiles",
                                 datasetCreationOptions=[f"MINZOOM={minzoom}",
                                                         f"MAXZOOM={maxzoom}"],
                                 callback=_cb)
            try:
                os.remove(gpkg)                         # the GPKG was scratch
            except OSError:
                pass
            result[plane] = mb
            log(f"Tile bake: {plane} plane → {os.path.basename(mb)} "
                f"({n} source-layer(s))")
        except Exception as e:                          # noqa: BLE001
            log(f"Tile bake: {plane} plane FAILED ({e})")
    _p(100, "Chart tiles ready")
    return result
