"""A self-managing in-memory QGIS point layer of AIS vessels, keyed by MMSI.

Each vessel is one feature. Position reports move the feature; static messages
(Type 5 for Class A, Type 24 A/B for Class B) bind voyage/identity data — name,
callsign, IMO, ship type, dimensions, draught, destination, ETA — to the same
MMSI. Static data is cached per MMSI so it survives across messages: Type 24
arrives in two halves (Part A = name, Part B = callsign/type/size) and they
accumulate, and static that arrives before a position is applied as soon as the
vessel's first position creates its feature.

All methods run on the GUI thread. Styling is best-effort so a cosmetic API
difference between QGIS versions can never stop data plotting.
"""
import os
import time
from datetime import datetime

from qgis.core import (
    QgsVectorLayer, QgsFeature, QgsGeometry, QgsPointXY, QgsProject,
)


def _knob(name, default):
    """One performance-profile knob with a fallback (perf_profile is best-effort)."""
    try:
        from . import perf_profile
        v = perf_profile.get(name)
        return v if v is not None else default
    except Exception:                              # noqa: BLE001
        return default


def set_cpa_highlight_active(active):
    """Enable/disable the route-CPA highlight rule on the live AIS layer.

    The route-map window calls this when a route-CPA becomes active or is cleared.
    While off, the rule is inactive so the renderer skips its per-feature filter
    entirely — no array build / string compare per vessel per frame. Located by
    layer name (VesselLayer.LAYER_NAME); best-effort, never raises."""
    try:
        from qgis.core import QgsProject, QgsRuleBasedRenderer
        layers = QgsProject.instance().mapLayersByName(VesselLayer.LAYER_NAME)
        if not layers:
            return
        layer = layers[0]
        r = layer.renderer()
        if not isinstance(r, QgsRuleBasedRenderer):
            return
        want = bool(active)
        changed = False
        for child in r.rootRule().children():
            if child.label() == "Route-CPA vessel" and child.active() != want:
                child.setActive(want)
                changed = True
        if changed:
            layer.triggerRepaint()
    except Exception:                              # noqa: BLE001  # nosec B110
        pass


# Live-data model. Each tuple is (internal record key, layer field name, type).
# _idx is keyed by the record key so the write path is unchanged; the attribute
# table the user sees / exports uses the formal field names. Position is held in
# the feature geometry, so there are no lon/lat columns.
_FIELDS = [
    ("mmsi",           "vessel_stationID",                  "str"),
    ("name",           "vessel_name",                       "str"),
    ("callsign",       "vessel_callSign",                   "str"),
    ("imo",            "vessel_Id",                         "int"),
    ("ship_type",      "vessel_typeCode",                   "int"),
    ("ship_type_desc", "vessel_typeDescription",            "str"),
    ("length",         "vessel_length",                     "double"),
    ("beam",           "vessel_beam",                       "double"),
    ("draught",        "vessel_draught",                    "double"),
    ("sog",            "vessel_speedOverGround",            "double"),
    ("cog",            "vessel_courseOverGround",           "double"),
    ("heading",        "vessel_heading",                    "int"),
    ("rot",            "vessel_rateOfTurn",                 "double"),
    ("nav_status",     "vessel_navigationStatusCode",       "int"),
    ("nav_status_desc","vessel_navigationStatusDescription", "str"),
    ("destination",    "voyage_destination",                "str"),
    ("eta",            "voyage_ETA",                        "str"),
    ("msg_type",       "transponder_messageType",           "int"),
    ("last_seen",      "transponder_timestamp",             "str"),
    ("fairway_name",   "fairway_name",                      "str"),
    ("fairway_km",     "fairway_routeKm",                   "double"),
    ("fairway_on",     "fairway_onFairway",                 "str"),
    ("fairway_dir",    "fairway_direction",                 "str"),
    ("fairway_flow",   "fairway_flow",                      "str"),
    ("fairway_side",   "fairway_side",                      "str"),
]

# Keys carried from a decoded record into the per-MMSI static cache.
_STATIC_KEYS = ("name", "callsign", "imo", "ship_type", "ship_type_desc",
                "length", "beam", "draught", "destination", "eta")
# Attributes refreshed on every position report.
_DYNAMIC_KEYS = ("cog", "sog", "heading", "rot", "nav_status",
                 "nav_status_desc", "msg_type",
                 "fairway_name", "fairway_km", "fairway_on", "fairway_dir",
                 "fairway_flow", "fairway_side")

# Bundled style (authored in QGIS) applied on layer creation if present.
_QML_RELPATH = os.path.join("styles", "ais_vessels.qml")

# Default map tip used until a QML with its own map tip is bundled.
_DEFAULT_MAPTIP = (
    '<table style="font-family:sans-serif;font-size:11px;">'
    '<tr><td colspan="2"><b>[% coalesce("vessel_name", \'Unknown vessel\') %]</b></td></tr>'
    '<tr><td>MMSI</td><td>[% "vessel_stationID" %]</td></tr>'
    '<tr><td>Callsign</td><td>[% coalesce("vessel_callSign", \'—\') %]</td></tr>'
    '<tr><td>IMO</td><td>[% coalesce("vessel_Id", \'—\') %]</td></tr>'
    '<tr><td>Type</td><td>[% coalesce("vessel_typeDescription", \'—\') %]</td></tr>'
    '<tr><td>Size</td><td>[% coalesce("vessel_length", \'?\') %] × '
    '[% coalesce("vessel_beam", \'?\') %] m</td></tr>'
    '<tr><td>Draught</td><td>[% coalesce("vessel_draught", \'?\') %] m</td></tr>'
    '<tr><td>Destination</td><td>[% coalesce("voyage_destination", \'—\') %]</td></tr>'
    '<tr><td>ETA</td><td>[% coalesce("voyage_ETA", \'—\') %]</td></tr>'
    '<tr><td>SOG / COG</td><td>[% coalesce("vessel_speedOverGround", \'?\') %] kn / '
    '[% coalesce("vessel_courseOverGround", \'?\') %]°</td></tr>'
    '<tr><td>Status</td><td>[% coalesce("vessel_navigationStatusDescription", \'—\') %]</td></tr>'
    '<tr><td>ROT</td><td>[% coalesce("vessel_rateOfTurn", \'—\') %] °/min</td></tr>'
    '<tr><td>Fairway</td><td>[% coalesce("fairway_name", \'—\') %]'
    ' [% coalesce("fairway_direction", \'\') %]</td></tr>'
    '<tr><td>Route km</td><td>[% coalesce("fairway_routeKm", \'—\') %]</td></tr>'
    '<tr><td>Flow</td><td>[% coalesce("fairway_flow", \'—\') %]'
    ' [% if(coalesce("fairway_side",\'\')!=\'\', \'(\' || "fairway_side" || \')\', \'\') %]</td></tr>'
    '<tr><td>Last seen</td><td>[% "transponder_timestamp" %]</td></tr>'
    '</table>'
)


# Detailed live label: name (max 10 chars) on line 1; heading° - SOG kn on line 2
# EXCEPT for vessels with a static nav-status (1 at anchor / 5 moored), where the
# speed/heading line is dropped (it is 0/meaningless). e.g.  MSC Atlant
#                                                            351° - 20 kn
def _label_expr():
    """Detailed-label expression, rebuilt per unit system so the speed reads in the
    active unit (kn / km/h / mph)."""
    from . import units
    f, suf = units.speed_factor_suffix()
    return (
        'left(coalesce("vessel_name", "vessel_stationID"), 10)'
        ' || if("vessel_navigationStatusCode" in (1, 5), \'\','
        ' char(10)'
        ' || coalesce(format_number(coalesce("vessel_heading", "vessel_courseOverGround"), 0), \'---\')'
        ' || \'° - \''
        ' || coalesce(format_number(to_real("vessel_speedOverGround") * %(f)s, 0), \'0\')'
        ' || \' %(suf)s\')'
    ) % {"f": f, "suf": suf}

# Fixed label offset (mm) at the ship's heading + 135° (falls back to COG, then 0)
# — the STARBOARD QUARTER, halfway between the starboard beam (+90°) and full aft
# (+180°). A compass bearing maps to screen offset (east, -north) = (sin b, -cos b):
# +x is right, but QGIS's label Y-offset is positive-DOWN, so the north term is
# negated (verified on-screen). sin/cos take radians, so radians() converts first
# (no unit bug). A leader line (callout) joins the symbol to the label.
_LABEL_DIST_MM = 11.0
_OFFSET_EXPR = (
    "with_variable('b',"
    " radians(coalesce(\"vessel_heading\", \"vessel_courseOverGround\", 0) + 135),"
    " to_string(%(d)s * sin(@b)) || ',' || to_string(-1 * %(d)s * cos(@b)))"
    % {"d": _LABEL_DIST_MM}
)


def _qfield(name, kind):
    """Build a QgsField that works on both Qt5 (QGIS 3) and Qt6 (QGIS 4),
    without the QGIS 3.38+ deprecation warning. kind: 'str'/'int'/'double'."""
    from .qgis_compat import make_field
    return make_field(name, kind)


class VesselLayer:
    LAYER_NAME = "liveFeedAIS"
    REPAINT_INTERVAL = 0.7  # seconds; throttle canvas refreshes

    def __init__(self, registry=None):
        """Start with no layer; an optional VesselRegistry supplies last-known
        identity/dimensions for vessels seen in earlier sessions."""
        self.layer = None
        self._layer_id = None           # project layer id of our layer, if any
        self.registry = registry       # persistent MMSI store (optional)
        self._fid_by_mmsi = {}
        self._static_by_mmsi = {}   # mmsi -> accumulated static field dict
        self._idx = {}
        self._last_repaint = 0.0
        self._label_detailed = True     # detailed 135° ship labels on by default
        self._last_ts = {}             # mmsi -> epoch of last position (staleness)
        self._last_lonlat = {}         # mmsi -> (lon, lat) of last position
        # High-rate feeds coalesce writes: buffer the latest fix per MMSI and flush
        # them in one batched provider edit on the repaint tick (invisible, since
        # the map only repaints every REPAINT_INTERVAL anyway). Off = immediate.
        self._batch = False
        self._pending = {}             # mmsi -> (rec, ts) awaiting a batched flush

    # ---- layer lifecycle -------------------------------------------------
    def _alive(self):
        """True only if our layer still exists in the project.

        Checked via the stored id string, never by calling .id() on self.layer:
        once the user removes the layer, QGIS deletes the underlying C++ object
        and touching the Python wrapper would raise.
        """
        if not self._layer_id:
            return False
        return QgsProject.instance().mapLayer(self._layer_id) is not None

    def ensure_layer(self):
        """Return the live layer, recreating it if it was removed.

        Called both on every message and on Connect, so deleting the layer from
        the panel can never strand the feed: the next Connect (or message)
        rebuilds it, re-applies the style, and refills from the live stream.
        """
        proj = QgsProject.instance()
        if self._layer_id:
            existing = proj.mapLayer(self._layer_id)
            if existing is not None:
                self.layer = existing       # refresh the handle, keep plotting
                return existing
        layer = QgsVectorLayer("Point?crs=EPSG:4326", self.LAYER_NAME, "memory")
        pr = layer.dataProvider()
        pr.addAttributes([_qfield(fname, kind) for _rk, fname, kind in _FIELDS])
        layer.updateFields()
        self._idx = {rk: layer.fields().indexFromName(fname)
                     for rk, fname, _k in _FIELDS}
        self._style(layer)
        proj.addMapLayer(layer)
        self.layer = layer
        self._layer_id = layer.id()
        self._fid_by_mmsi = {}       # feature ids reset with a fresh layer
        return layer

    def _style(self, layer):
        """Apply the default symbol/labels/map-tip, then let a bundled QML (if
        present) override them, then overlay the route-CPA highlight rule."""
        # Apply sensible programmatic defaults first (symbol, labels, map tip)…
        self._apply_default_style(layer)
        # …then let a bundled QML, authored in QGIS, override them if present.
        qml = os.path.join(os.path.dirname(__file__), _QML_RELPATH)
        if os.path.exists(qml):
            try:
                res = layer.loadNamedStyle(qml)
                ok = res[1] if isinstance(res, (tuple, list)) and len(res) > 1 else True
                if ok:
                    layer.triggerRepaint()
            except Exception:  # nosec B110
                pass
        # …overlay the route-CPA highlight on top of whatever symbology is now in
        # place (preserving it)…
        self._add_cpa_highlight_rule(layer)
        # …swap the detailed SVG ship symbols for cheap dots when zoomed out…
        self._apply_scale_symbology(layer)
        # …and (re)apply our switchable ship labels last, so they survive a QML
        # load and reflect the current on/off toggle.
        self._set_labeling(layer)

    def _add_cpa_highlight_rule(self, layer):
        """Overlay a blue dot on the two route-CPA vessels.

        Converts the current renderer to rule-based (which preserves the existing
        symbology exactly) and appends ONE rule that draws a blue circle on any
        target whose ``vessel_stationID`` matches the project variables
        ``@routeCPAVesselOne`` / ``@routeCPAVesselTwo`` — set live by the route-map
        window while a route-CPA is active (empty otherwise, so nothing matches).
        Appended last, so the dot sits on top of the normal symbol. Idempotent and
        best-effort: any cosmetic-API difference just leaves the base symbology."""
        try:
            from qgis.core import QgsRuleBasedRenderer, QgsMarkerSymbol
        except Exception:
            return
        try:
            base = layer.renderer()
            if base is None:
                return
            if isinstance(base, QgsRuleBasedRenderer):
                rb = base.clone()
            else:
                rb = QgsRuleBasedRenderer.convertFromRenderer(base)
            if rb is None:
                return
            root = rb.rootRule()
            label = "Route-CPA vessel"
            for child in root.children():           # idempotent: never add twice
                if child.label() == label:
                    return
            blue = QgsMarkerSymbol.createSimple({
                "name": "circle", "color": "0,90,255,255",
                "outline_color": "255,255,255,255", "outline_width": "0.4",
                "size": "2.6"})
            rule = QgsRuleBasedRenderer.Rule(blue)
            rule.setLabel(label)
            # array_contains + coalesce so empty/unset variables give a definite
            # FALSE (not SQL NULL) — i.e. no dots when no CPA is active.
            rule.setFilterExpression(
                "array_contains(array("
                "coalesce(to_string(@routeCPAVesselOne), ''), "
                "coalesce(to_string(@routeCPAVesselTwo), '')), "
                'to_string("vessel_stationID"))')
            # Dormant until a route-CPA is active: an inactive rule is skipped
            # entirely by the renderer, so its filter isn't evaluated per feature
            # per frame (a real saving at high vessel counts). set_cpa_highlight_
            # active() flips it on/off as the route-map window sets/clears the CPA.
            rule.setActive(False)
            root.appendChild(rule)
            layer.setRenderer(rb)
        except Exception:  # nosec B110
            pass

    # Scale threshold (denominator): the authored SVG ship symbols draw only when
    # zoomed in past ~1:25000; further out, every vessel is a cheap dot. SVG is the
    # expensive render path, so this keeps a dense/firehose scene light at wide zoom
    # (where the ship shape isn't legible anyway).
    _SVG_MAX_DENOM = 25000.0

    def _apply_scale_symbology(self, layer):
        """Gate the SVG ship symbols to close zoom and add a simple-dot rule for
        wide zoom. Best-effort; leaves the base symbology on any problem."""
        try:
            from qgis.core import QgsRuleBasedRenderer, QgsMarkerSymbol
        except Exception:
            return
        try:
            r = layer.renderer()
            if isinstance(r, QgsRuleBasedRenderer):
                rb = r.clone()
            else:
                rb = QgsRuleBasedRenderer.convertFromRenderer(r)
            if rb is None:
                return
            root = rb.rootRule()
            dot_label = "Zoomed-out dot"
            svg_denom = _knob("svg_denom", self._SVG_MAX_DENOM)
            has_dot = False
            for child in root.children():
                lab = child.label()
                if lab == dot_label:
                    has_dot = True
                    try:                            # keep the dot's cutover in step
                        child.setMinimumScale(0.0)      # (re-apply may change denom)
                        child.setMaximumScale(svg_denom)
                    except Exception:  # nosec B110
                        pass
                    continue
                if lab == "Route-CPA vessel":
                    continue                        # highlight shows at all scales
                try:                                # SVG etc.: only when zoomed in
                    child.setMinimumScale(svg_denom)    # visible when denom <= svg
                    child.setMaximumScale(0.0)
                except Exception:  # nosec B110
                    pass
            if not has_dot:
                dot = QgsMarkerSymbol.createSimple({
                    "name": "circle", "color": "44,127,184,255",
                    "outline_color": "255,255,255,200", "outline_width": "0.2",
                    "size": "1.8"})
                rule = QgsRuleBasedRenderer.Rule(dot)
                rule.setLabel(dot_label)
                rule.setMinimumScale(0.0)
                rule.setMaximumScale(svg_denom)          # visible when denom >= svg
                root.appendChild(rule)
            layer.setRenderer(rb)
            layer.triggerRepaint()
        except Exception:  # nosec B110
            pass

    def _apply_default_style(self, layer):
        """Programmatic defaults: orange triangle marker, name/MMSI labels and
        the HTML map tip — each step best-effort across QGIS versions."""
        try:
            from qgis.core import QgsMarkerSymbol, QgsSingleSymbolRenderer
            sym = QgsMarkerSymbol.createSimple({
                "name": "triangle", "size": "4",
                "color": "255,90,0,255", "outline_color": "60,20,0,255",
            })
            layer.setRenderer(QgsSingleSymbolRenderer(sym))
        except Exception:  # nosec B110
            pass
        # Labels are configured by _set_labeling (called from _style) so the
        # switchable detailed label is the single source of truth.
        try:
            layer.setMapTipTemplate(_DEFAULT_MAPTIP)
        except Exception:  # nosec B110
            pass

    def _set_labeling(self, layer):
        """Apply the switchable detailed ship label — name (≤10 chars) plus, for
        moving vessels, heading/speed on a second line — held a fixed distance off
        the symbol at the ship's heading + 135° with a leader line back to it. Or
        disable labels entirely when the toggle is off. Best-effort throughout."""
        try:
            from qgis.core import (QgsPalLayerSettings,
                                   QgsVectorLayerSimpleLabeling)
        except Exception:
            return
        if not self._label_detailed:
            try:
                layer.setLabelsEnabled(False)
                layer.triggerRepaint()
            except Exception:  # nosec B110
                pass
            return
        try:
            s = QgsPalLayerSettings()
            s.fieldName = _label_expr()
            s.isExpression = True
            # Labels (and their leader-line callouts — the expensive part) only when
            # zoomed in past the same 1:25000 the SVG ships appear at: at wider zoom
            # you see dots, and hundreds of labels would overlap into mush anyway.
            try:
                s.scaleVisibility = True
                s.maximumScale = 0.0                # no zoomed-in cut-off
                s.minimumScale = _knob("label_denom", self._SVG_MAX_DENOM)
            except Exception:  # nosec B110
                pass
            # fixed-distance offset at (heading + 135°), data-defined per feature
            try:
                from qgis.core import QgsUnitTypes
                s.placement = QgsPalLayerSettings.OverPoint
                try:
                    s.quadOffset = QgsPalLayerSettings.QuadrantOver
                except Exception:  # nosec B110
                    pass
                s.offsetUnits = QgsUnitTypes.RenderMillimeters
                try:
                    from qgis.core import QgsProperty
                    props = s.dataDefinedProperties()
                    props.setProperty(QgsPalLayerSettings.OffsetXY,
                                      QgsProperty.fromExpression(_OFFSET_EXPR))
                    s.setDataDefinedProperties(props)
                except Exception:
                    s.xOffset = _LABEL_DIST_MM * 0.7    # static SE fallback
                    s.yOffset = -_LABEL_DIST_MM * 0.7
            except Exception:  # nosec B110
                pass
            # leader line (callout) from the vessel symbol to the label
            try:
                from qgis.core import QgsSimpleLineCallout
                callout = QgsSimpleLineCallout()
                try:
                    callout.setEnabled(True)
                except Exception:  # nosec B110
                    pass
                try:
                    callout.setMinimumLength(0.0)
                except Exception:  # nosec B110
                    pass
                s.setCallout(callout)
                try:
                    s.setCalloutEnabled(True)
                except Exception:  # nosec B110
                    pass
            except Exception:  # nosec B110
                pass
            # small text with a white buffer so it reads over the chart
            try:
                from qgis.core import QgsTextFormat, QgsTextBufferSettings
                from . import ais_palette as _p
                fmt = QgsTextFormat()
                fmt.setSize(8)
                fmt.setColor(_p.color("label_text"))    # light on dusk/night
                buf = QgsTextBufferSettings()
                buf.setEnabled(True)
                buf.setSize(0.8)
                buf.setColor(_p.color("label_halo"))    # dark halo on dusk/night
                fmt.setBuffer(buf)
                s.setFormat(fmt)
            except Exception:  # nosec B110
                pass
            layer.setLabeling(QgsVectorLayerSimpleLabeling(s))
            layer.setLabelsEnabled(True)
            layer.triggerRepaint()
        except Exception:  # nosec B110
            pass

    def set_detailed_labels(self, on):
        """Toggle the detailed ship labels at runtime (from the Settings panel)."""
        self._label_detailed = bool(on)
        if self.layer is not None and self._alive():
            self._set_labeling(self.layer)

    def refresh_labels(self):
        """Re-apply labelling so the text/halo colours follow the active day/dusk/
        night palette (called when the chart-light mode changes)."""
        if self.layer is not None and self._alive():
            self._set_labeling(self.layer)

    def refresh_scale_symbology(self):
        """Re-apply the dot/SVG scale thresholds and labels from the current perf
        profile — called after a live profile change so the looser (or tighter)
        dot/label scale takes effect on the existing layer."""
        if self.layer is not None and self._alive():
            self._apply_scale_symbology(self.layer)
            self._set_labeling(self.layer)

    # ---- updates ---------------------------------------------------------
    def update(self, rec):
        """Apply one decoded record: accumulate + persist any static fields, then
        move/create the vessel feature on a position report (or bind static-only
        data to an existing feature). Throttled repaint at the end."""
        mmsi = rec.get("mmsi")
        if not mmsi:
            return
        self.ensure_layer()

        # Accumulate any static fields present in this message.
        static = {k: rec[k] for k in _STATIC_KEYS if rec.get(k) is not None}
        if static:
            self._static_by_mmsi.setdefault(mmsi, {}).update(static)

        # Static identity/dimension data is authoritative — persist for reuse
        # across sightings and sessions (name, length, beam, ship type).
        if static and self.registry is not None:
            self.registry.set(mmsi,
                              name=rec.get("name"),
                              callsign=rec.get("callsign"),
                              imo=rec.get("imo"),
                              length=rec.get("length"),
                              beam=rec.get("beam"),
                              ship_type=rec.get("ship_type"),
                              ship_type_desc=rec.get("ship_type_desc"))

        if rec.get("has_position"):
            if self._batch:
                # buffer the latest fix for this vessel; flushed on the repaint tick
                ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                self._pending[mmsi] = (rec, ts)
                self._last_ts[mmsi] = time.time()          # keep staleness fresh
                self._last_lonlat[mmsi] = (rec["lon"], rec["lat"])
            else:
                self._upsert_position(mmsi, rec)
        elif static and mmsi in self._fid_by_mmsi:
            # Static-only message for a vessel already on the map: bind it now.
            self._change_attrs(self._fid_by_mmsi[mmsi],
                               {self._idx[k]: v for k, v in static.items()})

        self._maybe_repaint()

    def set_batch_mode(self, on):
        """Turn write-coalescing on/off (driven by the message rate). Leaving batch
        mode flushes anything still buffered so nothing is left unplotted."""
        on = bool(on)
        if on == self._batch:
            return
        self._batch = on
        if not on and self._pending:
            self._flush_pending()
            if self.layer is not None:
                self.layer.triggerRepaint()

    def _dyn_changes(self, rec, ts):
        """Attribute-change dict for an existing vessel: last-seen + the dynamic
        fields present in this fix."""
        changes = {self._idx["last_seen"]: ts}
        for key in _DYNAMIC_KEYS:
            if rec.get(key) is not None:
                changes[self._idx[key]] = rec[key]
        return changes

    def _new_feature(self, mmsi, rec, ts):
        """Build a QgsFeature for a first-seen vessel: attrs from this fix + the
        session static cache + a registry fallback (last-known name/dims/type, so a
        previously-seen vessel is labelled/sized before fresh static arrives)."""
        attrs = [None] * len(_FIELDS)
        attrs[self._idx["mmsi"]] = str(mmsi)
        attrs[self._idx["last_seen"]] = ts
        for key in _DYNAMIC_KEYS:
            if rec.get(key) is not None:
                attrs[self._idx[key]] = rec[key]
        for key, val in self._static_by_mmsi.get(mmsi, {}).items():
            attrs[self._idx[key]] = val
        if self.registry is not None:
            cached = self.registry.get_record(mmsi)
            for key in ("name", "callsign", "imo", "length", "beam",
                        "ship_type", "ship_type_desc"):
                if attrs[self._idx[key]] is None and cached.get(key) is not None:
                    attrs[self._idx[key]] = cached[key]
        feat = QgsFeature(self.layer.fields())
        feat.setGeometry(QgsGeometry.fromPointXY(
            QgsPointXY(rec["lon"], rec["lat"])))
        feat.setAttributes(attrs)
        return feat

    def _upsert_position(self, mmsi, rec):
        """Immediate write: move an existing vessel feature to its new position, or
        create it. Used below the batch-rate; above it, _flush_pending batches."""
        pr = self.layer.dataProvider()
        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        self._last_ts[mmsi] = time.time()               # for staleness sweeps
        self._last_lonlat[mmsi] = (rec["lon"], rec["lat"])
        if mmsi in self._fid_by_mmsi:
            fid = self._fid_by_mmsi[mmsi]
            pr.changeGeometryValues({fid: QgsGeometry.fromPointXY(
                QgsPointXY(rec["lon"], rec["lat"]))})
            self._change_attrs(fid, self._dyn_changes(rec, ts))
        else:
            ok, out = pr.addFeatures([self._new_feature(mmsi, rec, ts)])
            if ok and out:
                self._fid_by_mmsi[mmsi] = out[0].id()

    def _flush_pending(self):
        """Apply all buffered fixes in ONE batched provider edit: a single
        changeGeometryValues + changeAttributeValues for existing vessels and a
        single addFeatures for first-seen ones. Turns hundreds of per-message
        provider calls/sec into a handful per repaint (the measured win)."""
        pend = self._pending
        if not pend:
            return
        self._pending = {}
        pr = self.layer.dataProvider()
        geom_changes, attr_changes, new_feats, new_mmsis = {}, {}, [], []
        for mmsi, (rec, ts) in pend.items():
            fid = self._fid_by_mmsi.get(mmsi)
            if fid is not None:
                geom_changes[fid] = QgsGeometry.fromPointXY(
                    QgsPointXY(rec["lon"], rec["lat"]))
                attr_changes[fid] = self._dyn_changes(rec, ts)
            else:
                new_feats.append(self._new_feature(mmsi, rec, ts))
                new_mmsis.append(mmsi)
        try:
            if geom_changes:
                pr.changeGeometryValues(geom_changes)
            if attr_changes:
                pr.changeAttributeValues(attr_changes)
            if new_feats:
                ok, out = pr.addFeatures(new_feats)
                if ok:
                    for m, of in zip(new_mmsis, out):
                        self._fid_by_mmsi[m] = of.id()
        except Exception:  # nosec B110
            pass

    def _change_attrs(self, fid, changes):
        """Write {field_index: value} onto one feature (best-effort)."""
        try:
            self.layer.dataProvider().changeAttributeValues({fid: changes})
        except Exception:  # nosec B110
            pass

    def _maybe_repaint(self, force=False):
        """Flush any buffered writes and repaint, throttled to the profile's repaint
        interval unless forced. In batch mode the flush is where the coalesced fixes
        hit the provider — once per interval, not once per message."""
        now = time.monotonic()
        if force or (now - self._last_repaint) >= _knob("repaint_s",
                                                        self.REPAINT_INTERVAL):
            self._last_repaint = now
            if self._pending:
                self._flush_pending()
            if self.layer is not None:
                self.layer.triggerRepaint()

    def vessel_count(self):
        """Number of distinct vessels currently plotted on the layer."""
        return len(self._fid_by_mmsi)

    # ---- staleness / removal --------------------------------------------
    def stale_mmsis(self, max_age_s, now=None):
        """MMSIs whose last position is older than max_age_s (for stale sweeps)."""
        now = time.time() if now is None else now
        return [m for m, ts in list(self._last_ts.items())
                if now - ts > max_age_s]

    def last_position(self, mmsi):
        """Last known (lon, lat) for an MMSI, or (None, None)."""
        return self._last_lonlat.get(mmsi, (None, None))

    def remove(self, mmsi):
        """Delete a vessel feature and forget its per-MMSI state (a stale/lost
        target). Best-effort so a layer hiccup can't break the feed."""
        fid = self._fid_by_mmsi.pop(mmsi, None)
        self._last_ts.pop(mmsi, None)
        self._last_lonlat.pop(mmsi, None)
        self._static_by_mmsi.pop(mmsi, None)
        if fid is not None and self.layer is not None and self._alive():
            try:
                self.layer.dataProvider().deleteFeatures([fid])
                self._maybe_repaint(force=True)
            except Exception:  # nosec B110
                pass
