#-----------------------------------------------------------------------------------------------------------------------
# Ethergold progressive CSV reader — single source for Topbox (local file) and OHD (HTTP URL).
#
# CSV format (UTF-8, header row required):
#   game_id,pennies,updated_utc
#   game_id       - int, 1..16 (1-origin, matches DyzineFrame curr_index / VID_NAMES)
#   pennies       - int, target progressive value in pennies (internal only)
#   updated_utc   - optional ISO-8601 timestamp (informational)
#
# Game 15 has no progressive; pennies should be 0 and get_target(15) returns None.
#
# Source resolution (first match wins):
#   ETHERGOLD_PROGRESSIVES_SOURCE  — local path or http(s) URL
#   ETHERGOLD_PROGRESSIVES_CSV     — local path only (bridge write path alias)
#   config.ini [TOPBOX] progressives_source
#   default local: $ETHERGOLD_ROOT/progressives.csv or ../EtherGold/progressives.csv
#-----------------------------------------------------------------------------------------------------------------------
from __future__ import annotations

import csv
import io
import json
import os
import random
import threading
import time
import urllib.error
import urllib.request
from collections import deque
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from typing import Any, Deque, Dict, Optional, Sequence, Tuple

# Games that show a progressive on OHD (index 15 excluded).
PROGRESSIVE_GAME_IDS = list(range(1, 15)) + [16]

DEFAULT_CSV_NAME = "progressives.csv"
DEFAULT_ETHERGOLD_DIR = "EtherGold"
DEFAULT_ETHERGOLD_ROOT = "/home/topbox/EtherGold"
PROGRESSIVE_SYNC_STATE_NAME = "progressives-sync-state.json"
CONFIG_FILENAME = "config.ini"


def read_config_ini_file(config, ini_path: str) -> None:
    """Read config.ini as UTF-8 (BOM ok) with cp1252 fallback for legacy field files."""
    try:
        with open(ini_path, encoding="utf-8-sig") as handle:
            config.read_file(handle)
    except UnicodeDecodeError:
        with open(ini_path, encoding="cp1252") as handle:
            config.read_file(handle)
HTTP_FETCH_TIMEOUT_S = 5.0
# Master progressive broadcast (~every 5 s). Remote HTTP checks use this cadence; activity
# samples advance only when CSV values change, not on every poll attempt.
REMOTE_POLL_INTERVAL_S = 5.0
# Poll CSV shortly after the Pi's expected broadcast so the new write is on disk first.
REMOTE_POLL_LAG_S = 0.005
# Re-fetch progressives-sync.json to correct clock skew and poll phase drift.
REMOTE_RESYNC_INTERVAL_S = 45.0
# OHD Win11: HTTP polls run on a daemon thread so pygame never blocks on curl.
REMOTE_BACKGROUND_POLL = True
REMOTE_POLL_WAKE_S = 0.05
ACTIVITY_HISTORY_MIN_SAMPLES_DEFAULT = 12
ACTIVITY_HISTORY_WINDOW_SEC_DEFAULT = 60.0

# Demo: add 1..N pennies per tick per game; occasionally reset one jackpot (won).
DEMO_CONTRIBUTION_PENNIES_MIN = 1
DEMO_CONTRIBUTION_PENNIES_MAX = 75
DEMO_WIN_EVERY_N_TICKS = 12
DEMO_SEED_PENNIES = 5000000  # $50,000.00 after a win


def format_dollars(pennies: int) -> str:
    sign = "-" if pennies < 0 else ""
    pennies = abs(pennies)
    return f"{sign}${pennies // 100:,}.{pennies % 100:02d}"


def format_dollars_float(pennies: float) -> str:
    """HUD for display_float; settled cents plus sub-cent when chase is mid-cent."""
    whole = max(0, int(pennies))
    frac = pennies - whole
    if frac > 1e-4:
        return f"{format_dollars(whole)}+{frac:.3f}c"
    return format_dollars(whole)


def parse_display_to_pennies(value: Any, *, default: int = 0) -> int:
    """
    Parse layout/demo amounts to internal cents.
    int = cents; float = dollars; str = "$1,234.56" or "1234.56".
    """
    if value is None:
        return max(0, int(default))
    if isinstance(value, bool):
        return max(0, int(default))
    if isinstance(value, int):
        return max(0, value)
    if isinstance(value, float):
        return max(0, int(round(value * 100)))
    text = str(value).strip().replace("$", "").replace(",", "")
    if not text:
        return max(0, int(default))
    if "." in text:
        whole, frac = text.split(".", 1)
        dollars = int(whole or "0")
        cents = int((frac + "00")[:2])
        return max(0, dollars * 100 + cents)
    return max(0, int(text) * 100)


def is_remote_progressives_source(source: str) -> bool:
    text = source.strip().lower()
    return text.startswith("http://") or text.startswith("https://")


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def default_local_progressives_path() -> str:
    explicit = os.environ.get("ETHERGOLD_PROGRESSIVES_CSV", "").strip()
    if explicit and not is_remote_progressives_source(explicit):
        return explicit
    root = os.environ.get("ETHERGOLD_ROOT", "").strip()
    if root:
        return os.path.join(root, DEFAULT_CSV_NAME)
    base = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(os.path.dirname(base), DEFAULT_ETHERGOLD_DIR, DEFAULT_CSV_NAME)


def resolve_progressives_source(*, config_path: str | None = None) -> str:
    """Resolve progressive CSV source: local filesystem path or http(s) URL."""
    explicit = os.environ.get("ETHERGOLD_PROGRESSIVES_SOURCE", "").strip()
    if explicit:
        return explicit
    csv_env = os.environ.get("ETHERGOLD_PROGRESSIVES_CSV", "").strip()
    if csv_env and not is_remote_progressives_source(csv_env):
        return csv_env

    ini_path = config_path
    if ini_path is None:
        ini_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), CONFIG_FILENAME)
    if os.path.isfile(ini_path):
        try:
            from extended_configparser.parser import ExtendedConfigParser

            config = ExtendedConfigParser()
            read_config_ini_file(config, ini_path)
            if config.has_option("TOPBOX", "PROGRESSIVES_SOURCE"):
                value = config.get("TOPBOX", "PROGRESSIVES_SOURCE").strip()
                if value:
                    return value
        except Exception as exc:
            print(f"EthergoldReader: config read failed ({ini_path}): {exc}")

    return default_local_progressives_path()


def progressives_sync_url_from_source(source: str) -> str:
    """Derive progressives-sync.json URL from a progressives.csv http(s) URL."""
    if source.endswith("progressives.csv"):
        return source[: -len("progressives.csv")] + "progressives-sync.json"
    return source.rstrip("/") + "/progressives-sync.json"


def resolve_progressives_sync_state_path(csv_path: str) -> str:
    return os.path.join(os.path.dirname(csv_path) or ".", PROGRESSIVE_SYNC_STATE_NAME)


def read_progressives_sync_state(csv_path: str) -> Dict[str, Any]:
    """Read bridge-written sidecar (sync_epoch bumps on bridge/listen startup)."""
    state_path = resolve_progressives_sync_state_path(csv_path)
    if not os.path.isfile(state_path):
        return {}
    try:
        with open(state_path, encoding="utf-8") as handle:
            payload = json.load(handle)
        if isinstance(payload, dict):
            return payload
    except (OSError, ValueError, json.JSONDecodeError):
        pass
    return {}


def _http_date_to_epoch(http_date: str | None) -> float | None:
    if not http_date:
        return None
    try:
        return parsedate_to_datetime(http_date).timestamp()
    except (TypeError, ValueError, OverflowError):
        return None


def build_progressives_sync_payload(
    *,
    csv_path: str,
    broadcast_interval_s: float = REMOTE_POLL_INTERVAL_S,
    version: str = "",
) -> Dict[str, float | str]:
    """JSON payload for OHD to phase-lock remote CSV polls to Pi broadcast cadence."""
    if os.path.isfile(csv_path):
        last_epoch = os.path.getmtime(csv_path)
    else:
        last_epoch = 0.0
    interval = max(0.5, float(broadcast_interval_s))
    poll_after = (last_epoch + interval) if last_epoch > 0 else time.time()
    state = read_progressives_sync_state(csv_path)
    sync_epoch = float(state.get("sync_epoch", 0.0) or 0.0)
    payload: Dict[str, float | str] = {
        "last_update_epoch": last_epoch,
        "broadcast_interval_s": interval,
        "poll_after_epoch": poll_after,
        "server_now_epoch": time.time(),
        "version": version,
    }
    if sync_epoch > 0:
        payload["sync_epoch"] = sync_epoch
    return payload


def parse_progressives_csv_text(text: str) -> Dict[int, int]:
    targets: Dict[int, int] = {}
    reader = csv.DictReader(io.StringIO(text))
    if not reader.fieldnames or "game_id" not in reader.fieldnames or "pennies" not in reader.fieldnames:
        raise ValueError("CSV must have header: game_id,pennies[,updated_utc]")
    for row in reader:
        game_id = int(row["game_id"].strip())
        pennies = int(row["pennies"].strip())
        if game_id < 1 or game_id > 16:
            raise ValueError(f"game_id out of range: {game_id}")
        targets[game_id] = pennies
    return targets


def _parse_csv_file(path: str) -> Dict[int, int]:
    with open(path, newline="", encoding="utf-8") as handle:
        return parse_progressives_csv_text(handle.read())


class EthergoldReader:
    """Read progressives.csv from a local path or remote URL with reload-on-change."""

    def __init__(
        self,
        source: str | None = None,
        *,
        csv_path: str | None = None,
        remote_poll_interval_s: float | None = None,
    ):
        if source is not None and csv_path is not None:
            raise ValueError("Pass source or csv_path, not both")
        if csv_path is not None:
            source = csv_path
        self.source = source or resolve_progressives_source()
        self._remote = is_remote_progressives_source(self.source)
        self._remote_poll_interval_s = max(
            0.5, float(remote_poll_interval_s or REMOTE_POLL_INTERVAL_S)
        )
        self._lock = threading.Lock()
        self._targets: Dict[int, int] = {}
        self._prev_snapshot: Dict[int, int] = {}
        self._last_deltas: Dict[int, int] = {}
        self._delta_history: Dict[int, Deque[Tuple[float, int]]] = {}
        self._activity_min_samples = ACTIVITY_HISTORY_MIN_SAMPLES_DEFAULT
        self._activity_window_sec = ACTIVITY_HISTORY_WINDOW_SEC_DEFAULT
        self._mtime: float = 0.0
        self._remote_etag: str | None = None
        self._remote_last_modified: str | None = None
        self._remote_next_poll_mono: float = 0.0
        self._remote_last_sync_mono: float = 0.0
        self._remote_wall_skew: float = 0.0
        self._remote_sync_epoch: float | None = None
        self._last_server_now_epoch: float = 0.0
        self._demo_thread: Optional[threading.Thread] = None
        self._demo_stop = threading.Event()
        self._demo_interval_sec = 3.0
        self._demo_tick_count = 0
        self._warned_once: set[str] = set()
        self._background_poll = bool(self._remote and REMOTE_BACKGROUND_POLL)
        self._remote_changed_flag = False
        self._poll_stop = threading.Event()
        self._poll_thread: Optional[threading.Thread] = None
        self._poll_state_lock = threading.Lock()
        self._last_success_mono: float = 0.0
        self._had_any_success: bool = False
        if not self._remote:
            self.load(force=True)

    @property
    def csv_path(self) -> str:
        """Local path when source is a file; empty string when remote."""
        return "" if self._remote else self.source

    @property
    def source_kind(self) -> str:
        return "remote" if self._remote else "local"

    @property
    def remote_poll_interval_s(self) -> float:
        return self._remote_poll_interval_s

    @property
    def seconds_since_last_success(self) -> float:
        if not self._had_any_success:
            return float("inf")
        return time.monotonic() - self._last_success_mono

    @property
    def had_any_success(self) -> bool:
        return self._had_any_success

    def _warn_once(self, key: str, message: str) -> None:
        if key in self._warned_once:
            return
        self._warned_once.add(key)
        print(message, flush=True)

    def _clear_load_warnings(self) -> None:
        prefix = f"load:{self.source}:"
        self._warned_once = {k for k in self._warned_once if not k.startswith(prefix)}

    def load(self, force: bool = False) -> bool:
        """Load CSV when missing or changed. Returns True if data was reloaded."""
        if self._remote:
            return self._load_remote(force=force)
        return self._load_local(force=force)

    def start_background_poll(self) -> None:
        """Start daemon thread for remote sync JSON + CSV (OHD; no pygame blocking)."""
        if not self._background_poll or self._poll_thread is not None:
            return
        self._poll_stop.clear()
        with self._poll_state_lock:
            self._remote_next_poll_mono = 0.0
        self._poll_thread = threading.Thread(
            target=self._remote_poll_loop,
            name="ethergold_remote_poll",
            daemon=True,
        )
        self._poll_thread.start()
        self._warn_once(
            "bg_poll",
            "EthergoldReader: background remote poll thread started",
        )

    def stop_background_poll(self) -> None:
        if self._poll_thread is None:
            return
        self._poll_stop.set()
        self._poll_thread.join(timeout=REMOTE_POLL_WAKE_S + HTTP_FETCH_TIMEOUT_S + 1.0)
        self._poll_thread = None

    def _remote_poll_once(self) -> bool:
        """One remote poll cycle (sync JSON gate + CSV fetch). Thread-only when background poll on."""
        now_mono = time.monotonic()
        with self._poll_state_lock:
            if (now_mono - self._remote_last_sync_mono) >= REMOTE_RESYNC_INTERVAL_S:
                self._remote_last_sync_mono = now_mono
                do_resync = True
            else:
                do_resync = False
            not_due = now_mono < self._remote_next_poll_mono
        if do_resync:
            self._apply_pi_sync_schedule()
        if not_due:
            return False
        hints = self._fetch_sync_hints()
        if hints is not None:
            last_epoch, interval, poll_after, force_immediate, server_now = hints
            # Fetch after Pi's last CSV write (+ lag), not poll_after (that is the *next* broadcast).
            fetch_after = (
                last_epoch + REMOTE_POLL_LAG_S if last_epoch > 0 else 0.0
            )
            if (
                not force_immediate
                and fetch_after > 0
                and server_now > 0
                and server_now < fetch_after
            ):
                delay = max(0.05, fetch_after - server_now)
                with self._poll_state_lock:
                    self._remote_next_poll_mono = time.monotonic() + delay
                return False
            if (
                not force_immediate
                and fetch_after > 0
                and server_now <= 0
                and time.time() < fetch_after - self._remote_wall_skew
            ):
                delay = max(
                    0.05, fetch_after - self._remote_wall_skew - time.time()
                )
                with self._poll_state_lock:
                    self._remote_next_poll_mono = time.monotonic() + delay
                return False
        changed = self.load(force=False)
        if changed and hints is not None and hints[4] > 0 and hints[0] > 0:
            late_s = hints[4] - (hints[0] + REMOTE_POLL_LAG_S)
            if abs(late_s) > 0.15:
                self._warn_once(
                    "poll_late",
                    f"EthergoldReader: CSV fetch vs Pi last write "
                    f"{'late' if late_s > 0 else 'early'} {abs(late_s):.2f}s "
                    f"(check OHD clock sync if consistently ~1s late)",
                )
        return changed

    def _remote_poll_loop(self) -> None:
        while not self._poll_stop.is_set():
            try:
                self._remote_poll_once()
            except Exception as exc:
                self._warn_once(
                    "bg_poll_exc",
                    f"EthergoldReader: background poll error: {exc}",
                )
            with self._poll_state_lock:
                wait_s = max(0.05, self._remote_next_poll_mono - time.monotonic())
            self._poll_stop.wait(min(wait_s, REMOTE_POLL_WAKE_S))

    def reload_if_changed(self) -> bool:
        if self._remote:
            if self._background_poll:
                if self._poll_thread is None:
                    self.start_background_poll()
                with self._lock:
                    if self._remote_changed_flag:
                        self._remote_changed_flag = False
                        return True
                return False
            return self._remote_poll_once()
        return self._load_local(force=False)

    def _fetch_sync_hints(
        self,
    ) -> Optional[Tuple[float, float, float, bool, float]]:
        """Return (last_update_epoch, interval_s, poll_after_epoch, force_immediate, server_now)."""
        url = progressives_sync_url_from_source(self.source)
        request = urllib.request.Request(url, headers={"Accept": "application/json"})
        try:
            with urllib.request.urlopen(request, timeout=HTTP_FETCH_TIMEOUT_S) as response:
                payload = json.loads(response.read().decode("utf-8"))
            last_epoch = float(payload.get("last_update_epoch", 0.0))
            interval = max(
                0.5,
                float(payload.get("broadcast_interval_s", self._remote_poll_interval_s)),
            )
            poll_after = float(
                payload.get(
                    "poll_after_epoch",
                    last_epoch + interval if last_epoch > 0 else time.time(),
                )
            )
            sync_epoch = float(payload.get("sync_epoch", 0.0) or 0.0)
            server_now = float(payload.get("server_now_epoch", 0.0) or 0.0)
            if server_now > 0:
                measured_skew = server_now - time.time()
                self._last_server_now_epoch = server_now
                if abs(measured_skew) > 0.25:
                    self._warn_once(
                        "clock_skew",
                        f"EthergoldReader: OHD clock skew vs Pi {measured_skew:+.2f}s "
                        "(sync Time settings on OHD if polls stay ~1s late)",
                    )
                if self._remote_wall_skew == 0.0:
                    self._remote_wall_skew = measured_skew
                else:
                    self._remote_wall_skew = (
                        0.85 * self._remote_wall_skew + 0.15 * measured_skew
                    )
            force_immediate = False
            if sync_epoch > 0:
                if (
                    self._remote_sync_epoch is not None
                    and sync_epoch != self._remote_sync_epoch
                ):
                    force_immediate = True
                    with self._poll_state_lock:
                        self._remote_next_poll_mono = 0.0
                    self._remote_etag = None
                    self._remote_last_modified = None
                    self._warn_once(
                        f"sync_epoch:{sync_epoch}",
                        "EthergoldReader: Pi sync_epoch changed "
                        f"({self._remote_sync_epoch:.3f} -> {sync_epoch:.3f}); "
                        "resyncing poll phase",
                    )
                self._remote_sync_epoch = sync_epoch
            return last_epoch, interval, poll_after, force_immediate, server_now
        except (OSError, urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError):
            return None

    def _poll_delay_seconds(self, poll_at_epoch: float, server_now_epoch: float | None = None) -> float:
        """Seconds until poll_at (Pi epoch + lag), using Pi clock when available."""
        server_now = server_now_epoch if server_now_epoch and server_now_epoch > 0 else self._last_server_now_epoch
        if server_now > 0:
            return max(0.05, poll_at_epoch - server_now)
        return max(0.05, poll_at_epoch - self._remote_wall_skew - time.time())

    def _apply_pi_sync_schedule(
        self,
        *,
        poll_after_epoch: float | None = None,
        interval_s: float | None = None,
        last_update_epoch: float | None = None,
        last_modified_header: str | None = None,
        server_now_epoch: float | None = None,
    ) -> None:
        """Schedule next remote CSV poll — prefer Pi progressives-sync.json timing."""
        if poll_after_epoch is None and last_update_epoch is None and last_modified_header is None:
            hints = self._fetch_sync_hints()
            if hints is not None:
                _, interval_s, poll_after_epoch, _, server_now_epoch = hints
        interval = max(0.5, float(interval_s or self._remote_poll_interval_s))
        if poll_after_epoch is not None:
            poll_at = poll_after_epoch + REMOTE_POLL_LAG_S
            delay = self._poll_delay_seconds(poll_at, server_now_epoch)
        elif last_update_epoch is not None and last_update_epoch > 0:
            poll_at = last_update_epoch + interval + REMOTE_POLL_LAG_S
            delay = self._poll_delay_seconds(poll_at, server_now_epoch)
        else:
            last_mod_epoch = _http_date_to_epoch(last_modified_header)
            if last_mod_epoch is not None:
                poll_at = last_mod_epoch + interval + REMOTE_POLL_LAG_S
                delay = self._poll_delay_seconds(poll_at, server_now_epoch)
            else:
                delay = interval
        with self._poll_state_lock:
            self._remote_next_poll_mono = time.monotonic() + delay

    def _schedule_next_remote_poll(
        self,
        *,
        poll_after_epoch: float | None = None,
        last_update_epoch: float | None = None,
        interval_s: float | None = None,
        last_modified_header: str | None = None,
    ) -> None:
        self._apply_pi_sync_schedule(
            poll_after_epoch=poll_after_epoch,
            interval_s=interval_s,
            last_update_epoch=last_update_epoch,
            last_modified_header=last_modified_header,
        )

    def peek_target(self, game_id: int) -> Optional[int]:
        """Cached target only — no reload (use after reload_if_changed)."""
        if game_id not in PROGRESSIVE_GAME_IDS:
            return None
        with self._lock:
            val = self._targets.get(game_id)
        if val is None or val <= 0:
            return None
        return val

    def get_target(self, game_id: int) -> Optional[int]:
        """Target value in pennies, or None if game has no progressive."""
        if game_id not in PROGRESSIVE_GAME_IDS:
            return None
        self.reload_if_changed()
        return self.peek_target(game_id)

    def get_all_targets(self) -> Dict[int, int]:
        self.reload_if_changed()
        with self._lock:
            return {
                g: self._targets[g]
                for g in PROGRESSIVE_GAME_IDS
                if g in self._targets and self._targets[g] > 0
            }

    def configure_activity_averaging(self, min_samples: int, window_sec: float) -> None:
        with self._lock:
            self._activity_min_samples = max(1, int(min_samples))
            self._activity_window_sec = max(1.0, float(window_sec))

    def get_activity_deltas(self) -> Dict[int, int]:
        """Pennies gained per game since the previous CSV load (0 if unchanged)."""
        self.reload_if_changed()
        with self._lock:
            return dict(self._last_deltas)

    def get_average_activity(self, floor_games: Sequence[int]) -> Dict[int, float]:
        """Average delta per game over the rolling capture window (modes 3/4)."""
        self.reload_if_changed()
        with self._lock:
            averages: Dict[int, float] = {}
            for game_id in floor_games:
                history = self._delta_history.get(game_id)
                if not history or len(history) < self._activity_min_samples:
                    continue
                averages[game_id] = sum(delta for _, delta in history) / len(history)
            return averages

    def get_activity_diagnostics(
        self, floor_games: Sequence[int]
    ) -> Dict[int, Dict[str, float | int]]:
        """Per-game sample counts, rolling avg, and latest delta (pennies) for mode 4 dumps."""
        self.reload_if_changed()
        with self._lock:
            out: Dict[int, Dict[str, float | int]] = {}
            for game_id in floor_games:
                history = self._delta_history.get(game_id)
                samples = len(history) if history else 0
                avg = (
                    sum(delta for _, delta in history) / samples
                    if samples
                    else 0.0
                )
                out[game_id] = {
                    "samples": samples,
                    "avg_pennies": avg,
                    "last_delta_pennies": int(self._last_deltas.get(game_id, 0)),
                }
            return out

    def get_highest_value_game(self, floor_games: Optional[Sequence[int]] = None) -> Optional[int]:
        """Game with the largest progressive value; tie → lowest game_id."""
        allowed = tuple(floor_games) if floor_games else PROGRESSIVE_GAME_IDS
        targets = self.get_all_targets()
        filtered = {g: val for g, val in targets.items() if g in allowed}
        if not filtered:
            return allowed[0] if allowed else PROGRESSIVE_GAME_IDS[0]
        return max(filtered.items(), key=lambda item: (item[1], -item[0]))[0]

    def get_most_active_game(
        self,
        floor_games: Optional[Sequence[int]] = None,
        *,
        require_average: bool = False,
    ) -> Optional[int]:
        """Most active game by latest delta, or by rolling average when require_average=True."""
        allowed = tuple(floor_games) if floor_games else PROGRESSIVE_GAME_IDS
        if require_average:
            averages = self.get_average_activity(allowed)
            if not averages:
                return None
            return max(averages.items(), key=lambda item: (item[1], -item[0]))[0]

        deltas = self.get_activity_deltas()
        active = {g: d for g, d in deltas.items() if g in allowed and d > 0}
        if not active:
            return None
        return max(active.items(), key=lambda item: (item[1], -item[0]))[0]

    def _record_targets(self, targets: Dict[int, int]) -> None:
        now = time.monotonic()
        deltas: Dict[int, int] = {}
        for game_id in PROGRESSIVE_GAME_IDS:
            new_val = int(targets.get(game_id, 0))
            old_val = self._prev_snapshot.get(game_id)
            if old_val is not None and new_val >= old_val:
                deltas[game_id] = new_val - old_val
            else:
                deltas[game_id] = 0

            if old_val is not None:
                history = self._delta_history.setdefault(game_id, deque())
                history.append((now, deltas[game_id]))
                while history and (now - history[0][0]) > self._activity_window_sec:
                    history.popleft()

        self._last_deltas = deltas
        self._prev_snapshot = {game_id: int(targets.get(game_id, 0)) for game_id in range(1, 17)}
        self._targets = dict(targets)

    def write_csv(self, targets: Optional[Dict[int, int]] = None) -> None:
        """Write full 1..16 rows to a local CSV (demo spoofer; not valid for remote sources)."""
        if self._remote:
            raise ValueError("write_csv() requires a local progressives source")
        path = self.source
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
        with self._lock:
            data = dict(self._targets) if targets is None else dict(targets)
        rows = []
        stamp = _utc_now_iso()
        for game_id in range(1, 17):
            pennies = data.get(game_id, 0)
            if game_id not in PROGRESSIVE_GAME_IDS:
                pennies = 0
            rows.append((game_id, pennies, stamp))
        with open(path, "w", newline="", encoding="utf-8") as handle:
            writer = csv.writer(handle)
            writer.writerow(["game_id", "pennies", "updated_utc"])
            writer.writerows(rows)
        with self._lock:
            self._record_targets({g: p for g, p, _ in rows})
            self._mtime = os.path.getmtime(path)

    def demo_tick(self) -> None:
        """Simulate contributions and occasional jackpot win (resets one game to seed)."""
        self.load(force=True)
        with self._lock:
            working = dict(self._targets)
        for game_id in PROGRESSIVE_GAME_IDS:
            working[game_id] = working.get(game_id, DEMO_SEED_PENNIES) + random.randint(
                DEMO_CONTRIBUTION_PENNIES_MIN, DEMO_CONTRIBUTION_PENNIES_MAX
            )
        working[15] = 0
        self._demo_tick_count += 1
        if self._demo_tick_count % DEMO_WIN_EVERY_N_TICKS == 0:
            win_game = random.choice(PROGRESSIVE_GAME_IDS)
            working[win_game] = DEMO_SEED_PENNIES
            print(
                f"EthergoldReader demo: jackpot win simulated on game {win_game} "
                f"-> {format_dollars(DEMO_SEED_PENNIES)}"
            )
        self.write_csv(working)

    def _demo_loop(self, interval_sec: float) -> None:
        print(f"EthergoldReader demo: writing {self.source} every {interval_sec}s")
        while not self._demo_stop.wait(interval_sec):
            self.demo_tick()

    def start_demo(self, interval_sec: float = 3.0) -> None:
        if self._remote:
            raise ValueError("start_demo() requires a local progressives source")
        if self._demo_thread and self._demo_thread.is_alive():
            return
        self._demo_interval_sec = interval_sec
        self._demo_stop.clear()
        self._demo_thread = threading.Thread(
            target=self._demo_loop,
            args=(interval_sec,),
            name="ethergold_demo",
            daemon=True,
        )
        self._demo_thread.start()

    def stop_demo(self) -> None:
        self._demo_stop.set()
        if self._demo_thread:
            self._demo_thread.join(timeout=self._demo_interval_sec + 2.0)

    def _load_local(self, *, force: bool) -> bool:
        path = self.source
        if not os.path.isfile(path):
            self._warn_once(
                f"load:{path}:missing",
                f"EthergoldReader: progressives file not found: {path}",
            )
            return False
        mtime = os.path.getmtime(path)
        with self._lock:
            if not force and mtime == self._mtime and self._targets:
                return False
        try:
            targets = _parse_csv_file(path)
        except (OSError, ValueError, csv.Error) as exc:
            self._warn_once(
                f"load:{path}:parse",
                f"EthergoldReader: failed to load {path}: {exc}",
            )
            with self._lock:
                self._mtime = mtime
            return False
        with self._lock:
            self._record_targets(targets)
            self._mtime = mtime
        self._last_success_mono = time.monotonic()
        self._had_any_success = True
        self._clear_load_warnings()
        return True

    def _load_remote(self, *, force: bool) -> bool:
        headers: Dict[str, str] = {"Accept": "text/csv, text/plain, */*"}
        with self._lock:
            if self._remote_etag:
                headers["If-None-Match"] = self._remote_etag
            if self._remote_last_modified:
                headers["If-Modified-Since"] = self._remote_last_modified
            if force:
                headers.pop("If-None-Match", None)
                headers.pop("If-Modified-Since", None)

        request = urllib.request.Request(self.source, headers=headers)
        try:
            with urllib.request.urlopen(request, timeout=HTTP_FETCH_TIMEOUT_S) as response:
                body = response.read().decode("utf-8")
                etag = response.headers.get("ETag")
                last_modified = response.headers.get("Last-Modified")
        except urllib.error.HTTPError as exc:
            if exc.code == 304 and not force:
                self._last_success_mono = time.monotonic()
                self._had_any_success = True
                self._apply_pi_sync_schedule()
                return False
            self._warn_once(
                f"load:{self.source}:http{exc.code}",
                f"EthergoldReader: HTTP {exc.code} for {self.source}",
            )
            self._apply_pi_sync_schedule()
            return False
        except (OSError, urllib.error.URLError, TimeoutError) as exc:
            self._warn_once(
                f"load:{self.source}:fetch",
                f"EthergoldReader: failed to fetch {self.source}: {exc}",
            )
            self._apply_pi_sync_schedule()
            return False

        try:
            targets = parse_progressives_csv_text(body)
        except (ValueError, csv.Error) as exc:
            self._warn_once(
                f"load:{self.source}:csv",
                f"EthergoldReader: invalid CSV from {self.source}: {exc}",
            )
            return False

        self._last_success_mono = time.monotonic()
        self._had_any_success = True
        with self._lock:
            if not force and targets == self._targets:
                self._apply_pi_sync_schedule()
                return False
            self._record_targets(targets)
            if etag:
                self._remote_etag = etag
            if last_modified:
                self._remote_last_modified = last_modified
            if self._background_poll:
                self._remote_changed_flag = True
        self._apply_pi_sync_schedule()
        self._clear_load_warnings()
        return True


# Backward-compatible alias for older call sites.
default_csv_path = default_local_progressives_path