# Copyright (C) 2026 Wilhelmus Dingenus de Pooter
# SPDX-License-Identifier: GPL-3.0-or-later
"""
validation/s100.py
------------------
The S-158 validation engine for S-100 **coverage** products carried in HDF5:
S-102 (bathymetric surface), S-104 (water level) and S-111 (surface currents).

Where the S-58 engine (``validation/engine.py``) validates S-57/S-101 *vector*
cells, this module validates the HDF5 datasets against the official IHO S-158
check catalogues (``s158/S158_{102,104,111}.json`` — extracted verbatim from the
IHO S-158-1xx Validation Checks workbooks). It is deliberately self-contained
(stdlib + GDAL only, no QGIS) so it is unit-testable off the GUI thread.

Design
======
* The **catalogue** is authoritative data, not code: every official check (ID,
  classification, message, description, solution, references) is bundled and
  loaded per product. This is how we guarantee "the correct validation checks".
* An **executable predicate** is a small function decorated with ``@check(...)``
  that cites the catalogue ID it implements *per product* (S-104 and S-111 share
  a numbering, so one predicate serves both). It receives an :class:`H5Dataset`
  and yields ``(location, detail)`` per offending element. The runner attaches
  the catalogue's classification/message/solution to each finding.
* Every catalogue check is reported with a status: FAIL (predicate found a
  violation), PASS (predicate ran, no violation) or NOT-IMPLEMENTED (no
  predicate yet) — so the report is honest about coverage.

The raw HDF5 structure is read through GDAL's multidimensional API
(``gdal.OpenEx(path, OF_MULTIDIM_RASTER)``) — no h5py dependency.
"""

import json
import os
import re


# -- product configuration ----------------------------------------------------
# Each S-100 coverage product has one feature-container group under the root;
# S-104/S-111 share the TWCWG check numbering, S-102 has its own.
PRODUCTS = {
    "101": {"spec": "S-101", "kind": "vector"},
    "102": {"spec": "S-102", "kind": "hdf5", "container": "BathymetryCoverage"},
    "104": {"spec": "S-104", "kind": "hdf5", "container": "WaterLevel"},
    "111": {"spec": "S-111", "kind": "hdf5", "container": "SurfaceCurrent"},
}

# S-158 classification -> local Severity code (matches validation.engine.Severity).
# The "gate" rows ("--", must be corrected before further checks) are treated as
# Critical; anything unmapped is a Warning.
SEVERITY = {
    "Critical Error": "C",
    "Error": "E",
    "Warning": "W",
    "--": "C",
    "?": "W",
    "None": "W",
}

_HERE = os.path.dirname(os.path.abspath(__file__))
_CAT_DIR = os.path.join(os.path.dirname(_HERE), "s158")   # plugin_root/s158
_CAT_CACHE = {}

# Check IDs whose executable predicate has been CONFIRMED against an official IHO
# S-158 test dataset (github.com/iho-ohi/S-1xx-Test-Datasets). Until a check is on
# this list it is treated as NOT YET VALIDATED: it still runs on every audit, but
# it is flagged with a trailing '*' in the tree and report (the same convention as
# the S-58 / Inland-ENC auditor), and its findings are provisional. Populate this
# as checks are calibrated against the official test data.
VALIDATED_IDS = {
    # Orientation/bearing family — calibrated against the S-158/V1 orientation test
    # datasets 00AA_00004 & 00AA_00005 (finding counts match the expected-results
    # reports exactly) AND confirmed to raise 0 findings across the 38 inland BE/NL
    # + 32 official S-101_1.2 conformant cells. 101_1286 is deliberately NOT here
    # yet: it fires one more than that dataset's report lists for S101_A_010.
    "101_1279",   # NavigationLine bearing (5°, reciprocal)
    "101_1287",   # RecommendedTrack two-way bearing (5°, reciprocal)
    "101_1293",   # RecommendedRouteCentreline one-way bearing (5°, direction)
    "101_1294",   # RecommendedRouteCentreline two-way bearing (5°, reciprocal)
    "101_1295",   # DeepWaterRouteCentreline one-way bearing (5°, direction)
    "101_1296",   # DeepWaterRouteCentreline two-way bearing (5°, reciprocal)
    # Skin-of-the-Earth coverage topology — calibrated against S-158/V1 00AA_00003
    # (S101_4_003 "gap in group 1 coverage" and S101_4_005 "data-coverage limit
    # references several SotE objects": one expected error each, matched exactly)
    # AND 0 findings across the 70 conformant cells. Siblings 0001/0003 stay
    # provisional (they fire on official cell DS0015, unverifiable without its own
    # expected-results report).
    "101_0002",   # SotE combined coverage != DataCoverage extent
    "101_0004",   # DataCoverage-boundary curve shared by >1 SotE feature
    # Mandatory meta feature present — 00AA_00008 deletes M_NSYS (its report:
    # "Mandatory feature objects are missing must be triggered"); we fire exactly
    # there and 0 across the 70 conformant cells. (1054/DataCoverage is not here:
    # no official dataset removes M_COVR, so it has no positive test yet.)
    "101_1056",   # NavigationalSystemOfMarks (M_NSYS) missing
    # Date-range order/identity — 00AA_00012 seeds Group-1 features with equal
    # fixedDateRange dates and identical periodicDateRange dates; these violate the
    # exact catalogue wording (1383 "dateEnd <= dateStart", 1385 "values are
    # identical"). Our checks fire ONLY there (0 across 70 conformant + every other
    # S-158/V1 cell). 1386/1387 stay provisional (they fire on other S-158 cells
    # not yet reconciled), and 1384/1392 have no positive test.
    "101_1383",   # fixedDateRange.dateEnd <= dateStart
    "101_1385",   # periodicDateRange.dateStart identical to dateEnd
}


def is_validated(cid):
    """True if check ``cid`` has been confirmed against an official IHO test
    dataset (so it needs no '*' provisional marker)."""
    return cid in VALIDATED_IDS


def catalogue(product):
    """The full ordered list of official S-158 checks for a product, each a dict
    (id, cls, message, description, solution, std_ref, clause, dq, …). Cached."""
    product = str(product)
    if product not in _CAT_CACHE:
        path = os.path.join(_CAT_DIR, "S158_%s.json" % product)
        with open(path, encoding="utf-8") as fh:
            _CAT_CACHE[product] = json.load(fh)
    return _CAT_CACHE[product]


def catalogue_index(product):
    """{check_id: record} for a product's catalogue."""
    return {r["id"]: r for r in catalogue(product) if r.get("id")}


# -- product detection --------------------------------------------------------
def detect_product(path):
    """Return the product code ('101'/'102'/'104'/'111') for a dataset file, or
    None. Cheap filename test first (the S-100 file-naming convention prefixes
    the product number), confirmed for HDF5 by the root ``productSpecification``
    when the name is ambiguous."""
    b = os.path.basename(path).lower()
    is_h5 = b.endswith((".h5", ".hdf5"))
    for code in ("102", "104", "111"):
        if b.startswith(code) and is_h5:
            return code
    if b.startswith("101") and (b.endswith(".000") or ".000" in b):
        return "101"
    if is_h5:                                            # sniff the spec attr
        try:
            spec = str(H5Dataset(path).open().root_attrs()
                       .get("productSpecification", ""))
            m = re.search(r"S-?1(02|04|11)", spec)
            if m:
                return "1" + m.group(1)
        except Exception:                                # noqa: BLE001
            return None
    return None


# -- HDF5 dataset model (GDAL multidimensional API) ---------------------------
def _gdal():
    from osgeo import gdal
    gdal.UseExceptions()
    return gdal


class H5Dataset:
    """Thin read-only view of an S-100 HDF5 dataset over GDAL's multidim API:
    root/group attributes as plain dicts, group navigation, and the feature
    container + its instance groups. Attribute values come back as float for
    numerics and str for text; compound arrays (which the SWIG bindings cannot
    buffer) are skipped rather than raising."""

    def __init__(self, path):
        self.path = path
        self._ds = None
        self.root = None

    def open(self):
        gdal = _gdal()
        self._ds = gdal.OpenEx(self.path, gdal.OF_MULTIDIM_RASTER)
        if self._ds is None:
            raise IOError("not an HDF5 dataset: %s" % self.path)
        self.root = self._ds.GetRootGroup()
        if self.root is None:
            raise IOError("no root group: %s" % self.path)
        return self

    # - attributes -
    @staticmethod
    def attrs(group):
        """{name: value} for a group's HDF5 attributes; unreadable ones (e.g.
        compound types the SWIG bindings can't buffer) are skipped."""
        out = {}
        if group is None:
            return out
        try:
            alist = group.GetAttributes()
        except Exception:                                # noqa: BLE001
            return out
        for a in alist:
            try:
                out[a.GetName()] = a.Read()
            except Exception:                            # noqa: BLE001
                continue
        return out

    def root_attrs(self):
        return self.attrs(self.root)

    # - navigation -
    def group(self, *path):
        """Open a nested group by name(s); None if any level is missing."""
        g = self.root
        for name in path:
            if g is None:
                return None
            try:
                g = g.OpenGroup(name)
            except Exception:                            # noqa: BLE001
                return None
        return g

    def group_names(self, group=None):
        g = self.root if group is None else group
        try:
            return list(g.GetGroupNames())
        except Exception:                                # noqa: BLE001
            return []

    def array_names(self, group=None):
        g = self.root if group is None else group
        try:
            return list(g.GetMDArrayNames())
        except Exception:                                # noqa: BLE001
            return []

    def container(self, product):
        """(name, group) of the product's feature container. Uses the configured
        name if present, else the sole non-Group_F child of the root."""
        want = PRODUCTS.get(product, {}).get("container")
        names = self.group_names()
        if want and want in names:
            return want, self.group(want)
        others = [n for n in names if n != "Group_F"]
        if len(others) == 1:
            return others[0], self.group(others[0])
        return (want or (others[0] if others else None),
                self.group(want) if want else None)

    @staticmethod
    def instances(container_group):
        """[(name, group)] of the NAME.01/.02… feature-instance groups."""
        out = []
        if container_group is None:
            return out
        try:
            names = list(container_group.GetGroupNames())
        except Exception:                                # noqa: BLE001
            return out
        for n in names:
            if re.search(r"\.\d{2,}$", n):
                out.append((n, container_group.OpenGroup(n)))
        return out

    @staticmethod
    def value_groups(instance_group):
        """[(name, group)] of the Group_NNN data-value groups in an instance."""
        out = []
        if instance_group is None:
            return out
        try:
            names = list(instance_group.GetGroupNames())
        except Exception:                                # noqa: BLE001
            return out
        for n in names:
            if re.match(r"Group_\d+$", n):
                out.append((n, instance_group.OpenGroup(n)))
        return out

    @staticmethod
    def has_array(group, name):
        """True if ``group`` holds an MD array called ``name``."""
        if group is None:
            return False
        try:
            return name in group.GetMDArrayNames()
        except Exception:                                # noqa: BLE001
            return False

    @staticmethod
    def array_len(group, name):
        """Length of the first dimension of ``group``'s ``name`` array, or None."""
        try:
            dims = group.OpenMDArray(name).GetDimensions()
            return dims[0].GetSize() if dims else None
        except Exception:                                # noqa: BLE001
            return None

    @staticmethod
    def read_string_array(group, name):
        """Read a 1-D string MD array as a list of str, or None if unreadable
        (compound/numeric buffers the SWIG bindings can't return raise → None)."""
        try:
            v = group.OpenMDArray(name).Read()
        except Exception:                                # noqa: BLE001
            return None
        if isinstance(v, (list, tuple)):
            return [str(x) for x in v]
        return None if v is None else [str(v)]

    # - compound-array introspection (Group_F feature info + values datasets) -
    @staticmethod
    def array_rank(group, name):
        """Number of dimensions of ``group``'s ``name`` array, or None."""
        try:
            return len(group.OpenMDArray(name).GetDimensions())
        except Exception:                                # noqa: BLE001
            return None

    @staticmethod
    def array_dims(group, name):
        """Dimension sizes of ``group``'s ``name`` array as a list, or None."""
        try:
            return [d.GetSize() for d in group.OpenMDArray(name).GetDimensions()]
        except Exception:                                # noqa: BLE001
            return None

    def compound_components(self, group, name):
        """``[(component_name, type_label)]`` for a COMPOUND MD array (type_label =
        ``'string'`` / a GDAL numeric name like ``'Float32'`` / ``'compound'``), or
        ``None`` if the array is missing or not a compound. This is the piece the
        SWIG buffer read cannot give us — the compound STRUCTURE reads fine even
        when its VALUES do not."""
        gdal = _gdal()
        try:
            edt = group.OpenMDArray(name).GetDataType()
            if edt.GetClass() != gdal.GEDTC_COMPOUND:
                return None
            out = []
            for c in edt.GetComponents():
                t = c.GetType()
                if t.GetClass() == gdal.GEDTC_STRING:
                    label = "string"
                elif t.GetClass() == gdal.GEDTC_NUMERIC:
                    label = gdal.GetDataTypeName(t.GetNumericDataType())
                else:
                    label = "compound"
                out.append((c.GetName(), label))
            return out
        except Exception:                                # noqa: BLE001
            return None

    def read_fids(self, group, name):
        """The Group_F feature-information dataset as
        ``{attribute_code: {field: value}}`` — read one string component at a time
        via ``GetView('["field"]')``. The whole compound-of-strings cannot be
        buffered by the SWIG bindings, but each single string field extracts
        cleanly. Fields per S-100 Table 10c-8: code, name, uom.name, fillValue,
        datatype, lower, upper, closure."""
        try:
            arr = group.OpenMDArray(name)
        except Exception:                                # noqa: BLE001
            return {}
        got = {}
        for f in _FIDS_FIELDS:
            try:
                got[f] = [str(x) for x in arr.GetView('["%s"]' % f).Read()]
            except Exception:                            # noqa: BLE001
                got[f] = []
        n = len(got.get("code", []))
        out = {}
        for i in range(n):
            out[got["code"][i]] = {k: (got[k][i] if i < len(got[k]) else "")
                                   for k in _FIDS_FIELDS}
        return out

    def read_values_struct(self, group, name):
        """A numeric-compound values array as a numpy structured array (fields
        named by component), or ``None``. Unlike the all-string FIDS, an all-numeric
        compound's ``Read()`` returns the raw packed bytes, which we frame with the
        component dtypes (verified identical to the native S-111 driver). ``None``
        if numpy is absent, the array is not an all-numeric compound, or the read
        fails."""
        try:
            import numpy as np
        except Exception:                                # noqa: BLE001
            return None
        gdal = _gdal()
        try:
            arr = group.OpenMDArray(name)
            edt = arr.GetDataType()
            if edt.GetClass() != gdal.GEDTC_COMPOUND:
                return None
            names, formats, offsets = [], [], []
            for c in edt.GetComponents():
                t = c.GetType()
                if t.GetClass() != gdal.GEDTC_NUMERIC:
                    return None
                npt = _GDAL_NP.get(gdal.GetDataTypeName(t.GetNumericDataType()))
                if npt is None:
                    return None
                names.append(c.GetName())
                formats.append(npt)
                offsets.append(c.GetOffset())
            # Honour the compound member OFFSETS + total size: an HDF5 compound is
            # not necessarily tightly packed (e.g. {Float32, Int16, Float32} is
            # padded), so a packed dtype would mis-frame every record.
            dt = np.dtype({"names": names, "formats": formats,
                           "offsets": offsets, "itemsize": edt.GetSize()})
            return np.frombuffer(bytes(arr.Read()), dtype=dt)
        except Exception:                                # noqa: BLE001
            return None


# -- checks: registry + decorator ---------------------------------------------
# CHECKS[product] = [ (check_id, predicate) ]; predicate(ds, ctx) -> iter of
# (location, detail).  A predicate that yields nothing PASSED.
CHECKS = {"102": [], "104": [], "111": []}


def check(ids):
    """Register a predicate as the implementation of one catalogue check per
    product. ``ids`` maps product code -> catalogue check ID, e.g.
    ``{"111": "111_1001", "104": "104_1001", "102": "102_1001"}``. The predicate
    is shared across all listed products (S-104/S-111 are structurally identical;
    S-102 shares the metadata/structure checks)."""
    def _wrap(fn):
        for product, cid in ids.items():
            CHECKS.setdefault(product, []).append((cid, fn))
        fn._s158_ids = ids
        return fn
    return _wrap


# -- helpers for predicates ----------------------------------------------------
def _num(v):
    """Coerce an attribute value to float, or None."""
    if v is None:
        return None
    if isinstance(v, (list, tuple)):
        v = v[0] if v else None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def _str(v):
    if isinstance(v, (list, tuple)):
        v = v[0] if v else ""
    return "" if v is None else str(v)


def _is_int_gt0(v):
    """True if ``v`` is an integer value strictly greater than zero (attribute
    numerics arrive as float, e.g. 325.0)."""
    f = _num(v)
    return f is not None and f > 0 and float(f).is_integer()


def _is_datetime(v):
    """True if ``v`` is an S-100 compact ISO-8601 DateTime, e.g.
    ``20240723T230000Z`` (trailing Z optional)."""
    return bool(re.match(r"^\d{8}T\d{6}Z?$", _str(v).strip()))


# Root general-metadata attributes required on every S-100 coverage dataset.
_REQ_ROOT = ("productSpecification", "issueDate", "horizontalCRS",
             "westBoundLongitude", "eastBoundLongitude",
             "southBoundLatitude", "northBoundLatitude")

# The Group_F feature-information dataset's compound members (S-100 Table 10c-8).
_FIDS_FIELDS = ("code", "name", "uom.name", "fillValue", "datatype", "lower",
                "upper", "closure")

# GDAL numeric type name -> numpy dtype code, for framing a values compound.
_GDAL_NP = {"Float32": "<f4", "Float64": "<f8", "Int32": "<i4", "Int16": "<i2",
            "UInt32": "<u4", "UInt16": "<u2", "Byte": "u1", "Int8": "i1"}


@check({"111": "111_1001", "104": "104_1001", "102": "102_1001"})
def chk_feature_information_group(ds, ctx):
    """No feature information group (Group_F) under the root."""
    if "Group_F" not in ds.group_names():
        yield ("/", "root group has no Group_F feature information group")


@check({"111": "111_1002", "104": "104_1002", "102": "102_1002"})
def chk_general_metadata_present(ds, ctx):
    """Mandatory General Metadata attribute missing from the root group."""
    attrs = ds.root_attrs()
    for name in _REQ_ROOT:
        if name not in attrs or _str(attrs.get(name)).strip() == "":
            yield ("/", "missing mandatory root metadata attribute '%s'" % name)


@check({"111": "111_1004", "104": "104_1004", "102": "102_1006"})
def chk_general_metadata_values(ds, ctx):
    """General Metadata attribute value out of range / wrong string format."""
    a = ds.root_attrs()
    w, e = _num(a.get("westBoundLongitude")), _num(a.get("eastBoundLongitude"))
    s, n = _num(a.get("southBoundLatitude")), _num(a.get("northBoundLatitude"))
    for label, v in (("westBoundLongitude", w), ("eastBoundLongitude", e)):
        if v is not None and not (-180.0 <= v <= 180.0):
            yield ("/", "%s out of range: %s" % (label, v))
    for label, v in (("southBoundLatitude", s), ("northBoundLatitude", n)):
        if v is not None and not (-90.0 <= v <= 90.0):
            yield ("/", "%s out of range: %s" % (label, v))
    if w is not None and e is not None and w > e:
        yield ("/", "westBoundLongitude (%s) > eastBoundLongitude (%s)" % (w, e))
    if s is not None and n is not None and s > n:
        yield ("/", "southBoundLatitude (%s) > northBoundLatitude (%s)" % (s, n))


@check({"111": "111_1004", "104": "104_1004", "102": "102_1005"})
def chk_date_format(ds, ctx):
    """Wrong format for a date/time general-metadata attribute (issueDate must be
    YYYYMMDD; issueTime, when present, HHMMSS[Z])."""
    a = ds.root_attrs()
    idt = _str(a.get("issueDate"))
    if idt and not re.match(r"^\d{8}$", idt):
        yield ("/", "issueDate not YYYYMMDD: %r" % idt)
    itm = _str(a.get("issueTime"))
    if itm and not re.match(r"^\d{6}(\.\d+)?Z?$", itm):
        yield ("/", "issueTime not HHMMSS[Z]: %r" % itm)


@check({"111": "111_2001", "104": "104_2001", "102": "102_2001"})
def chk_data_coding_format(ds, ctx):
    """Missing or invalid feature container attribute dataCodingFormat."""
    name, g = ds.container(ctx["product"])
    if g is None:
        yield ("/", "feature container group is missing")
        return
    dcf = _num(ds.attrs(g).get("dataCodingFormat"))
    if dcf is None or not (1 <= int(dcf) <= 9):
        yield ("/%s" % name, "dataCodingFormat missing or invalid: %s" % dcf)


@check({"111": "111_2002", "104": "104_2002"})
def chk_dimension(ds, ctx):
    """Missing or invalid feature container attribute dimension."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    dim = _num(ds.attrs(g).get("dimension"))
    if dim is None or int(dim) not in (1, 2, 3):
        yield ("/%s" % name, "dimension missing or invalid: %s" % dim)


@check({"111": "111_2015", "104": "104_2015"})
def chk_num_instances(ds, ctx):
    """Feature-instance group count does not match numInstances."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    declared = _num(ds.attrs(g).get("numInstances"))
    actual = len(ds.instances(g))
    if declared is not None and int(declared) != actual:
        yield ("/%s" % name,
               "numInstances=%d but %d instance group(s) present"
               % (int(declared), actual))


@check({"111": "111_2011", "104": "104_2011"})
def chk_minmax_present(ds, ctx):
    """min/maxDataset* speed/height attribute missing or invalid."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    a = ds.attrs(g)
    lo = next((k for k in a if k.startswith("minDataset")), None)
    hi = next((k for k in a if k.startswith("maxDataset")), None)
    if lo is None or hi is None:
        yield ("/%s" % name, "min/maxDataset* attribute missing")


@check({"111": "111_2012", "104": "104_2012"})
def chk_minmax_order(ds, ctx):
    """minDataset* greater than maxDataset*."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    a = ds.attrs(g)
    lo = next((_num(a[k]) for k in a if k.startswith("minDataset")), None)
    hi = next((_num(a[k]) for k in a if k.startswith("maxDataset")), None)
    if lo is not None and hi is not None and lo > hi:
        yield ("/%s" % name, "minDataset* (%s) > maxDataset* (%s)" % (lo, hi))


@check({"111": "111_3002", "104": "104_3002", "102": "102_3002"})
def chk_instance_bbox(ds, ctx):
    """Feature-instance bounding-box coordinates invalid (W>E or S>N)."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        w, e = _num(a.get("westBoundLongitude")), _num(a.get("eastBoundLongitude"))
        s, n = _num(a.get("southBoundLatitude")), _num(a.get("northBoundLatitude"))
        if w is not None and e is not None and w > e:
            yield (iname, "west (%s) > east (%s)" % (w, e))
        if s is not None and n is not None and s > n:
            yield (iname, "south (%s) > north (%s)" % (s, n))


@check({"111": "111_3010", "104": "104_3010", "102": "102_3005"})
def chk_grid_origin(ds, ctx):
    """gridOrigin longitude/latitude missing, or (geographic CRS only) out of
    range. S-102 may use a PROJECTED CRS, where the grid origin is in metres, so
    the lon/lat range test only applies when the horizontal CRS is geographic
    (EPSG:4326 — always the case for S-104/S-111)."""
    geographic = int(_num(ds.root_attrs().get("horizontalCRS")) or 4326) == 4326
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        lon = _num(a.get("gridOriginLongitude"))
        lat = _num(a.get("gridOriginLatitude"))
        if lon is None or lat is None:
            yield (iname, "gridOriginLongitude/Latitude missing")
        elif geographic:
            if not (-180.0 <= lon <= 180.0):
                yield (iname, "gridOriginLongitude out of range: %s" % lon)
            if not (-90.0 <= lat <= 90.0):
                yield (iname, "gridOriginLatitude out of range: %s" % lat)


@check({"111": "111_3013", "104": "104_3013"})
def chk_num_points(ds, ctx):
    """numPointsLongitudinal/Latitudinal missing or not a positive integer."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        for key in ("numPointsLongitudinal", "numPointsLatitudinal"):
            v = _num(a.get(key))
            if v is None or int(v) <= 0:
                yield (iname, "%s missing or <= 0: %s" % (key, v))


# -- feature-container checks (2000s) — shared S-111 / S-104 -------------------
@check({"111": "111_2003", "104": "104_2003"})
def chk_common_point_rule(ds, ctx):
    """Attribute commonPointRule is missing, invalid, or not an allowed value."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    v = _num(ds.attrs(g).get("commonPointRule"))
    if v is None or int(v) not in (1, 2, 3, 4, 5):
        yield ("/%s" % name, "commonPointRule missing or invalid: %s" % v)


@check({"111": "111_2005", "104": "104_2005"})
def chk_position_uncertainty(ds, ctx):
    """Horizontal or vertical position uncertainty attribute missing/invalid."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    a = ds.attrs(g)
    for key in ("horizontalPositionUncertainty", "verticalUncertainty"):
        if _num(a.get(key)) is None:
            yield ("/%s" % name, "%s missing or non-numeric" % key)


@check({"111": "111_2013", "104": "104_2013", "102": "102_2003"})
def chk_axis_names(ds, ctx):
    """The feature container has no axisNames dataset, or its length does not
    match the stated dimension."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    if not ds.has_array(g, "axisNames"):
        yield ("/%s" % name, "axisNames dataset missing")
        return
    dim = _num(ds.attrs(g).get("dimension"))
    n = ds.array_len(g, "axisNames")
    if dim is not None and n is not None and int(dim) != int(n):
        yield ("/%s" % name, "axisNames length %d != dimension %d"
               % (int(n), int(dim)))


@check({"111": "111_2014", "104": "104_2014", "102": "102_1012"})
def chk_has_instances(ds, ctx):
    """No feature-instance groups in the feature container."""
    name, g = ds.container(ctx["product"])
    if g is not None and not ds.instances(g):
        yield ("/%s" % name, "no feature-instance groups present")


@check({"111": "111_2016", "104": "104_2016"})
def chk_sequencing_type(ds, ctx):
    """Sequencing rule type not present or of the wrong datatype."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    if _num(ds.attrs(g).get("sequencingRule.type")) is None:
        yield ("/%s" % name, "sequencingRule.type missing or non-numeric")


@check({"111": "111_2017", "104": "104_2017"})
def chk_sequencing_scandir(ds, ctx):
    """Sequencing rule scanDirection not present."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    if not _str(ds.attrs(g).get("sequencingRule.scanDirection")).strip():
        yield ("/%s" % name, "sequencingRule.scanDirection missing")


@check({"111": "111_2019", "104": "104_2019"})
def chk_interpolation_required(ds, ctx):
    """interpolationType is mandatory for dataCodingFormat = 2 or 3."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    a = ds.attrs(g)
    dcf = _num(a.get("dataCodingFormat"))
    if dcf is not None and int(dcf) in (2, 3) \
            and _num(a.get("interpolationType")) is None:
        yield ("/%s" % name,
               "interpolationType mandatory for dataCodingFormat %d" % int(dcf))


# -- feature-instance checks (3000s) — shared S-111 / S-104 -------------------
@check({"111": "111_3001", "104": "104_3001", "102": "102_3001"})
def chk_instance_bbox_present(ds, ctx):
    """Bounding-box attribute(s) missing in a feature instance."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        for key in ("westBoundLongitude", "eastBoundLongitude",
                    "southBoundLatitude", "northBoundLatitude"):
            if _num(a.get(key)) is None:
                yield (iname, "bounding-box attribute %s missing" % key)


@check({"111": "111_3003", "104": "104_3003"})
def chk_number_of_times(ds, ctx):
    """Missing numberOfTimes, or value not an integer > 0."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        if not _is_int_gt0(ds.attrs(ig).get("numberOfTimes")):
            yield (iname, "numberOfTimes missing or not an integer > 0")


@check({"111": "111_3004", "104": "104_3004"})
def chk_time_record_interval(ds, ctx):
    """Missing timeRecordInterval, or value not an integer > 0."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        if not _is_int_gt0(ds.attrs(ig).get("timeRecordInterval")):
            yield (iname, "timeRecordInterval missing or not an integer > 0")


@check({"111": "111_3005", "104": "104_3005"})
def chk_datetime_records(ds, ctx):
    """dateTimeOfFirstRecord / dateTimeOfLastRecord missing or not a DateTime."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        for key in ("dateTimeOfFirstRecord", "dateTimeOfLastRecord"):
            if not _is_datetime(a.get(key)):
                yield (iname, "%s missing or not a DateTime: %r"
                       % (key, _str(a.get(key))))


@check({"111": "111_3006", "104": "104_3006"})
def chk_datetime_order(ds, ctx):
    """First record DateTime is after the last record DateTime."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        f = _str(a.get("dateTimeOfFirstRecord")).rstrip("Z")
        l = _str(a.get("dateTimeOfLastRecord")).rstrip("Z")
        if _is_datetime(a.get("dateTimeOfFirstRecord")) \
                and _is_datetime(a.get("dateTimeOfLastRecord")) and f > l:
            yield (iname, "dateTimeOfFirstRecord after dateTimeOfLastRecord")


@check({"111": "111_3007", "104": "104_3007"})
def chk_num_grp(ds, ctx):
    """Missing numGRP, or value not an integer > 0."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        if not _is_int_gt0(ds.attrs(ig).get("numGRP")):
            yield (iname, "numGRP missing or not an integer > 0")


@check({"111": "111_3008", "104": "104_3008"})
def chk_data_dynamicity(ds, ctx):
    """Missing or invalid attribute dataDynamicity."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        if _num(ds.attrs(ig).get("dataDynamicity")) is None:
            yield (iname, "dataDynamicity missing or non-numeric")


@check({"111": "111_3011", "104": "104_3011", "102": "102_3006"})
def chk_grid_spacing(ds, ctx):
    """gridSpacingLongitudinal / gridSpacingLatitudinal missing or <= 0."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        for key in ("gridSpacingLongitudinal", "gridSpacingLatitudinal"):
            v = _num(a.get(key))
            if v is None or v <= 0:
                yield (iname, "%s missing or <= 0: %s" % (key, v))


@check({"111": "111_3025", "104": "104_3025"})
def chk_has_value_groups(ds, ctx):
    """No data-value groups (Group_NNN) in a feature instance."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        if not ds.value_groups(ig):
            yield (iname, "no data-value groups (Group_NNN) present")


@check({"111": "111_3026", "104": "104_3026"})
def chk_value_group_count(ds, ctx):
    """Count of value groups does not match numGRP."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        declared = _num(ds.attrs(ig).get("numGRP"))
        actual = len(ds.value_groups(ig))
        if declared is not None and int(declared) != actual:
            yield (iname, "numGRP=%d but %d value group(s) present"
                   % (int(declared), actual))


# -- data-value-group checks (5000s) — shared S-111 / S-104 -------------------
@check({"111": "111_5001", "104": "104_5001"})
def chk_timepoint(ds, ctx):
    """A data-value group has no timePoint attribute, or it is not a DateTime."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        for vname, vg in ds.value_groups(ig):
            if not _is_datetime(ds.attrs(vg).get("timePoint")):
                yield ("%s/%s" % (iname, vname),
                       "timePoint missing or not a DateTime")


@check({"111": "111_5015", "104": "104_5005", "102": "102_5001"})
def chk_values_dataset(ds, ctx):
    """A data-value group has no values dataset."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        for vname, vg in ds.value_groups(ig):
            if not ds.has_array(vg, "values"):
                yield ("%s/%s" % (iname, vname), "values dataset missing")


@check({"111": "111_2007", "104": "104_2007"})
def chk_num_instances_present(ds, ctx):
    """Attribute numInstances is missing or not a positive integer."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    if not _is_int_gt0(ds.attrs(g).get("numInstances")):
        yield ("/%s" % name, "numInstances missing or not an integer > 0")


@check({"111": "111_2022", "104": "104_2022"})
def chk_interpolation_not_used(ds, ctx):
    """interpolationType should not be present for dataCodingFormat 1, 4 or 8."""
    name, g = ds.container(ctx["product"])
    if g is None:
        return
    a = ds.attrs(g)
    dcf = _num(a.get("dataCodingFormat"))
    if dcf is not None and int(dcf) in (1, 4, 8) \
            and _num(a.get("interpolationType")) is not None:
        yield ("/%s" % name,
               "interpolationType present but not used for dataCodingFormat %d"
               % int(dcf))


@check({"111": "111_3016", "104": "104_3016"})
def chk_start_sequence(ds, ctx):
    """Missing startSequence in a feature instance."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        if not _str(ds.attrs(ig).get("startSequence")).strip():
            yield (iname, "startSequence missing")


@check({"111": "111_5002", "104": "104_5002"})
def chk_timepoint_in_range(ds, ctx):
    """A value group's timePoint is outside the [first, last] record time range."""
    _n, g = ds.container(ctx["product"])
    for iname, ig in ds.instances(g):
        a = ds.attrs(ig)
        if not (_is_datetime(a.get("dateTimeOfFirstRecord"))
                and _is_datetime(a.get("dateTimeOfLastRecord"))):
            continue
        first = _str(a.get("dateTimeOfFirstRecord")).rstrip("Z")
        last = _str(a.get("dateTimeOfLastRecord")).rstrip("Z")
        for vname, vg in ds.value_groups(ig):
            tp_raw = ds.attrs(vg).get("timePoint")
            tp = _str(tp_raw).rstrip("Z")
            if _is_datetime(tp_raw) and not (first <= tp <= last):
                yield ("%s/%s" % (iname, vname),
                       "timePoint %s outside [%s, %s]" % (tp, first, last))


# -- S-102-specific checks ----------------------------------------------------
@check({"102": "102_1010"})
def chk_vertical_cs_102(ds, ctx):
    """Attribute verticalCS must be the EPSG code 6498 (S-102 §Table)."""
    v = _num(ds.root_attrs().get("verticalCS"))
    if v is not None and int(v) != 6498:
        yield ("/", "verticalCS is %s, must be EPSG 6498" % int(v))


@check({"102": "102_1011"})
def chk_bathy_featurecode_102(ds, ctx):
    """Bathymetry data feature missing from the Group_F featureCode array."""
    gf = ds.group("Group_F")
    if gf is None:
        return
    codes = ds.read_string_array(gf, "featureCode")
    if codes is not None and "BathymetryCoverage" not in codes:
        yield ("/Group_F", "featureCode does not list BathymetryCoverage: %s"
               % codes)


@check({"111": "111_9005", "104": "104_9004", "102": "102_9004"})
def chk_file_size(ds, ctx):
    """Dataset file exceeds the 10 MB navigation-product guidance."""
    try:
        mb = os.path.getsize(ds.path) / (1024.0 * 1024.0)
    except OSError:
        return
    if mb > 10.0:
        yield (os.path.basename(ds.path), "file is %.1f MB (> 10 MB)" % mb)


# -- discovery-metadata (exchange-set catalogue) checks -----------------------
# These cross-check the HDF5 dataset against the S100_ROOT/CATALOG.XML discovery
# metadata. They skip gracefully in "loose mode" (no catalogue found), exactly
# like the S-58 exchange-set checks.
def _catalog(ds, ctx):
    """The exchange-set catalogue governing this dataset (parsed once, cached on
    ``ctx``), or ``None`` in loose mode."""
    if "_catalog" in ctx:
        return ctx["_catalog"]
    cat = None
    try:
        from . import s158_catalog
        cat = s158_catalog.load_for(ds.path)
    except Exception:                                    # noqa: BLE001
        cat = None
    ctx["_catalog"] = cat
    return cat


# Conservative bbox tolerance (degrees). The discovery boundingBox and the HDF5
# root bbox are meant to be identical, but valid S-102 samples show a sub-cell
# latitude convention offset up to ~1.3e-3° (longitude matches to ~1e-7). So the
# check flags only a MATERIAL mismatch (transposed / wrong-tile / degrees-off
# bbox), not that noise — provisional pending the exact root-vs-discovery rule.
_BBOX_TOL = 5e-3


@check({"111": "111_9002", "104": "104_9002", "102": "102_9002"})
def chk_bbox_matches_discovery(ds, ctx):
    """The S100_DatasetDiscoveryMetadata boundingBox differs from the HDF5 root
    group bounding box (PS 12.2). Compared with a conservative tolerance, so a
    material mismatch is reported while a sub-cell rounding/convention offset is
    not. Skipped in loose mode (no catalogue)."""
    cat = _catalog(ds, ctx)
    if cat is None:
        return
    dd = cat.for_file(ds.path)
    if dd is None or dd.bbox is None:
        return
    a = ds.root_attrs()
    root = (_num(a.get("westBoundLongitude")), _num(a.get("eastBoundLongitude")),
            _num(a.get("southBoundLatitude")), _num(a.get("northBoundLatitude")))
    if any(v is None for v in root):
        return
    labels = ("westBoundLongitude", "eastBoundLongitude",
              "southBoundLatitude", "northBoundLatitude")
    for lbl, disc, dat in zip(labels, dd.bbox, root):
        if abs(disc - dat) > _BBOX_TOL:
            yield ("CATALOG.XML",
                   "discovery %s (%s) differs from the dataset root value (%s)"
                   % (lbl, disc, dat))


@check({"102": "102_9003"})
def chk_support_files_prohibited(ds, ctx):
    """S-102 uses no support files (PS 11.3, 12.5): a support-file block in the
    exchange catalogue is prohibited. Skipped in loose mode (no catalogue)."""
    cat = _catalog(ds, ctx)
    if cat is None:
        return
    if cat.support_file_count > 0:
        yield ("CATALOG.XML",
               "%d support-file block(s) in the exchange catalogue; S-102 uses "
               "no support files" % cat.support_file_count)


# -- Group_F feature-information + values-dataset compound checks -------------
# The compound STRUCTURE reads fine via GDAL multidim even though the compound
# VALUES buffer does not: the all-string Group_F FIDS is read one component at a
# time (read_fids/GetView) and the values dataset's members via GetComponents.
def _feature_codes(ds):
    gf = ds.group("Group_F")
    if gf is None:
        return gf, []
    return gf, (ds.read_string_array(gf, "featureCode") or [])


@check({"111": "111_1005", "104": "104_1005"})
def chk_featurecode_array(ds, ctx):
    """Group_F has no 1-dimensional 'featureCode' array."""
    gf = ds.group("Group_F")
    if gf is None:                                       # missing Group_F → 1001
        return
    if not ds.has_array(gf, "featureCode"):
        yield ("/Group_F", "no 'featureCode' array in the feature information group")
        return
    r = ds.array_rank(gf, "featureCode")
    if r is not None and r != 1:
        yield ("/Group_F/featureCode",
               "featureCode is not 1-dimensional (rank=%d)" % r)


@check({"111": "111_1006", "104": "104_1006"})
def chk_featurecode_entries(ds, ctx):
    """Group_F.featureCode array contains no entries."""
    gf = ds.group("Group_F")
    if gf is None or not ds.has_array(gf, "featureCode"):
        return
    if not (ds.read_string_array(gf, "featureCode") or []):
        yield ("/Group_F/featureCode", "featureCode array has no entries")


@check({"111": "111_1009", "104": "104_1009"})
def chk_fids_component_type(ds, ctx):
    """A feature-information dataset's compound members differ from S-100
    Table 10c-8 (code, name, uom.name, fillValue, datatype, lower, upper,
    closure)."""
    gf, codes = _feature_codes(ds)
    for fname in codes:
        comps = ds.compound_components(gf, fname)
        if comps is None:                                # missing/not compound
            continue
        names = {c[0] for c in comps}
        if names != set(_FIDS_FIELDS):
            yield ("/Group_F/%s" % fname,
                   "feature-information compound members %s do not match S-100 "
                   "Table 10c-8 %s" % (sorted(names), sorted(_FIDS_FIELDS)))


@check({"111": "111_5023", "104": "104_5009"})
def chk_values_member_names(ds, ctx):
    """The values dataset members' component names do not correspond to the
    attributes in the feature-information dataset (Group_F FIDS)."""
    gf, codes = _feature_codes(ds)
    attr_codes = set()
    for fname in codes:
        attr_codes |= set(ds.read_fids(gf, fname).keys())
    if not attr_codes:
        return
    _n, cont = ds.container(ctx["product"])
    if cont is None:
        return
    for iname, inst in ds.instances(cont):
        for gname, g in ds.value_groups(inst):
            comps = ds.compound_components(g, "values")
            if comps is None:                            # not compound → other id
                continue
            names = {c[0] for c in comps}
            if names != attr_codes:
                yield ("%s/%s/values" % (iname, gname),
                       "values components %s do not match the feature-information "
                       "attributes %s" % (sorted(names), sorted(attr_codes)))


# FIDS datatype string -> broad class; values numeric type name -> broad class.
_FIDS_DT_CLASS = {"H5T_FLOAT": "float", "H5T_NATIVE_FLOAT": "float",
                  "H5T_INTEGER": "int", "H5T_ENUM": "int"}
_NP_DT_CLASS = {"Float32": "float", "Float64": "float", "Int32": "int",
                "Int16": "int", "UInt32": "int", "UInt16": "int",
                "Byte": "int", "Int8": "int"}


@check({"111": "111_5022", "104": "104_5008"})
def chk_values_member_types(ds, ctx):
    """A values dataset member's datatype does not correspond to its
    feature-information dataset attribute (float vs integer). Only checked when the
    member names already match the FIDS attributes (a name mismatch is 5023/5009)."""
    gf, codes = _feature_codes(ds)
    fids = {}
    for fname in codes:
        fids.update(ds.read_fids(gf, fname))
    if not fids:
        return
    _n, cont = ds.container(ctx["product"])
    if cont is None:
        return
    for iname, inst in ds.instances(cont):
        for gname, g in ds.value_groups(inst):
            comps = ds.compound_components(g, "values")
            if comps is None:
                continue
            comp_cls = {n: _NP_DT_CLASS.get(t) for n, t in comps}
            if set(comp_cls) != set(fids):               # name mismatch → 5023/5009
                continue
            for code, rec in fids.items():
                want = _FIDS_DT_CLASS.get((rec.get("datatype") or "").strip())
                got = comp_cls.get(code)
                if want and got and want != got:
                    yield ("%s/%s/values" % (iname, gname),
                           "component '%s' is %s but its FIDS datatype is %s"
                           % (code, got, rec.get("datatype")))


def _fids_range_violations(ds, ctx, attr_code, human):
    """Yield ONE finding per feature instance where the ``attr_code`` component of
    the values datasets falls outside the range encoded in the Group_F FIDS for
    that attribute (excluding the FIDS fillValue). Bounds are treated INCLUSIVELY
    (a value exactly on a bound passes) — conservative, so only genuinely
    out-of-range values are reported. Findings are AGGREGATED (a count + a few
    examples) rather than one per cell."""
    try:
        import numpy as np
    except Exception:                                    # noqa: BLE001
        return
    gf, codes = _feature_codes(ds)
    fids = {}
    for fname in codes:
        fids.update(ds.read_fids(gf, fname))
    entry = fids.get(attr_code)
    if entry is None:
        return
    lower, upper = _num(entry.get("lower")), _num(entry.get("upper"))
    fill = _num(entry.get("fillValue"))
    if lower is None and upper is None:
        return
    _n, cont = ds.container(ctx["product"])
    if cont is None:
        return
    for iname, inst in ds.instances(cont):
        n_bad = n_total = 0
        examples = []
        for _gn, g in ds.value_groups(inst):
            arr = ds.read_values_struct(g, "values")
            if arr is None or attr_code not in (arr.dtype.names or ()):
                continue
            v = arr[attr_code]
            bad = np.zeros(v.shape, dtype=bool)
            if lower is not None:
                bad |= v < lower
            if upper is not None:
                bad |= v > upper
            if fill is not None:
                bad &= v != fill
            cnt = int(bad.sum())
            n_bad += cnt
            n_total += int(v.size)
            if cnt and len(examples) < 3:
                examples.extend(float(x) for x in np.unique(v[bad])[:3])
        if n_bad:
            rng = "[%s, %s]" % ("-inf" if lower is None else lower,
                                "+inf" if upper is None else upper)
            yield (iname,
                   "%d of %d %s values are outside the encoded range %s "
                   "(excluding fillValue %s); e.g. %s"
                   % (n_bad, n_total, human, rng, fill,
                      ", ".join("%.2f" % e for e in examples[:3])))


@check({"111": "111_5025"})
def chk_speed_range(ds, ctx):
    """surfaceCurrentSpeed is outside the range encoded in the feature-information
    dataset (Group_F FIDS lower/upper), excluding the fillValue."""
    for f in _fids_range_violations(ds, ctx, "surfaceCurrentSpeed",
                                    "surface current speed"):
        yield f


@check({"111": "111_5026"})
def chk_direction_range(ds, ctx):
    """surfaceCurrentDirection is outside the range encoded in the feature-
    information dataset (Group_F FIDS lower/upper), excluding the fillValue."""
    for f in _fids_range_violations(ds, ctx, "surfaceCurrentDirection",
                                    "surface current direction"):
        yield f


@check({"104": "104_5011"})
def chk_waterlevel_height_range(ds, ctx):
    """waterLevelHeight is outside the range encoded in the feature-information
    dataset (Group_F FIDS lower/upper), excluding the fillValue."""
    for f in _fids_range_violations(ds, ctx, "waterLevelHeight",
                                    "water level height"):
        yield f


@check({"104": "104_5013"})
def chk_uncertainty_range(ds, ctx):
    """uncertainty is outside the range encoded in the feature-information dataset
    (Group_F FIDS lower/upper), excluding the fillValue."""
    for f in _fids_range_violations(ds, ctx, "uncertainty", "uncertainty"):
        yield f


def _values_dim_info(ds, ctx):
    """Yield (iname, gname, dims, nlat, nlon) for each present values dataset."""
    _n, cont = ds.container(ctx["product"])
    if cont is None:
        return
    for iname, inst in ds.instances(cont):
        ia = ds.attrs(inst)
        nlat = _num(ia.get("numPointsLatitudinal"))
        nlon = _num(ia.get("numPointsLongitudinal"))
        for gname, g in ds.value_groups(inst):
            if not ds.has_array(g, "values"):
                continue
            dims = ds.array_dims(g, "values")
            if dims is not None:
                yield iname, gname, dims, nlat, nlon


@check({"104": "104_5006"})
def chk_values_rank_2d(ds, ctx):
    """An S-104 values dataset (DCF2) is not a 2-dimensional array."""
    for iname, gname, dims, _nlat, _nlon in _values_dim_info(ds, ctx):
        if len(dims) != 2:
            yield ("%s/%s/values" % (iname, gname),
                   "values is %d-dimensional (DCF2 requires 2-D)" % len(dims))


@check({"104": "104_5007"})
def chk_values_dims_match(ds, ctx):
    """An S-104 values dataset's dimensions are not numPointsLatitudinal x
    numPointsLongitudinal."""
    for iname, gname, dims, nlat, nlon in _values_dim_info(ds, ctx):
        if (len(dims) == 2 and nlat is not None and nlon is not None
                and (dims[0] != int(nlat) or dims[1] != int(nlon))):
            yield ("%s/%s/values" % (iname, gname),
                   "values dimensions %s != (numPointsLatitudinal %d, "
                   "numPointsLongitudinal %d)" % (dims, int(nlat), int(nlon)))


@check({"102": "102_5001"})
def chk_values_2d(ds, ctx):
    """An S-102 values dataset is not a 2-dimensional array of dimensions
    numPointsLatitudinal x numPointsLongitudinal."""
    _n, cont = ds.container(ctx["product"])
    if cont is None:
        return
    for iname, inst in ds.instances(cont):
        ia = ds.attrs(inst)
        nlat = _num(ia.get("numPointsLatitudinal"))
        nlon = _num(ia.get("numPointsLongitudinal"))
        for gname, g in ds.value_groups(inst):
            if not ds.has_array(g, "values"):
                continue
            dims = None
            try:
                dims = [d.GetSize()
                        for d in g.OpenMDArray("values").GetDimensions()]
            except Exception:                            # noqa: BLE001
                continue
            if len(dims) != 2:
                yield ("%s/%s/values" % (iname, gname),
                       "values is %d-dimensional (must be 2-D "
                       "numPointsLatitudinal x numPointsLongitudinal)" % len(dims))
            elif (nlat is not None and nlon is not None
                  and (dims[0] != int(nlat) or dims[1] != int(nlon))):
                yield ("%s/%s/values" % (iname, gname),
                       "values dimensions %s != (numPointsLatitudinal %d, "
                       "numPointsLongitudinal %d)" % (dims, int(nlat), int(nlon)))


# -- runner -------------------------------------------------------------------
class S100Finding:
    """One S-158 violation. Mirrors the fields the report needs."""
    __slots__ = ("check_id", "severity", "message", "location", "detail",
                 "solution", "product", "cls")

    def __init__(self, rec, product, location, detail):
        self.check_id = rec.get("id", "")
        self.cls = rec.get("cls", "")
        self.severity = SEVERITY.get(self.cls, "W")
        self.message = rec.get("message", "")
        self.solution = rec.get("solution", "")
        self.product = product
        self.location = location
        self.detail = detail


class S100Report:
    """Result of validating one HDF5 dataset: findings + which catalogue checks
    were executed (so a check can be shown FAIL / PASS / NOT-IMPLEMENTED)."""
    def __init__(self, product, path):
        self.product = product
        self.path = path
        self.findings = []
        self.ran_ids = set()
        self.error = None

    @property
    def catalogue(self):
        return catalogue(self.product)

    def status_rows(self):
        """[(record, status)] over the whole catalogue: FAIL / PASS / N/I."""
        failed = {f.check_id for f in self.findings}
        rows = []
        for rec in self.catalogue:
            cid = rec.get("id")
            if cid in failed:
                st = "FAIL"
            elif cid in self.ran_ids:
                st = "PASS"
            else:
                st = "NOT-IMPLEMENTED"
            rows.append((rec, st))
        return rows

    def counts(self):
        out = {}
        for f in self.findings:
            out[f.severity] = out.get(f.severity, 0) + 1
        return out


def run(path, product=None, only=None):
    """Validate one S-100 HDF5 dataset against its S-158 catalogue. Returns an
    :class:`S100Report`. ``product`` is auto-detected when omitted. ``only`` is an
    optional set of catalogue check ids to restrict to (None = every implemented
    check)."""
    product = product or detect_product(path)
    report = S100Report(product, path)
    if product is None or PRODUCTS.get(product, {}).get("kind") != "hdf5":
        report.error = "not an S-100 HDF5 coverage product"
        return report
    idx = catalogue_index(product)
    try:
        ds = H5Dataset(path).open()
    except Exception as exc:                             # noqa: BLE001
        report.error = "cannot open HDF5: %s" % exc
        return report
    ctx = {"product": product}
    for cid, fn in CHECKS.get(product, []):
        rec = idx.get(cid)
        if rec is None:                                  # cite mismatch — skip
            continue
        if only is not None and cid not in only:         # user deselected it
            continue
        report.ran_ids.add(cid)
        try:
            for location, detail in fn(ds, ctx):
                report.findings.append(S100Finding(rec, product, location, detail))
        except Exception:                                # noqa: BLE001
            # A faulty predicate must never abort the whole run.
            continue
    # Stable order: by severity (C,E,W) then check id.
    order = {"C": 0, "E": 1, "W": 2}
    report.findings.sort(key=lambda f: (order.get(f.severity, 3), f.check_id))
    return report


def coverage(product):
    """(implemented, total) executable-check coverage for a product."""
    total = len(catalogue(product))
    impl = len({cid for cid, _ in CHECKS.get(product, [])})
    return impl, total
