#!/usr/bin/env python3
"""EtherGold field configuration — schema, validation, env render, apply.

Canonical UI/API model is field-config.json. Runtime still uses
/etc/ethergold/{role.env,ethergold.env,edge.env}. See FIELD_CONFIG_MATRIX.md.
"""
from __future__ import annotations

import json
import os
import re
import shutil
import sys
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any

from pog422_gpio_hs import HS_OUT_IDLE_LOW_BCMS, park_hs_out_idle_low

# atlas_copper (ETHERGOLD_ROLE=topbox, inject=1) is withdrawn — not reliable.
# Field TopBox is listen-only: POG422 tap, never drive the bus.
PROFILES = (
    "listen_ohd",
    "listen_ohd_server",
    "listen_ohd_mini",
    "wifi_bank_bridge",
    "atlas_wifi_star",
)
LISTEN_PROFILES = ("listen_ohd", "listen_ohd_server", "listen_ohd_mini")
WITHDRAWN_PROFILES = ("atlas_copper",)
HS_PATHS = ("uart", "pi")
FLOOR_GAMES_SOURCES = ("local_p3", "http")
FLOOR_GAMES_HTTP_PORT = 31234
# Hostname or dotted IPv4 (no scheme/path).
FLOOR_GAMES_HOST_RE = re.compile(
    r"^(?:"
    r"(?:\d{1,3}\.){3}\d{1,3}"
    r"|"
    r"[A-Za-z0-9](?:[A-Za-z0-9\-]*[A-Za-z0-9])?"
    r"(?:\.[A-Za-z0-9](?:[A-Za-z0-9\-]*[A-Za-z0-9])?)*"
    r")$"
)
# Rev E J19: BCM10 OIS_OUT, BCM11 M-S_OUT. Idle = driven low (never high-Z).
# HS_OUT_IDLE_LOW_BCMS imported from pog422_gpio_hs.
_HS_PATH_ALIASES = {
    "uart": "uart",
    "uart_hs": "uart",
    "hat": "uart",
    "xr": "uart",
    "pi": "pi",
    "pi_hs": "pi",
    "gpio": "pi",
    "j19": "pi",
}
DEFAULT_AUTH_USER = "SGVI"
DEFAULT_AUTH_PASSWORD = "Southland4111Gaming"
_LEGACY_DEFAULT_PASSWORDS = ("SouthLand4111Gaming",)

TID_RE = re.compile(r"^[0-9A-Fa-f]{4}$")
_TID_PLACEHOLDERS = {"", "-", "--", "---", "—", "–", "N/A", "NONE", "NULL"}


def sanitize_field_tid(raw: Any) -> str:
    """Normalize UI/JSON TID. Placeholders and em-dashes become empty."""
    text = str(raw or "").strip().upper()
    if text in _TID_PLACEHOLDERS or (text and set(text) <= set("-—–")):
        return ""
    return text


DEFAULT_ETC = Path(os.environ.get("ETHERGOLD_ETC", "/etc/ethergold"))
DEFAULT_VAR = Path(os.environ.get("ETHERGOLD_VAR", "/var/lib/ethergold"))


@dataclass
class FieldConfig:
    profile: str = "listen_ohd"
    hs_path: str = "uart"  # uart = HAT XR HS; pi = GPIO J19. Independent of profile.
    asset_id: str = ""
    bank: int = 0
    node_role: str = ""  # side_a|side_b|master_hub|slave_spoke|""
    tid: str = ""  # 4-hex POG422 TID (editable; spoke required; not P3-locked)
    uplink: str = "wlan0"
    game_if: str = "eth1"
    wifi_ssid: str = "POGCOM"
    wifi_psk: str = ""
    ois_wifi_plane: bool = True  # field target: OIS off VPN
    display_dyzineframe: bool | None = None  # None = profile default
    bank_position: int = 0  # operator bank seat 1..x; not FIXTURE_POS
    enrollment_location: str = ""
    enrollment_path: str = ""
    log_upload_enabled: bool = True
    notes: str = ""
    pog_cc_theme: int = 1  # Heltec attract 1–5 (Slots/Numbers/Rainbow/Fire/Quiet)
    pog_cc_volume: int = 5  # Heltec DS1881 0–62
    # listen_ohd_mini only: where DF gets progressive_floor_games.
    floor_games_source: str = "local_p3"  # local_p3 | http
    floor_games_http_host: str = ""  # IP/hostname when floor_games_source=http
    # atlas_wifi_star spoke: optional clear-WLAN hub address (EDGE_HUB_IP).
    hub_uplink_ip: str = ""

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any] | None) -> "FieldConfig":
        data = dict(data or {})
        aliases = {
            "ASSET_ID": "asset_id",
            "asset": "asset_id",
            "FIELD_BANK": "bank",
            "FIELD_PROFILE": "profile",
            "FIELD_TID": "tid",
            "BANK_POSITION": "bank_position",
            "POG422_HS_PATH": "hs_path",
            "hs_mode": "hs_path",
            "EDGE_HUB_IP": "hub_uplink_ip",
            "hub_ip": "hub_uplink_ip",
        }
        for src, dest in aliases.items():
            if dest not in data and src in data:
                data[dest] = data[src]
        known = {f.name for f in cls.__dataclass_fields__.values()}  # type: ignore[attr-defined]
        filtered = {k: v for k, v in data.items() if k in known}
        if "bank" in filtered:
            try:
                filtered["bank"] = int(filtered["bank"])
            except (TypeError, ValueError):
                filtered["bank"] = 0
        if "tid" in filtered and filtered["tid"] is not None:
            filtered["tid"] = sanitize_field_tid(filtered["tid"])
        if "asset_id" in filtered and filtered["asset_id"] is not None:
            filtered["asset_id"] = str(filtered["asset_id"]).strip()
        if "enrollment_location" in filtered and filtered["enrollment_location"] is not None:
            filtered["enrollment_location"] = str(filtered["enrollment_location"]).strip()
        if "enrollment_path" in filtered and filtered["enrollment_path"] is not None:
            text = str(filtered["enrollment_path"]).strip()
            if text and not text.startswith("/"):
                text = "/" + text
            if text and not text.endswith("/"):
                text += "/"
            filtered["enrollment_path"] = text
        if "log_upload_enabled" in filtered:
            raw = filtered["log_upload_enabled"]
            if isinstance(raw, str):
                filtered["log_upload_enabled"] = raw.strip().lower() in (
                    "1",
                    "true",
                    "yes",
                    "on",
                )
            else:
                filtered["log_upload_enabled"] = bool(raw)
        if "bank_position" in filtered:
            try:
                filtered["bank_position"] = int(filtered["bank_position"])
            except (TypeError, ValueError):
                filtered["bank_position"] = 0
        if "pog_cc_theme" in filtered:
            try:
                filtered["pog_cc_theme"] = int(filtered["pog_cc_theme"])
            except (TypeError, ValueError):
                filtered["pog_cc_theme"] = 1
        if "pog_cc_volume" in filtered:
            try:
                filtered["pog_cc_volume"] = int(filtered["pog_cc_volume"])
            except (TypeError, ValueError):
                filtered["pog_cc_volume"] = 5
        if "hs_path" in filtered:
            filtered["hs_path"] = normalize_hs_path(filtered["hs_path"])
        if "floor_games_source" in filtered:
            filtered["floor_games_source"] = normalize_floor_games_source(
                filtered["floor_games_source"]
            )
        if "floor_games_http_host" in filtered and filtered["floor_games_http_host"] is not None:
            filtered["floor_games_http_host"] = sanitize_floor_games_http_host(
                filtered["floor_games_http_host"]
            )
        if "hub_uplink_ip" in filtered and filtered["hub_uplink_ip"] is not None:
            filtered["hub_uplink_ip"] = str(filtered["hub_uplink_ip"]).strip()
        return cls(**filtered)


def normalize_hs_path(raw: Any) -> str:
    text = str(raw or "uart").strip().lower().replace("-", "_").replace(" ", "_")
    return _HS_PATH_ALIASES.get(text, text)


def normalize_floor_games_source(raw: Any) -> str:
    text = str(raw or "local_p3").strip().lower().replace("-", "_").replace(" ", "_")
    aliases = {
        "local_p3": "local_p3",
        "local": "local_p3",
        "p3": "local_p3",
        "http": "http",
        "remote": "http",
        "url": "http",
    }
    return aliases.get(text, text)


def sanitize_floor_games_http_host(raw: Any) -> str:
    """Strip scheme/path/port; keep host only."""
    text = str(raw or "").strip()
    if not text:
        return ""
    lower = text.lower()
    for prefix in ("http://", "https://"):
        if lower.startswith(prefix):
            text = text[len(prefix) :]
            lower = text.lower()
            break
    text = text.split("/")[0].strip()
    if text.startswith("[") and "]" in text:
        # [ipv6]:port → drop for now (Mini uses IPv4/hostname)
        text = text[1 : text.index("]")]
    elif text.count(":") == 1 and not text.startswith("["):
        host, _, port = text.partition(":")
        if port.isdigit():
            text = host
    return text.strip()


def floor_games_http_url(host: str, *, port: int = FLOOR_GAMES_HTTP_PORT) -> str:
    h = sanitize_floor_games_http_host(host)
    return f"http://{h}:{int(port)}/EtherGold/floor-games.json"


def etc_dir(root: Path | None = None) -> Path:
    return Path(root) if root is not None else DEFAULT_ETC


def var_dir(root: Path | None = None) -> Path:
    return Path(root) if root is not None else DEFAULT_VAR


def field_config_path(etc: Path | None = None) -> Path:
    return etc_dir(etc) / "field-config.json"


def auth_path(etc: Path | None = None) -> Path:
    return etc_dir(etc) / "field-ui.auth"


def _parse_env_file(path: Path) -> dict[str, str]:
    parsed: dict[str, str] = {}
    try:
        for line in path.read_text(encoding="utf-8").splitlines():
            raw = line.strip()
            if not raw or raw.startswith("#") or "=" not in raw:
                continue
            key, _, val = raw.partition("=")
            parsed[key.strip()] = val.strip()
    except OSError:
        return {}
    return parsed


def _merge_env_identity(cfg: FieldConfig, etc: Path | None) -> FieldConfig:
    """Fill blank identity from applied ethergold.env so /config is not empty."""
    e = etc_dir(etc)
    env = _parse_env_file(e / "ethergold.env")
    edge = _parse_env_file(e / "edge.env")
    if not env and not edge:
        return cfg
    data = cfg.to_dict()
    if not (data.get("asset_id") or "").strip():
        data["asset_id"] = env.get("ASSET_ID") or ""
    if not data.get("bank"):
        try:
            data["bank"] = int(env.get("FIELD_BANK") or edge.get("EDGE_BANK") or 0)
        except (TypeError, ValueError):
            pass
    if not field_config_path(etc).is_file():
        prof = env.get("FIELD_PROFILE") or ""
        if prof in PROFILES:
            data["profile"] = prof
        if env.get("FIELD_TID"):
            data["tid"] = env["FIELD_TID"]
        hs = env.get("POG422_HS_PATH") or ""
        if hs:
            data["hs_path"] = normalize_hs_path(hs)
    if not data.get("bank_position"):
        try:
            data["bank_position"] = int(env.get("BANK_POSITION") or 0)
        except (TypeError, ValueError):
            pass
    # Recover Node Role from applied edge.env when JSON blank / UI reload.
    if not (data.get("node_role") or "").strip():
        topo = (edge.get("EDGE_TOPOLOGY") or "").strip().lower()
        role = (edge.get("EDGE_ROLE") or "").strip().lower()
        if topo == "hub" or role == "hub":
            data["node_role"] = "master_hub"
        elif topo == "spoke" or role == "spoke":
            data["node_role"] = "slave_spoke"
        elif role == "peer-a":
            data["node_role"] = "side_a"
        elif role == "peer-b":
            data["node_role"] = "side_b"
    if not (data.get("hub_uplink_ip") or "").strip() and edge.get("EDGE_HUB_IP"):
        data["hub_uplink_ip"] = edge["EDGE_HUB_IP"]
    if edge.get("EDGE_UPLINK"):
        data["uplink"] = edge["EDGE_UPLINK"]
    if edge.get("EDGE_GAME_IF"):
        data["game_if"] = edge["EDGE_GAME_IF"]
    if not (data.get("tid") or "").strip():
        tid = edge.get("FIELD_TID") or env.get("FIELD_TID") or ""
        if tid:
            data["tid"] = tid
    return FieldConfig.from_dict(data)


def load_field_config(etc: Path | None = None) -> FieldConfig:
    path = field_config_path(etc)
    cfg = FieldConfig()
    if path.is_file():
        try:
            with path.open(encoding="utf-8") as handle:
                cfg = FieldConfig.from_dict(json.load(handle))
        except (OSError, ValueError, json.JSONDecodeError):
            cfg = FieldConfig()
    return _merge_env_identity(cfg, etc)


def save_field_config(cfg: FieldConfig, etc: Path | None = None) -> Path:
    path = field_config_path(etc)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".json.tmp")
    with tmp.open("w", encoding="utf-8") as handle:
        json.dump(cfg.to_dict(), handle, indent=2, sort_keys=True)
        handle.write("\n")
    tmp.replace(path)
    return path


def load_auth_credentials(etc: Path | None = None) -> tuple[str, str]:
    user = os.environ.get("ETHERGOLD_FIELD_UI_USER", "").strip()
    password = os.environ.get("ETHERGOLD_FIELD_UI_PASSWORD", "").strip()
    path = auth_path(etc)
    if path.is_file():
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
            for line in lines:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                if line.startswith("user="):
                    user = user or line.split("=", 1)[1].strip()
                elif line.startswith("password="):
                    password = password or line.split("=", 1)[1].strip()
                elif ":" in line and not user:
                    u, p = line.split(":", 1)
                    user, password = u.strip(), p.strip()
        except OSError:
            pass
    return user or DEFAULT_AUTH_USER, password or DEFAULT_AUTH_PASSWORD


def ensure_default_auth_file(etc: Path | None = None) -> Path:
    path = auth_path(etc)
    desired = (
        "# EtherGold field UI login (server-side only)\n"
        f"user={DEFAULT_AUTH_USER}\n"
        f"password={DEFAULT_AUTH_PASSWORD}\n"
    )
    if path.is_file():
        try:
            raw = path.read_text(encoding="utf-8")
        except OSError:
            return path
        if any(f"password={old}" in raw for old in _LEGACY_DEFAULT_PASSWORDS):
            path.write_text(desired, encoding="utf-8")
            try:
                os.chmod(path, 0o600)
            except OSError:
                pass
        return path
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(desired, encoding="utf-8")
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass
    return path


def uart_socket_id(_cfg: FieldConfig | None = None) -> int:
    """Internal uartd socket index. Always 1 on a production Pi (one HAT).

    Not a bank seat and not GA FIXTURE_POS. Launchers may still read
    FIXTURE_POS from the environment as a socket override; Apply must not
    write that key.
    """
    return 1


def derive_fixture_pos(cfg: FieldConfig) -> int:
    """Deprecated alias of uart_socket_id. Do not use for cabinet identity."""
    return uart_socket_id(cfg)


def hs_path_value(cfg: FieldConfig) -> str:
    hs = normalize_hs_path(cfg.hs_path)
    return hs if hs in HS_PATHS else "uart"


def apply_hs_path_pins(hs_path: str) -> dict[str, Any]:
    """Always park BCM10/11 output-low. UART vs Pi only selects HS protocol.

    UART: XR GPIO1 still owns handshake sense/drive — Apply does not pulse Pi
    HS. Pi: same park, then listen/bridge use BCM8/9 IN (and SuperBall DE on
    BCM11 when that soak is running).
    Best-effort. Missing pinctrl (Windows tests, non-Pi) is skipped, not a failed Apply.
    """
    path = normalize_hs_path(hs_path)
    parked = park_hs_out_idle_low()
    parked["hs_path"] = path if path in HS_PATHS else "uart"
    parked["protocol"] = "pi" if parked["hs_path"] == "pi" else "uart"
    return parked


def is_listen_profile(cfg: FieldConfig) -> bool:
    return cfg.profile in LISTEN_PROFILES


def profile_wants_dyzineframe(cfg: FieldConfig) -> bool:
    if cfg.display_dyzineframe is not None:
        return bool(cfg.display_dyzineframe)
    # listen_ohd = Atlas TopBox HDMI, DyzineFrame TBX 1920×1080 (machine hooks)
    # plus HTTP :31234 progressives (Win11 OHD can use the same feed).
    # listen_ohd_mini = 1920×360 Mini panel + local listen CSV; no Heltec.
    # listen_ohd_server = same listen + HTTP, no local DF (OHD-only site).
    # wifi_bank_bridge stays headless. atlas_wifi_star keeps cabinet DF.
    return cfg.profile in ("listen_ohd", "listen_ohd_mini", "atlas_wifi_star")


def normalize_machine_role_hint(raw: str | None) -> str:
    """Return ``M`` / ``S`` / ``\"\"`` from GAMESTATS / UI Mode text."""
    text = (raw or "").strip().upper()
    if not text:
        return ""
    if text in ("M", "MASTER") or text.startswith("MASTER"):
        return "M"
    if text in ("S", "SLAVE") or text.startswith("SLAVE"):
        return "S"
    return ""


def resolve_machine_role_for_validate(explicit: str | None = None) -> str:
    """Prefer explicit UI/API hint, else cabinet identity when known."""
    role = normalize_machine_role_hint(explicit)
    if role:
        return role
    try:
        from cabinet_identity import load_cabinet_identity

        cabinet = load_cabinet_identity() or {}
        role = normalize_machine_role_hint(str(cabinet.get("machine_role") or ""))
        if role:
            return role
    except Exception:
        pass
    try:
        from pog_edge_identity import latch_cabinet_identity

        ident = latch_cabinet_identity()
        role = normalize_machine_role_hint(getattr(ident, "machine_role", "") or "")
        if role:
            return role
    except Exception:
        pass
    return ""


def field_config_warnings(cfg: FieldConfig) -> list[str]:
    """Non-blocking Apply notes (reserved for soft cautions)."""
    return []


def validate_field_config(
    cfg: FieldConfig,
    *,
    machine_role: str | None = None,
) -> list[str]:
    errors: list[str] = []
    if cfg.profile in WITHDRAWN_PROFILES:
        errors.append(
            "atlas_copper is withdrawn — use listen_ohd (TopBox), "
            "listen_ohd_mini (1920×360), or listen_ohd_server (headless OHD HTTP), "
            "never drive the bus"
        )
    elif cfg.profile not in PROFILES:
        errors.append(f"unknown profile {cfg.profile!r}")
    if normalize_hs_path(cfg.hs_path) not in HS_PATHS:
        errors.append(f"unknown hs_path {cfg.hs_path!r} (uart or pi)")
    if not (cfg.asset_id or "").strip():
        errors.append("asset_id is required")
    try:
        bank = int(cfg.bank)
    except (TypeError, ValueError):
        errors.append("bank must be an integer")
        bank = -1
    else:
        if bank < 0:
            errors.append("bank must be >= 0")
    try:
        pos = int(cfg.bank_position)
    except (TypeError, ValueError):
        errors.append("bank_position must be an integer")
        pos = 0
    else:
        if pos < 0:
            errors.append("bank_position must be >= 0 (0 = unset)")
    try:
        theme = int(cfg.pog_cc_theme)
    except (TypeError, ValueError):
        errors.append("pog_cc_theme must be an integer 1–5")
    else:
        if theme < 1 or theme > 5:
            errors.append("pog_cc_theme must be 1–5 (Slots/Numbers/Rainbow/Fire/Quiet)")
    try:
        vol = int(cfg.pog_cc_volume)
    except (TypeError, ValueError):
        errors.append("pog_cc_volume must be an integer 0–62")
    else:
        if vol < 0 or vol > 62:
            errors.append("pog_cc_volume must be 0–62 (DS1881)")

    if cfg.profile == "wifi_bank_bridge":
        if cfg.node_role not in ("side_a", "side_b"):
            errors.append("wifi_bank_bridge requires node_role side_a or side_b")
    elif cfg.profile == "atlas_wifi_star":
        if bank < 1:
            errors.append(
                "atlas_wifi_star requires bank number >= 1 "
                "(bank 0 is not valid — bank ID partitions hub enroll/overlay)"
            )
        if cfg.node_role not in ("master_hub", "slave_spoke"):
            errors.append("atlas_wifi_star requires node_role master_hub or slave_spoke")
        if int(cfg.bank_position or 0) < 1:
            errors.append(
                "atlas_wifi_star requires Position Number >= 1 (seat label; Master Hub is Node Role)"
            )
        if cfg.node_role == "slave_spoke":
            if not TID_RE.match(sanitize_field_tid(cfg.tid)):
                errors.append("slave_spoke requires tid as 4 hex digits (POG422 TID)")
            try:
                from wlan_star_apply import spoke_id_from_bank_position

                spoke_id_from_bank_position(int(cfg.bank_position))
            except ValueError as exc:
                errors.append(str(exc))
        # When GAMESTATS / cabinet identity is known, Node Role must match M/S.
        # Master = Hub and Slave = Spoke (locked WiFi-star law).
        cab_role = resolve_machine_role_for_validate(machine_role)
        if cab_role == "S" and cfg.node_role == "master_hub":
            errors.append(
                "MACHINE_ROLE=Slave cannot be Master Hub — Node Role must be Slave Spoke "
                "(Master=Hub per bank)"
            )
        elif cab_role == "M" and cfg.node_role == "slave_spoke":
            errors.append(
                "MACHINE_ROLE=Master cannot be Slave Spoke — Node Role must be Master Hub "
                "(one hub per bank)"
            )
    elif cfg.node_role:
        errors.append(f"profile {cfg.profile} does not use node_role")

    # listen_ohd_mini: operator chooses Local P3 vs HTTP for active floor games.
    # Progressives meters stay local CSV from this listen (empty progressives_source).
    if cfg.profile == "listen_ohd_mini":
        src = normalize_floor_games_source(cfg.floor_games_source)
        if src not in FLOOR_GAMES_SOURCES:
            errors.append(
                f"floor_games_source must be local_p3 or http (got {cfg.floor_games_source!r})"
            )
        elif src == "http":
            host = sanitize_floor_games_http_host(cfg.floor_games_http_host)
            if not host:
                errors.append("floor_games_http_host is required when floor_games_source is http")
            elif not FLOOR_GAMES_HOST_RE.match(host):
                errors.append(
                    f"floor_games_http_host must be an IPv4 or hostname (got {host!r})"
                )
            else:
                # Reject obviously bad dotted quads (e.g. 999.1.1.1).
                if re.fullmatch(r"(?:\d{1,3}\.){3}\d{1,3}", host):
                    octets = [int(p) for p in host.split(".")]
                    if any(o > 255 for o in octets):
                        errors.append(f"floor_games_http_host invalid IPv4 {host!r}")

    # Listen uses the same 4-hex GAMESTATS TID as POG_TID / listen filter.
    # Optional until Grab GAMESTATS succeeds. Leftover junk is ignored, not
    # an error — do not block listen Apply. Hub / wifi_bank_bridge leftover
    # is also ignored (spoke tunnel key only).

    path = (cfg.enrollment_path or "").strip()
    loc = (cfg.enrollment_location or "").strip()
    if path.lower().strip("/") in ("__new__", "new"):
        errors.append("enrollment_path looks like the new-site placeholder — type a real Path")
    if path and not loc:
        errors.append("enrollment_location is required when enrollment_path is set")
    if loc and not path:
        errors.append("enrollment_path is required when enrollment_location is set")

    if cfg.profile in ("wifi_bank_bridge", "atlas_wifi_star"):
        if not (cfg.uplink or "").strip():
            errors.append("uplink interface is required for WiFi profiles")

    return errors


def _env_lines(mapping: dict[str, str]) -> str:
    return "".join(f"{k}={v}\n" for k, v in mapping.items())


def render_role_env(cfg: FieldConfig) -> dict[str, str]:
    role = "listen" if is_listen_profile(cfg) else "edge"
    return {"ETHERGOLD_ROLE": role}


def render_ethergold_env(cfg: FieldConfig) -> dict[str, str]:
    inject = "0" if is_listen_profile(cfg) else "1"
    out = {
        "CHANNEL": "both",
        "POG422_UARTD": "1",
        "POG422_UARTD_INJECT": inject,
        "ASSET_ID": cfg.asset_id.strip(),
        "FIELD_BANK": str(int(cfg.bank)),
        "FIELD_PROFILE": cfg.profile,
        "UART_SOCKET_ID": str(uart_socket_id(cfg)),
        "POG422_HS_PATH": hs_path_value(cfg),
    }
    if int(cfg.bank_position) > 0:
        out["BANK_POSITION"] = str(int(cfg.bank_position))
    if is_listen_profile(cfg):
        out["ETHERGOLD_CURRENT_GAME_IPC"] = "write"
    tid = sanitize_field_tid(cfg.tid)
    if TID_RE.match(tid) and (
        is_listen_profile(cfg)
        or cfg.profile == "atlas_wifi_star"
    ):
        out["FIELD_TID"] = tid
    if cfg.ois_wifi_plane and cfg.profile in ("wifi_bank_bridge", "atlas_wifi_star"):
        out["OIS_WIFI_PLANE"] = "1"
    return out


def render_edge_env(cfg: FieldConfig) -> dict[str, str]:
    if is_listen_profile(cfg):
        return {"EDGE_MODE": "0"}

    out: dict[str, str] = {
        "EDGE_MODE": "1",
        "EDGE_UPLINK": cfg.uplink or "wlan0",
        "EDGE_GAME_IF": cfg.game_if or "eth1",
        "EDGE_WIFI_SSID": cfg.wifi_ssid or "POGCOM",
        "EDGE_DHCP_ONCE": "1",
    }
    if cfg.wifi_psk:
        out["EDGE_WIFI_PSK"] = cfg.wifi_psk

    if cfg.profile == "wifi_bank_bridge":
        out["EDGE_ROLE"] = "peer-a" if cfg.node_role == "side_a" else "peer-b"
        # single-hop: no EDGE_TOPOLOGY
    else:
        # atlas_wifi_star — gretap-only AF_PACKET (UART owns local T340).
        out["EDGE_GAME_IF"] = "none"
        out["EDGE_BANK"] = str(int(cfg.bank))
        out["EDGE_BANK_WLAN"] = "1"
        out["EDGE_TID_UNICAST"] = "1"
        out["EDGE_OPEN_ENROLL"] = "1"
        if cfg.node_role == "master_hub":
            out["EDGE_TOPOLOGY"] = "hub"
            out["EDGE_ROLE"] = "hub"
            hub_tid = sanitize_field_tid(cfg.tid)
            if TID_RE.match(hub_tid):
                out["FIELD_TID"] = hub_tid
        else:
            out["EDGE_TOPOLOGY"] = "spoke"
            out["EDGE_ROLE"] = "spoke"
            spoke_tid = sanitize_field_tid(cfg.tid)
            if TID_RE.match(spoke_tid):
                out["FIELD_TID"] = spoke_tid
            from wlan_star_apply import spoke_id_from_bank_position

            out["EDGE_SPOKE_ID"] = str(spoke_id_from_bank_position(int(cfg.bank_position)))
            hub_ip = (cfg.hub_uplink_ip or "").strip()
            if hub_ip:
                out["EDGE_HUB_IP"] = hub_ip
    if cfg.ois_wifi_plane:
        out["OIS_WIFI_PLANE"] = "1"
    return out


def render_all(cfg: FieldConfig) -> dict[str, dict[str, str]]:
    return {
        "role.env": render_role_env(cfg),
        "ethergold.env": render_ethergold_env(cfg),
        "edge.env": render_edge_env(cfg),
    }


def mask_secrets(cfg: FieldConfig) -> dict[str, Any]:
    data = cfg.to_dict()
    if data.get("wifi_psk"):
        data["wifi_psk"] = "***"
    return data


def preview_apply(
    cfg: FieldConfig,
    *,
    machine_role: str | None = None,
) -> dict[str, Any]:
    errors = validate_field_config(cfg, machine_role=machine_role)
    warnings = field_config_warnings(cfg)
    rendered = render_all(cfg) if not errors else {}
    return {
        "ok": not errors,
        "errors": errors,
        "warnings": warnings,
        "uart_socket_id": uart_socket_id(cfg) if not errors else None,
        "dyzineframe": profile_wants_dyzineframe(cfg) if not errors else None,
        "generated": rendered,
        "services_will_change": _service_notes(cfg) if not errors else [],
    }


def _service_notes(cfg: FieldConfig) -> list[str]:
    notes = [
        "CHANNEL=both (machine + OIS)",
        f"ETHERGOLD_ROLE={render_role_env(cfg)['ETHERGOLD_ROLE']}",
        "Apply writes env files only — no reboot, no listen/collector restart",
    ]
    if cfg.profile in ("wifi_bank_bridge", "atlas_wifi_star"):
        notes.append("WireGuard/gretap tunnel bring-up (machine on VPN)")
        if cfg.ois_wifi_plane:
            notes.append("OIS target: 6GHz WiFi + HMAC (not inside VPN)")
    if cfg.profile == "listen_ohd_server":
        notes.append("DyzineFrame off — HTTP :31234 progressives for Win11 OHD 3840×1080, no local display")
    elif cfg.profile == "listen_ohd_mini":
        notes.append("DyzineFrame autostart enabled (MINI 1920×360; no Heltec/door)")
        notes.append("Progressives meters: local ~/EtherGold/progressives.csv from this listen")
        fg = normalize_floor_games_source(cfg.floor_games_source)
        if fg == "http":
            host = sanitize_floor_games_http_host(cfg.floor_games_http_host)
            notes.append(
                f"Active games: HTTP from {host}:{FLOOR_GAMES_HTTP_PORT}/EtherGold/floor-games.json"
            )
        else:
            notes.append("Active games: Local P3 → ~/EtherGold/floor-games.json")
        notes.append("HTTP :31234 still serves progressives for optional remote OHD pull")
    elif profile_wants_dyzineframe(cfg):
        notes.append("DyzineFrame autostart enabled (TBX 1920×1080 TopBox)")
        notes.append("HTTP :31234 progressives on (local DF and Win11 OHD can both consume)")
    else:
        notes.append("DyzineFrame autostart disabled")
    hs = normalize_hs_path(cfg.hs_path)
    if hs == "pi":
        notes.append(
            "Pi HS (GPIO J19): J3/J4 on PI. Apply parks BCM10/11 output-low. "
            "Listen senses HS on BCM9 (M-S_IN) / BCM8 (OIS_IN) and keeps XR UART RX. "
            "Workshop — not field-published."
        )
    else:
        notes.append(
            "UART HS (HAT / XR): J3/J4 on UART for protocol. "
            "Apply still parks BCM10/11 output-low (safe if jumpers are actually on PI)."
        )
    return notes


def backup_etc(etc: Path | None = None, var: Path | None = None) -> Path:
    e = etc_dir(etc)
    v = var_dir(var)
    stamp = time.strftime("%Y%m%d-%H%M%S")
    dest = v / "config-backup" / stamp
    dest.mkdir(parents=True, exist_ok=True)
    for name in ("field-config.json", "role.env", "ethergold.env", "edge.env"):
        src = e / name
        if src.is_file():
            shutil.copy2(src, dest / name)
    (dest / "README.txt").write_text(
        f"EtherGold field config backup {stamp}\n", encoding="utf-8"
    )
    latest = v / "config-backup" / "latest"
    if latest.exists() or latest.is_symlink():
        try:
            latest.unlink()
        except OSError:
            pass
    try:
        latest.symlink_to(dest)
    except OSError:
        (v / "config-backup" / "latest.path").write_text(str(dest), encoding="utf-8")
    return dest


def write_env_file(path: Path, mapping: dict[str, str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    with tmp.open("w", encoding="utf-8", newline="\n") as handle:
        handle.write("# Generated by EtherGold field_config — do not hand-edit\n")
        for key, val in mapping.items():
            handle.write(f"{key}={val}\n")
    tmp.replace(path)


def apply_field_config(
    cfg: FieldConfig,
    *,
    etc: Path | None = None,
    var: Path | None = None,
    invoke_role_script: bool = False,
    dry_run: bool = False,
    machine_role: str | None = None,
) -> dict[str, Any]:
    preview = preview_apply(cfg, machine_role=machine_role)
    if not preview["ok"]:
        return preview
    if dry_run:
        preview["applied"] = False
        preview["dry_run"] = True
        return preview

    e = etc_dir(etc)
    v = var_dir(var)
    backup = backup_etc(e, v)
    save_field_config(cfg, e)
    rendered = preview["generated"]
    write_env_file(e / "role.env", rendered["role.env"])
    write_env_file(e / "ethergold.env", rendered["ethergold.env"])
    write_env_file(e / "edge.env", rendered["edge.env"])
    ensure_default_auth_file(e)
    preview["hs_pins"] = apply_hs_path_pins(hs_path_value(cfg))
    preview["hostname"] = _maybe_stamp_hostname(cfg)
    preview["cabinet_identity"] = _persist_apply_identity(cfg)

    if cfg.profile == "atlas_wifi_star":
        try:
            from wlan_star_apply import apply_wlan_star_local

            preview["wlan_star"] = apply_wlan_star_local(cfg, etc=e)
        except Exception as exc:
            preview["wlan_star"] = {"ok": False, "error": str(exc)}

    # Field Apply never reboots and never restarts listen/collector.
    # set_ethergold_role.sh / install_collector_service.sh stay off this path.
    role_rc: int | None = None
    if invoke_role_script:
        role_rc = _invoke_set_role(cfg, e)

    preview["applied"] = True
    preview["backup"] = str(backup)
    preview["role_script_rc"] = role_rc
    # Onsite enroll is Identity, not a DyzineFrame-only side effect. POG-CC USB
    # /asset later re-upserts the same row when config.ini already has a Path.
    preview["enrollment_local"] = _ensure_enrollment_overlay(cfg)
    if profile_wants_dyzineframe(cfg):
        preview["dyzineframe"] = _stamp_dyzineframe_identity(cfg)
    else:
        preview["dyzineframe"] = {
            "ok": False,
            "skipped": "profile has no local DyzineFrame",
        }
    return preview


def _replace_ini_key(text: str, key: str, value: str) -> tuple[str, bool]:
    pattern = re.compile(rf"^(\s*{re.escape(key)}\s*=\s*).*$", re.IGNORECASE | re.MULTILINE)
    if not pattern.search(text):
        return text, False
    return pattern.sub(rf"\g<1>{value}", text, count=1), True


def _set_or_append_ini_key(text: str, key: str, value: str) -> tuple[str, bool]:
    text, hit = _replace_ini_key(text, key, value)
    if hit:
        return text, True
    if re.search(r"^\[TOPBOX\]", text, re.MULTILINE | re.IGNORECASE):
        text = re.sub(
            r"(^\[TOPBOX\][^\n]*\n)",
            rf"\1{key} = {value}\n",
            text,
            count=1,
            flags=re.MULTILINE | re.IGNORECASE,
        )
        return text, True
    return text, False


def _import_enrollment():
    candidates = []
    env = (os.environ.get("DYZINEFRAME_ROOT") or "").strip()
    if env:
        candidates.append(Path(env))
    here = Path(__file__).resolve()
    # EtherGold/dev → Cursor OHD Project/DyzineFrame (workshop tests)
    candidates.append(here.parents[2] / "DyzineFrame")
    home = os.environ.get("HOME") or os.environ.get("USERPROFILE") or ""
    if home:
        candidates.append(Path(home) / "DyzineFrame")
    candidates.append(Path("/home/topbox/DyzineFrame"))
    for root in candidates:
        if (root / "enrollment.py").is_file():
            s = str(root)
            if s not in sys.path:
                sys.path.insert(0, s)
            import enrollment  # noqa: WPS433

            return enrollment
    return None


def enrollment_locations_payload(csv_path: str = "") -> dict[str, Any]:
    mod = _import_enrollment()
    if mod is None:
        return {"ok": False, "error": "enrollment.py not found", "locations": [], "assets": {}}
    try:
        locations = mod.list_locations(csv_path)
        assets = mod.assets_by_number(csv_path)
    except OSError as exc:
        return {"ok": False, "error": str(exc), "locations": [], "assets": {}}
    return {"ok": True, "locations": locations, "assets": assets}


def _ensure_enrollment_overlay(cfg: FieldConfig) -> dict[str, Any]:
    """Identity Apply: upsert this asset in Enrollment.csv (onsite enroll or machine move)."""
    asset = (cfg.asset_id or "").strip()
    path = (cfg.enrollment_path or "").strip()
    location = (cfg.enrollment_location or "").strip()
    if not asset.isdigit() or not path or not location:
        return {"ok": True, "skipped": "no enrollment row"}
    mod = _import_enrollment()
    if mod is None:
        return {"ok": False, "error": "enrollment.py not found"}
    try:
        upsert = getattr(mod, "upsert_enrollment_from_identity", None)
        if upsert is not None:
            local = upsert(int(asset), location, path)
            if not local:
                return {"ok": True, "skipped": "no enrollment row"}
        else:
            upsert_row = getattr(mod, "upsert_enrollment_row", None) or mod.append_local_enrollment_row
            local = upsert_row(int(asset), location, path)
    except (OSError, ValueError) as exc:
        return {"ok": False, "error": str(exc)}
    return {"ok": True, "path": local, "asset": asset, "enrollment_path": path, "enrollment_location": location}


def _stamp_dyzineframe_identity(cfg: FieldConfig) -> dict[str, Any]:
    """Patch asset/bank in DyzineFrame config.ini in place. No process restart.

    Uses line replacement so comments in config.ini are preserved.
    """
    raw = (os.environ.get("DYZINEFRAME_CONFIG") or "/home/topbox/DyzineFrame/config.ini").strip()
    path = Path(raw)
    if not path.is_file():
        return {"ok": False, "skipped": str(path)}
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        return {"ok": False, "error": str(exc), "path": str(path)}
    original = text
    updated: list[str] = []
    asset = (cfg.asset_id or "").strip()
    if asset.isdigit():
        for key in ("asset_number", "pog_asset_number"):
            text, hit = _set_or_append_ini_key(text, key, asset)
            if hit:
                updated.append(key)
    text, hit = _set_or_append_ini_key(text, "bank_number", str(int(cfg.bank)))
    if hit:
        updated.append("bank_number")
    display_mode = "MINI" if cfg.profile == "listen_ohd_mini" else "TBX"
    text, hit = _set_or_append_ini_key(text, "display_mode", display_mode)
    if hit:
        updated.append("display_mode")
    if int(cfg.bank_position) > 0:
        text, hit = _set_or_append_ini_key(text, "bank_position", str(int(cfg.bank_position)))
        if hit:
            updated.append("bank_position")
    enabled = "True" if bool(cfg.log_upload_enabled) else "False"
    text, hit = _set_or_append_ini_key(text, "log_upload_enabled", enabled)
    if hit:
        updated.append("log_upload_enabled")
    if (cfg.enrollment_path or "").strip():
        text, hit = _set_or_append_ini_key(text, "enrollment_path", cfg.enrollment_path.strip())
        if hit:
            updated.append("enrollment_path")
    if (cfg.enrollment_location or "").strip():
        loc = cfg.enrollment_location.strip()
        text, hit = _set_or_append_ini_key(text, "enrollment_location", loc)
        if hit:
            updated.append("enrollment_location")
    try:
        theme = int(cfg.pog_cc_theme)
    except (TypeError, ValueError):
        theme = 1
    if theme < 1 or theme > 5:
        theme = 1
    text, hit = _set_or_append_ini_key(text, "pog_cc_theme", str(theme))
    if hit:
        updated.append("pog_cc_theme")
    try:
        vol = int(cfg.pog_cc_volume)
    except (TypeError, ValueError):
        vol = 5
    if vol < 0 or vol > 62:
        vol = 5
    text, hit = _set_or_append_ini_key(text, "pog_cc_volume", str(vol))
    if hit:
        updated.append("pog_cc_volume")
    if cfg.profile == "listen_ohd_mini":
        # Mini: no Heltec/door/attendant; ads off; meters stay local CSV.
        for key, val in (
            ("test_mode_no_serial", "True"),
            ("disable_ads", "True"),
            ("progressives_source", ""),
            (
                "floor_games_source",
                normalize_floor_games_source(cfg.floor_games_source),
            ),
        ):
            text, hit = _set_or_append_ini_key(text, key, val)
            if hit:
                updated.append(key)
        fg_src = normalize_floor_games_source(cfg.floor_games_source)
        host = sanitize_floor_games_http_host(cfg.floor_games_http_host) if fg_src == "http" else ""
        text, hit = _set_or_append_ini_key(text, "floor_games_http_host", host)
        if hit:
            updated.append("floor_games_http_host")
    if not updated:
        return {"ok": False, "error": "no matching keys", "path": str(path)}
    if text == original:
        return {
            "ok": True,
            "path": str(path),
            "unchanged": True,
            "updated": updated,
            "asset_id": asset,
            "bank": int(cfg.bank),
            "display_mode": display_mode,
        }
    tmp = path.with_suffix(path.suffix + ".tmp")
    try:
        with tmp.open("w", encoding="utf-8", newline="\n") as handle:
            handle.write(text)
        tmp.replace(path)
    except OSError as exc:
        return {"ok": False, "error": str(exc), "path": str(path)}
    return {
        "ok": True,
        "path": str(path),
        "updated": updated,
        "asset_id": asset,
        "bank": int(cfg.bank),
        "display_mode": display_mode,
    }


def _invoke_set_role(cfg: FieldConfig, etc: Path) -> int:
    """Call set_ethergold_role.sh when present (Pi). Returns exit code or -1 if skipped."""
    import subprocess

    script = Path(__file__).resolve().parent / "set_ethergold_role.sh"
    if not script.is_file():
        return -1
    role = render_role_env(cfg)["ETHERGOLD_ROLE"]
    env = os.environ.copy()
    env["UART_SOCKET_ID"] = "1"
    env["CHANNEL"] = "both"
    cmd = ["bash", str(script), role]
    if role == "edge":
        edge = render_edge_env(cfg)
        cmd.extend(
            [
                "--role",
                edge.get("EDGE_ROLE", "peer-a"),
                "--uplink",
                edge.get("EDGE_UPLINK", "wlan0"),
            ]
        )
        if edge.get("EDGE_GAME_IF"):
            cmd.extend(["--game-if", edge["EDGE_GAME_IF"]])
        if profile_wants_dyzineframe(cfg):
            cmd.append("--display-on")
        else:
            cmd.append("--display-off")
    try:
        completed = subprocess.run(cmd, env=env, check=False)
        return int(completed.returncode)
    except OSError:
        return -1


def status_snapshot(cfg: FieldConfig | None = None, etc: Path | None = None) -> dict[str, Any]:
    cfg = cfg or load_field_config(etc)
    eg_version = ""
    try:
        from pog422_core import EG_VERSION, EG_VERSION_SLUG

        eg_version = f"{EG_VERSION} ({EG_VERSION_SLUG})"
    except Exception:
        eg_version = "unknown"
    cabinet: dict[str, Any] = {}
    try:
        from cabinet_identity import load_cabinet_identity

        cabinet = load_cabinet_identity()
    except Exception:
        cabinet = {}
    term = int(cabinet.get("term_number") or cabinet.get("machine_number") or 0)
    role = str(cabinet.get("machine_role") or "")
    pos = int(cfg.bank_position or cabinet.get("bank_position") or 0)
    tid_hex = str(cabinet.get("tid_hex") or cabinet.get("pog_tid") or "") or None
    machine_number = int(cabinet.get("machine_number") or 0) or None
    return {
        "asset_id": cfg.asset_id,
        "bank": cfg.bank,
        "tid": cfg.tid or None,
        "tid_hex": tid_hex,
        "pog_tid": tid_hex,
        "machine_number": machine_number,
        "term_number": machine_number,
        "machine_role": role or None,
        "mode_label": cabinet.get("mode_label") or None,
        "bank_position": pos or None,
        "hostname": cabinet.get("topbox_host") or hostname_for_asset(cfg.asset_id),
        "profile": cfg.profile,
        "hs_path": hs_path_value(cfg),
        "node_role": cfg.node_role or None,
        "eg_version": eg_version,
        "channel": "both",
        "ois_wifi_plane": bool(cfg.ois_wifi_plane),
        "dyzineframe": profile_wants_dyzineframe(cfg),
        "fixture_pos_shown": False,
        "enrollment_location": cfg.enrollment_location or None,
        "enrollment_path": cfg.enrollment_path or None,
        "log_upload_enabled": bool(cfg.log_upload_enabled),
    }


def hostname_for_asset(asset_id: str) -> str:
    from cabinet_identity import hostname_for_asset as _hn

    return _hn(asset_id)


def _maybe_stamp_hostname(cfg: FieldConfig) -> dict[str, Any]:
    """Set OS hostname to topbox_<asset>. Never rename the Linux user topbox."""
    asset = (cfg.asset_id or "").strip()
    wanted = hostname_for_asset(asset)
    # Opt-in: live .144 stays hostname topbox until Chris asks to stamp topbox_<asset>.
    if os.environ.get("ETHERGOLD_STAMP_HOSTNAME", "").strip() not in ("1", "true", "yes"):
        return {"ok": False, "skipped": "ETHERGOLD_STAMP_HOSTNAME not set", "wanted": wanted}
    if os.name != "posix":
        return {"ok": False, "skipped": "not-posix", "wanted": wanted}
    if not asset:
        return {"ok": False, "skipped": "no-asset", "wanted": wanted}
    import shutil
    import subprocess

    ctl = shutil.which("hostnamectl")
    if not ctl:
        return {"ok": False, "skipped": "no-hostnamectl", "wanted": wanted}
    try:
        completed = subprocess.run(
            [ctl, "set-hostname", wanted],
            check=False,
            capture_output=True,
            text=True,
        )
    except OSError as exc:
        return {"ok": False, "error": str(exc), "wanted": wanted}
    return {
        "ok": completed.returncode == 0,
        "wanted": wanted,
        "rc": completed.returncode,
        "stderr": (completed.stderr or "")[:400],
    }


def _persist_apply_identity(cfg: FieldConfig) -> dict[str, Any]:
    out: dict[str, Any] = {"bank_position": int(cfg.bank_position or 0)}
    tid = sanitize_field_tid(cfg.tid)
    if is_listen_profile(cfg) and TID_RE.match(tid):
        try:
            from cabinet_identity import persist_gamestats_identity

            out["pog_tid"] = persist_gamestats_identity(pog_tid=tid)
        except Exception as exc:
            out["pog_tid_error"] = str(exc)
    if int(cfg.bank_position) <= 0:
        return out
    try:
        from cabinet_identity import persist_bank_position

        out["saved"] = persist_bank_position(int(cfg.bank_position))
    except Exception as exc:
        out["error"] = str(exc)
    return out