# Copyright (C) 2026 Wilhelmus Dingenus de Pooter
# SPDX-License-Identifier: GPL-3.0-or-later
"""
validation/engine.py
--------------------
The S-58 rules engine: a registry of checks, a runner, and a report.

Design
======
Every check is a small function decorated with ``@check(...)`` that carries the
S-58 metadata (number, severity, category, message template, suggested
solution, and the "Conformity to" reference). A check receives the whole
``EncDataset`` plus the live ``Catalogue`` and *yields* one ``Finding`` per
offending feature. The engine collects findings into a ``Report`` grouped by
severity, exactly as S-58 1.2 requires ("validation software must group
validation reports using these categories").

Adding a check is therefore a one-function change -- write the predicate, yield
findings, decorate it. Nothing else in the engine needs to know about it.

This module is pure stdlib so the engine and every non-geometric check are
unit-testable without QGIS or GDAL.
"""

import csv
import html
import io
import re
from datetime import datetime, timezone
from time import perf_counter


# Friendly sign-off placed at the foot of the auditor and validator documents
# (four blank lines, then a centred bold block).
SIGNATURE_HTML = (
    '<br><br><br><br>'
    '<div style="text-align:center;font-weight:700;color:#1a1a1a;'
    'margin:8px 0 4px">'
    'Kind regards from<br>'
    'teamScheepvaart | Rijkswaterstaat CIV IGA DOA<br><br>'
    'Sail the seven seas in peace</div>')


def _slug(text, maxlen=64):
    """Lowercase hyphen-slug of a phrase, with bracketed [detail] dropped."""
    text = re.sub(r"\[.*?\]", "", str(text)).strip().rstrip(".").lower()
    text = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
    if len(text) > maxlen:
        cut = text[:maxlen]
        if "-" in cut:
            cut = cut[:cut.rindex("-")]   # don't sever a word
        text = cut.strip("-")
    return text or "check"


def _derive_name(category, message):
    """Fallback plain-language documentation key, 'group/phrase', synthesised
    from a check's category and message. Used until an explicit name is
    authored on the check (see CheckMeta.name)."""
    return f"{_slug(category, 40)}/{_slug(message)}"


# -- severity -----------------------------------------------------------------
class Severity:
    """S-58 1.2 check classifications, plus a local ALERT tier. Ordering:
    CRITICAL is most severe.

    ALERT is NOT an S-58 severity — it is an advisory the auditor raises for
    things that are technically off-standard but must not fail the dataset (e.g.
    a mandatory meta object that IS present but encoded with its lower-case
    Inland class code instead of the standard upper-case one). It renders
    directly under the Critical section so it's seen, but it never counts toward
    the minimum-standard verdict (only CRITICAL findings do)."""
    CRITICAL = "C"
    ALERT = "A"
    ERROR = "E"
    WARNING = "W"

    LABELS = {
        "C": "Critical error",
        "A": "Alert",
        "E": "Error",
        "W": "Warning",
    }
    # Display / sort order, most severe first; ALERT sits just under Critical.
    ORDER = ("C", "A", "E", "W")


# -- check metadata + registry ------------------------------------------------
class CheckMeta:
    """Immutable description of a single S-58 check, attached to its function."""

    __slots__ = ("number", "severity", "category", "message",
                 "solution", "conformity", "func", "_name", "validated")

    def __init__(self, number, severity, category, message,
                 solution, conformity, func, name=None, validated=True):
        self.number = number
        self.severity = severity
        self.category = category
        self.message = message
        self.solution = solution
        self.conformity = conformity
        self.func = func
        self._name = name
        # validated=False marks a check implemented from the S-58 document text
        # but NOT yet confirmed to fire correctly against an official IHO S-58
        # test dataset — the audit report flags such checks with a trailing '*'.
        self.validated = validated

    @property
    def name(self):
        """Stable plain-language documentation key, ``group/phrase``.

        This is the identity used for documentation and the audit report — it
        does not change when a citation does. ``number``/``severity`` are
        citations (the S-58 reference) and may be revised between editions.
        Authored names win; otherwise a readable key is derived from the
        category and message so every check has one from day one.
        """
        return self._name or _derive_name(self.category, self.message)

    @property
    def title(self):
        """Short, human-readable check name for display (auditor tree + report):
        the message's leading clause, e.g. 'Mandatory meta object is missing
        (M_COVR, M_NSYS and M_QUAL required).' -> 'Mandatory meta object is
        missing'. The ``name`` slug stays the stable machine/doc key; this is
        just the friendly label a user reads. Falls back to the full message."""
        msg = (self.message or "").strip()
        head = msg.split("(", 1)[0].strip().rstrip(".").strip()
        return head or msg or self.name


# Ordered registry of every check the engine knows about.
REGISTRY = []


def check(number, severity, category, message, solution="", conformity="",
          name=None, validated=True):
    """Decorator registering a check function with its S-58 metadata.

    The wrapped function signature is ``func(dataset, catalogue, geom) -> iter``
    yielding ``Finding`` objects. ``geom`` is the spatial-operator helper module
    (passed in so non-geometric checks can ignore it and geometric checks share
    one tolerance-aware implementation).

    ``name`` is the stable plain-language documentation key (``group/phrase``);
    omit it and a readable key is derived from category + message.

    ``validated`` — set False for a check implemented from the S-58 document but
    not yet confirmed against an official IHO test dataset; the report marks it
    with a trailing '*' and shows a caveat note.
    """
    def _wrap(func):
        REGISTRY.append(CheckMeta(number, severity, category, message,
                                  solution, conformity, func, name, validated))
        return func
    return _wrap


# -- findings + report --------------------------------------------------------
class Finding:
    """One rule violation against one feature (or feature pair)."""

    __slots__ = ("check", "cell", "klass", "fid", "detail", "geom_wkt",
                 "source", "exclude")

    def __init__(self, check_meta, cell, klass, fid, detail="", geom_wkt=None):
        self.check = check_meta
        self.cell = cell
        self.klass = klass
        self.fid = fid
        self.detail = detail        # extra context, e.g. "WATLEV=2"
        self.geom_wkt = geom_wkt     # for zoom-to in the UI; optional
        # Source-validation triage identity: ("source", id, "") or
        # ("enc", lnam, objnam). None for auditor findings. The validator's
        # results menu reads this to write a recipe exclusion.
        self.exclude = None
        # The registered check that produced this finding. ``check`` above is
        # the *citation* (its S-58 number/severity), which for data-driven
        # families like the prohibited-attribute set is synthesised per row;
        # ``source`` is the single registered check whose function ran, and is
        # what per-check PASS/FAIL groups on. The runner stamps it.
        self.source = check_meta

    @property
    def message(self):
        base = self.check.message
        return f"{base} [{self.detail}]" if self.detail else base


class Report:
    """Collected findings, queryable and exportable."""

    def __init__(self):
        self.findings = []
        self.ran = []          # CheckMeta that actually executed
        self.errors = []       # (CheckMeta, exception) for checks that threw
        self.timings = []      # (CheckMeta, seconds, n_findings) per check
        self.started = datetime.now(timezone.utc)
        self.finished = None

    def slowest(self, n=5):
        """The n slowest checks this run, as (meta, seconds, n_findings)."""
        return sorted(self.timings, key=lambda t: t[1], reverse=True)[:n]

    def total_seconds(self):
        return sum(t[1] for t in self.timings)

    def add(self, finding):
        self.findings.append(finding)

    def counts(self):
        """Return {severity: n} across all findings."""
        out = {s: 0 for s in Severity.ORDER}
        for f in self.findings:
            out[f.check.severity] = out.get(f.check.severity, 0) + 1
        return out

    def by_severity(self):
        """Yield (severity, [findings]) in C, E, W order."""
        for sev in Severity.ORDER:
            group = [f for f in self.findings if f.check.severity == sev]
            yield sev, group

    def passes_minimum_standard(self):
        """S-58 8.0.0 1.3: an ENC meets the minimum standard if it has no
        Critical findings."""
        return self.counts().get(Severity.CRITICAL, 0) == 0

    # -- exports --------------------------------------------------------------
    def to_csv(self):
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(["Severity", "Check", "Category", "Cell", "Class",
                    "FID", "Message", "Solution", "Conformity"])
        for f in self.findings:
            w.writerow([Severity.LABELS.get(f.check.severity, f.check.severity),
                        f.check.number, f.check.category, f.cell, f.klass,
                        f.fid, f.message, f.check.solution, f.check.conformity])
        return buf.getvalue()

    def summary_line(self):
        c = self.counts()
        verdict = "PASS" if self.passes_minimum_standard() else "FAIL"
        alert = f", {c.get('A', 0)} alert" if c.get('A', 0) else ""
        return (f"{verdict} (minimum standard) — "
                f"{c.get('C', 0)} critical, {c.get('E', 0)} error, "
                f"{c.get('W', 0)} warning{alert} across {len(self.ran)} checks")

    # -- per-check PASS / FAIL ------------------------------------------------
    def per_check(self):
        """Outcome for every check that was in scope this run.

        Returns a list of ``(meta, status, findings)`` where ``meta`` is the
        *registered* check, ``status`` is ``"PASS"`` (ran, no findings),
        ``"FAIL"`` (ran, one or more findings) or ``"ERROR"`` (raised before
        completing), and ``findings`` are the findings it produced. Grouping is
        on ``finding.source`` (the registered check), so data-driven families
        like the prohibited-attribute set count as one check, not one per row.
        """
        by_source = {}
        for f in self.findings:
            by_source.setdefault(id(f.source), []).append(f)
        out = []
        for meta in self.ran:
            fs = by_source.get(id(meta), [])
            # An ALERT-severity check that produced findings is reported as
            # "ALERT", not "FAIL" — it is advisory and does not fail the dataset.
            if fs:
                status = "ALERT" if meta.severity == Severity.ALERT else "FAIL"
            else:
                status = "PASS"
            out.append((meta, status, fs))
        for meta, _exc in self.errors:
            out.append((meta, "ERROR", []))
        return out

    def per_check_by_severity(self):
        """Yield ``(severity, rows)`` in C, E, W order, where ``rows`` are the
        per_check() triples for checks of that severity, problems first
        (FAIL, then ERROR, then PASS), each block ordered by check name.
        Severities with no in-scope checks are skipped."""
        order = {"FAIL": 0, "ALERT": 0, "ERROR": 1, "PASS": 2}

        def _cat_prio(meta):
            # Product-specific Inland ENC FC checks rank above other-standard
            # checks within the same severity/status block.
            return 0 if meta.category == "Inland ENC Feature Catalogue" else 1

        results = self.per_check()
        for sev in Severity.ORDER:
            rows = [r for r in results if r[0].severity == sev]
            if rows:
                rows.sort(key=lambda r: (order.get(r[1], 9),
                                         _cat_prio(r[0]), r[0].name))
                yield sev, rows

    # -- exports --------------------------------------------------------------
    def to_check_csv(self):
        """The audit summary as CSV: one row per check, PASS/FAIL/ERROR."""
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(["Severity", "Status", "Check", "S-58", "Category",
                    "Findings", "Solution"])
        for sev, rows in self.per_check_by_severity():
            for meta, status, fs in rows:
                w.writerow([Severity.LABELS.get(sev, sev), status, meta.name,
                            meta.number, meta.category, len(fs), meta.solution])
        return buf.getvalue()

    def to_html(self, title="ENC audit report", source="", edition="",
                cells=None, notes=None, cell_meta=None, ienc_fcs=None):
        """A self-contained audit document: per-check PASS/FAIL grouped by
        severity, using each check's plain-language name, S-58 numbers as
        citations, with the offending features listed under each failure.

        ``source``  -- human label for what was audited (folder/zip path).
        ``edition`` -- profile/version manifest line for the header.
        ``cells``   -- iterable of cell names audited (so clean cells still show
                       in the header even when they raised no findings).
        ``notes``   -- optional iterable of caveat strings shown in a banner
                       (e.g. that CATALOG.031/exchange-set were not tested).
        """
        esc = html.escape
        results = self.per_check()
        n_pass = sum(1 for _m, s, _f in results if s == "PASS")
        n_fail = sum(1 for _m, s, _f in results if s == "FAIL")
        n_alert = sum(1 for _m, s, _f in results if s == "ALERT")
        n_err = sum(1 for _m, s, _f in results if s == "ERROR")
        n_unvalidated = sum(1 for m, _s, _f in results if not m.validated)
        c = self.counts()
        passed = self.passes_minimum_standard()
        verdict = "PASS" if passed else "FAIL"
        gen = (self.finished or datetime.now(timezone.utc)).strftime("%Y-%m-%d %H:%M UTC")
        if cells:
            cell_str = ", ".join(esc(str(x)) for x in cells)
        else:
            cell_str = ", ".join(sorted({esc(f.cell) for f in self.findings})) \
                or "&mdash;"

        sev_label = {"C": "Critical", "A": "Alert", "E": "Error", "W": "Warning"}
        sev_colour = {"C": "#c0392b", "A": "#7d3c98", "E": "#d68910",
                      "W": "#2471a3"}
        status_colour = {"PASS": "#1e8449", "FAIL": "#c0392b",
                         "ALERT": "#7d3c98", "ERROR": "#6c3483"}

        rows_html = []
        for sev, rows in self.per_check_by_severity():
            # Count problems generically (FAIL or ALERT) so the Alert section
            # header reads sensibly ("flagged" rather than "failed").
            n_f = sum(1 for _m, s, _f in rows if s in ("FAIL", "ALERT"))
            rows_html.append(
                f'<h2 style="color:{sev_colour.get(sev, "#000")}">'
                f'{sev_label.get(sev, sev)} checks '
                f'<span class="muted">({len(rows)} in scope, {n_f} flagged)'
                f'</span></h2>')
            rows_html.append('<table class="checks"><thead><tr>'
                             '<th>Status</th><th>Check</th>'
                             '<th>S-58</th><th>Findings</th></tr></thead><tbody>')
            for meta, status, fs in rows:
                sc = status_colour.get(status, "#000")
                rows_html.append(
                    f'<tr class="r-{status.lower()}">'
                    f'<td><span class="badge" style="background:{sc}">{status}'
                    f'</span></td>'
                    f'<td><div class="cname">{esc(meta.title)}'
                    f'{"" if meta.validated else " *"}'
                    f' <span class="muted">{esc(meta.name)}</span></div>'
                    f'<div class="cmsg">{esc(meta.message)}</div></td>'
                    f'<td class="num">{esc(str(meta.number))}</td>'
                    f'<td class="num">{len(fs)}</td></tr>')
                if fs:
                    items = []
                    for f in fs[:200]:
                        loc = f"{esc(f.cell)} &middot; {esc(f.klass)}"
                        if f.fid is not None:
                            loc += f" &middot; fid {esc(str(f.fid))}"
                        det = f" &mdash; {esc(f.detail)}" if f.detail else ""
                        cite = f.check.number
                        cn = (f' <span class="muted">[{esc(str(cite))}]</span>'
                              if str(cite) != str(meta.number) else "")
                        items.append(f"<li>{loc}{det}{cn}</li>")
                    more = (f'<li class="muted">… and {len(fs) - 200} more</li>'
                            if len(fs) > 200 else "")
                    rows_html.append(
                        f'<tr class="detail"><td></td><td colspan="3">'
                        f'<ul>{"".join(items)}{more}</ul>'
                        f'<div class="soln"><b>Solution:</b> '
                        f'{esc(meta.solution or "—")}</div></td></tr>')
            rows_html.append('</tbody></table>')

        err_html = ""
        if self.errors:
            lis = "".join(
                f"<li>{esc(m.name)} <span class='muted'>[#{esc(str(m.number))}]"
                f"</span>: {esc(str(e))}</li>" for m, e in self.errors)
            err_html = (
                '<h2 style="color:#6c3483">Checks that could not run</h2>'
                f'<ul class="errs">{lis}</ul>')

        timing_html = ""
        if self.timings:
            slow = self.slowest(8)
            lis = "".join(
                f"<li><span class='num'>{sec:0.2f}s</span> &nbsp; "
                f"{esc(m.name)} <span class='muted'>[#{esc(str(m.number))}]"
                f"{(', ' + str(nf) + ' finding(s)') if nf else ''}</span></li>"
                for m, sec, nf in slow)
            timing_html = (
                '<h2>Timing <span class="muted">(slowest checks; total '
                f'{self.total_seconds():0.1f}s)</span></h2>'
                f'<ul class="timing">{lis}</ul>')

        if notes:
            note_items = "".join(f"<li>{esc(str(n))}</li>" for n in notes)
            notes_html = (
                '<div class="notes"><b>Not tested:</b>'
                f'<ul>{note_items}</ul></div>')
        else:
            notes_html = ""

        if n_unvalidated:
            unvalidated_html = (
                '<div class="unval"><b>* Attention:</b> '
                f'{n_unvalidated} check(s) marked with an asterisk (*) were '
                'implemented from the S-58 document but are <b>not yet validated '
                'against an official S-58 test dataset</b>. Treat their findings '
                'as provisional pending that validation.</div>')
        else:
            unvalidated_html = ""

        # Coding & edition: the Inland ENC FC actually attached (selected by the
        # cells' PRED) and a per-cell DSID table — product spec (PRSP), edition
        # (PRED), cell edition (EDTN), issue date (ISDT), agency (AGEN).
        coding_html = ""
        if ienc_fcs:
            lines = []
            for fc in ienc_fcs:
                lines.append(
                    f'edition {esc(str(fc.get("edition") or "?"))} &rarr; '
                    f'{esc(str(fc.get("file") or ""))} '
                    f'<span class="muted">(PRED '
                    f'{esc(str(fc.get("pred") or "?"))}, '
                    f'{fc.get("cells", 0)} cell(s); {fc.get("features", 0)} '
                    f'classes, {fc.get("attributes", 0)} attrs'
                    + (f'; {esc(str(fc.get("why")))}' if fc.get("why") else "")
                    + ')</span>')
            label = ("Inland ENC FC used" if len(lines) == 1
                     else "Inland ENC FCs used (per PRED)")
            coding_html += (f'<div class="meta"><b>{label}:</b><br>'
                            + "<br>".join(lines) + "</div>")
        if cell_meta:
            preds = sorted({m.get("PRED") for m in cell_meta if m.get("PRED")})
            from collections import Counter as _Counter
            prod_counts = _Counter(m.get("product") or f'PRSP {m.get("PRSP")}'
                                   for m in cell_meta)
            prod_str = ", ".join(f"{esc(k)} ({n})"
                                 for k, n in sorted(prod_counts.items()))
            coding_html += (
                '<div class="meta"><b>Coding / edition:</b> '
                f'{prod_str or "&mdash;"} &middot; edition (PRED) '
                f'{", ".join(esc(p) for p in preds) or "&mdash;"}</div>')
            trs = "".join(
                "<tr>"
                f'<td>{esc(m.get("cell", ""))}</td>'
                f'<td>{esc(str(m.get("product", "") or ""))}</td>'
                f'<td class="num">{esc(str(m.get("PRSP", "") or ""))}</td>'
                f'<td class="num">{esc(str(m.get("PRED", "") or ""))}</td>'
                f'<td class="num">{esc(str(m.get("EDTN", "") or ""))}</td>'
                f'<td>{esc(str(m.get("ISDT", "") or ""))}</td>'
                f'<td class="num">{esc(str(m.get("AGEN", "") or ""))}</td>'
                "</tr>" for m in cell_meta)
            coding_html += (
                '<details><summary class="muted">Per-cell coding &amp; edition '
                f'({len(cell_meta)} cells)</summary>'
                '<table class="checks"><thead><tr><th>Cell</th>'
                '<th>Product</th><th>PRSP</th><th>Edition (PRED)</th>'
                '<th>Cell edn (EDTN)</th><th>Issued (ISDT)</th>'
                '<th>Agency (AGEN)</th></tr></thead><tbody>'
                f'{trs}</tbody></table></details>')

        banner_bg = "#1e8449" if passed else "#c0392b"
        return f"""<!DOCTYPE html>
<html lang="en"><head><meta charset="utf-8">
<title>{esc(title)}</title>
<style>
 body{{font:14px/1.5 -apple-system,Segoe UI,Roboto,Helvetica,Arial,sans-serif;
   color:#1a1a1a;max-width:1000px;margin:24px auto;padding:0 18px}}
 h1{{font-size:22px;margin:0 0 4px}}
 .banner{{color:#fff;background:{banner_bg};padding:10px 14px;border-radius:6px;
   font-weight:600;margin:14px 0}}
 .meta{{color:#444;font-size:13px;margin:8px 0 4px}}
 .meta b{{color:#1a1a1a}}
 .notes{{background:#fff8e1;border:1px solid #ffe082;border-radius:6px;
   padding:8px 14px;margin:12px 0;font-size:13px;color:#5d4037}}
 .notes ul{{margin:4px 0 0;padding-left:18px}}
 .unval{{background:#fdecea;border:1px solid #f5c6c0;border-radius:6px;
   padding:8px 14px;margin:12px 0;font-size:13px;color:#7a271a}}
 .tiles{{display:flex;gap:10px;flex-wrap:wrap;margin:12px 0 6px}}
 .tile{{border:1px solid #ddd;border-radius:6px;padding:8px 14px;min-width:90px}}
 .tile .n{{font-size:20px;font-weight:700}}
 .tile .l{{font-size:12px;color:#666}}
 h2{{font-size:16px;margin:22px 0 6px;border-bottom:1px solid #eee;
   padding-bottom:4px}}
 table.checks{{border-collapse:collapse;width:100%;margin:4px 0 8px}}
 table.checks th{{text-align:left;font-size:12px;color:#666;border-bottom:
   1px solid #ddd;padding:4px 8px}}
 table.checks td{{padding:5px 8px;vertical-align:top;border-bottom:1px solid #f2f2f2}}
 td.num{{text-align:right;font-variant-numeric:tabular-nums;white-space:nowrap}}
 .badge{{color:#fff;font-size:11px;font-weight:700;border-radius:4px;
   padding:2px 7px;letter-spacing:.3px}}
 .cname{{font-weight:600}}
 .cmsg{{color:#555;font-size:12px}}
 tr.detail td{{background:#fafafa}}
 tr.detail ul{{margin:4px 0;padding-left:18px}}
 tr.detail li{{font-size:12px;color:#333}}
 .soln{{font-size:12px;color:#444;margin-top:4px}}
 .muted{{color:#999;font-weight:400}}
 ul.errs li{{margin:3px 0}}
 ul.timing{{list-style:none;padding-left:0}}
 ul.timing li{{margin:2px 0;font-size:13px}}
 ul.timing .num{{display:inline-block;min-width:54px;text-align:right;
   font-variant-numeric:tabular-nums;font-weight:600}}
 footer{{margin:24px 0 8px;color:#999;font-size:11px;border-top:1px solid #eee;
   padding-top:8px}}
</style></head><body>
<h1>{esc(title)}</h1>
<div class="banner">{verdict} — minimum standard
 {"met" if passed else "NOT met"} ({c.get('C', 0)} critical finding(s))</div>
{notes_html}
<div class="meta"><b>Source:</b> {esc(source) or "&mdash;"}</div>
<div class="meta"><b>Cells:</b> {cell_str}</div>
<div class="meta"><b>Profile:</b> {esc(edition) or "&mdash;"}</div>
<div class="meta"><b>Generated:</b> {gen}</div>
{coding_html}
<div class="tiles">
 <div class="tile"><div class="n">{len(self.ran)}</div><div class="l">checks run</div></div>
 <div class="tile"><div class="n" style="color:#1e8449">{n_pass}</div><div class="l">passed</div></div>
 <div class="tile"><div class="n" style="color:#c0392b">{n_fail}</div><div class="l">failed</div></div>
 {('<div class="tile"><div class="n" style="color:#7d3c98">' + str(n_alert) + '</div><div class="l">alerts</div></div>') if n_alert else ''}
 <div class="tile"><div class="n" style="color:#6c3483">{n_err}</div><div class="l">could not run</div></div>
 <div class="tile"><div class="n">{len(self.findings)}</div><div class="l">findings</div></div>
</div>
{"".join(rows_html)}
{err_html}
{timing_html}
{unvalidated_html}
<footer>Generated by ENCoreGIS Auditor. Check names are the stable
 documentation keys; S-58 numbers and severities are citations against the
 stated profile and may change between editions.</footer>
{SIGNATURE_HTML}
</body></html>"""


# -- runner -------------------------------------------------------------------
class Validator:
    """Runs registered checks against a dataset and builds a Report."""

    def __init__(self, catalogue, geom_ops):
        self.catalogue = catalogue
        self.geom = geom_ops

    def run(self, dataset, numbers=None, severities=None, progress=None,
            on_check=None):
        """Execute checks against ``dataset``.

        numbers     -- optional iterable of check numbers to restrict to.
        severities  -- optional iterable of severities to restrict to.
        progress    -- optional callable(done, total, check_number) called
                       *before* each check starts (so the UI can show which one
                       is currently running — useful for slow checks).
        on_check    -- optional callable(done, total, meta, seconds, n_findings,
                       status) called *after* each check finishes; status is
                       "ok" or "error". Lets the UI log per-check timings live.

        Each check's wall-clock time is also recorded on ``report.timings`` as
        ``(meta, seconds, n_findings)`` regardless of the callbacks.
        """
        wanted = self._select(numbers, severities)
        report = Report()
        total = len(wanted)
        for i, meta in enumerate(wanted, 1):
            if progress:
                progress(i, total, meta.number)
            n_before = len(report.findings)
            t0 = perf_counter()
            status = "ok"
            try:
                for finding in meta.func(dataset, self.catalogue, self.geom):
                    finding.source = meta
                    report.add(finding)
                report.ran.append(meta)
            except Exception as exc:                      # noqa: BLE001
                # One faulty check must never abort the whole run.
                report.errors.append((meta, exc))
                status = "error"
            dt = perf_counter() - t0
            n_found = len(report.findings) - n_before
            report.timings.append((meta, dt, n_found))
            if on_check:
                on_check(i, total, meta, dt, n_found, status)
        report.finished = datetime.now(timezone.utc)
        return report

    @staticmethod
    def _select(numbers, severities):
        nums = set(numbers) if numbers else None
        sevs = set(severities) if severities else None
        out = []
        for meta in REGISTRY:
            if nums is not None and meta.number not in nums:
                continue
            if sevs is not None and meta.severity not in sevs:
                continue
            out.append(meta)
        return out


def load_all_checks():
    """Import the check modules so their @check decorators populate REGISTRY.

    Import is deferred to here (rather than at package import) so a caller can
    load the engine without dragging in every check module, and so the registry
    is built in a deterministic order.
    """
    from . import checks_attributes   # noqa: F401
    from . import checks_structure    # noqa: F401
    from . import checks_geometry     # noqa: F401
    from . import checks_records      # noqa: F401
    from . import checks_exchange     # noqa: F401
    from . import checks_ienc         # noqa: F401  (Inland ENC FC conformance)
    return REGISTRY
