LAN7800 EEPROM Programmer

"""Bulk program mode: program many LAN7800 adapters in one run with an
auto-incrementing identity (MAC + serial), a RESUMABLE manifest, and
verify-after-write.

The sequence math and manifest here are hardware-free and unit-tested. The
per-unit burn reuses the exact LED-mux-gated UsbEepromBackend.program_image +
read-back verify that `program` uses.

SAFETY: a real identity write requires --force-physical-write AND (per the mesh)
a session-scoped allow for the run. The default and the dev/test path is
--dry-run: compute + log the whole sequence and write DRYRUN manifest rows, with
NO device access and NO writes.
"""
from __future__ import annotations

import csv
import json
import os
from datetime import datetime, timezone

from .eeprom import Eeprom, EepromError, parse_mac


# ---- MAC / serial sequence math (pure, no hardware) -----------------------
def mac_to_int(octets) -> int:
    v = 0
    for b in octets:
        v = (v << 8) | b
    return v


def int_to_mac(v: int) -> list:
    if not 0 <= v <= 0xFFFFFFFFFFFF:
        raise EepromError(f"MAC integer out of 48-bit range: 0x{v:012X}")
    return [(v >> (8 * (5 - i))) & 0xFF for i in range(6)]


def mac_str(octets) -> str:
    return ":".join(f"{b:02X}" for b in octets)


def serial_for(octets, scheme: str = "mac-hex") -> str:
    """Default scheme: iSerial = the MAC as uppercase hex with no separators,
    so the serial tracks the MAC 1:1 (matches the factory 00800F780000)."""
    if scheme == "mac-hex":
        return "".join(f"{b:02X}" for b in octets)
    raise EepromError(f"unknown serial scheme {scheme!r} (known: mac-hex)")


def apply_local_admin(octets) -> list:
    """Force the locally-administered bit (0x02) and clear the multicast bit
    (0x01) of the first octet -- the '02:'-style prefix option."""
    o = list(octets)
    o[0] = (o[0] | 0x02) & ~0x01
    return o


def sequence(base_mac: str, count: int, step: int = 1,
             local_admin: bool = False, scheme: str = "mac-hex") -> list:
    """Compute the full [{index, mac, serial}] plan. Refuses a multicast base,
    a run that overflows 48 bits, or one that crosses out of the base OUI
    (top 24 bits) -- i.e. NIC space exhausted."""
    if count < 1:
        raise EepromError("count must be >= 1")
    if step == 0:
        raise EepromError("step must be non-zero")
    base = parse_mac(base_mac)
    if local_admin:
        base = apply_local_admin(base)
    if base[0] & 0x01:
        raise EepromError(f"base MAC {mac_str(base)} is MULTICAST (low bit of "
                          "first octet set); a NIC MAC must be unicast")
    base_int = mac_to_int(base)
    base_oui = base_int >> 24
    out = []
    seen = set()
    for i in range(count):
        v = base_int + i * step
        octets = int_to_mac(v)               # raises on 48-bit overflow
        if (v >> 24) != base_oui:
            raise EepromError(
                f"unit {i}: MAC {mac_str(octets)} crosses out of the base OUI "
                f"{base_oui:06X} (NIC space exhausted) -- reduce count/step or "
                "pick a base with more headroom")
        m = mac_str(octets)
        if m in seen:
            raise EepromError(f"internal: duplicate MAC {m} in sequence")
        seen.add(m)
        out.append({"index": i, "mac": m, "serial": serial_for(octets, scheme)})
    return out


# ---- per-unit image build -------------------------------------------------
def build_image(template: Eeprom, mac: str, serial: str,
                set_serial: bool = True) -> bytes:
    """Clone the template and stamp this unit's identity. Signature is ensured;
    program_image writes byte0 (0xA5) LAST for anti-brick ordering."""
    e = Eeprom.from_bytes(template.to_bytes())
    e.set_signature()
    e.set_mac(mac)
    if set_serial:
        try:
            e.set_string_inplace("serial", serial)
        except EepromError as ex:
            raise EepromError(
                f"cannot stamp serial {serial!r}: {ex}. Provide a --template "
                "whose 'serial' string slot fits the 12-char MAC-hex serial.")
    return e.to_bytes()


# ---- resumable manifest (CSV + JSON) --------------------------------------
MANIFEST_FIELDS = ["index", "timestamp", "usb_path", "old_mac", "old_serial",
                   "new_mac", "new_serial", "verify"]
_DONE_STATES = ("PASS", "DRYRUN")


class Manifest:
    """Append-only run record, mirrored to <path>.csv and <path>.json. Resumable:
    next_index() skips completed rows and burned_macs() blocks MAC reuse."""

    def __init__(self, csv_path: str):
        root = os.path.splitext(csv_path)[0]
        self.csv_path = root + ".csv" if not csv_path.endswith(".csv") else csv_path
        self.json_path = root + ".json"
        self.rows = []
        self._load()

    def _load(self):
        if os.path.exists(self.csv_path):
            with open(self.csv_path, newline="") as f:
                self.rows = [dict(r) for r in csv.DictReader(f)]

    def burned_macs(self) -> set:
        return {(r.get("new_mac") or "").upper()
                for r in self.rows
                if str(r.get("verify", "")).upper() in _DONE_STATES
                and r.get("new_mac")}

    def has_mac(self, mac: str) -> bool:
        return mac.upper() in self.burned_macs()

    def next_index(self) -> int:
        done = [int(r["index"]) for r in self.rows
                if str(r.get("verify", "")).upper() in _DONE_STATES]
        return (max(done) + 1) if done else 0

    def last_burned_mac(self):
        """MAC of the most recent REAL (PASS) burn, for the not-swapped guard.
        DRYRUN rows are ignored (a preview never programmed a physical unit)."""
        for r in reversed(self.rows):
            if str(r.get("verify", "")).upper() == "PASS" and r.get("new_mac"):
                return r["new_mac"]
        return None

    def append(self, *, index, usb_path, old_mac, old_serial,
               new_mac, new_serial, verify):
        if verify.upper() in _DONE_STATES and self.has_mac(new_mac):
            raise EepromError(
                f"REFUSING to reuse MAC {new_mac}: already burned in this run "
                f"(manifest {self.csv_path}). Each MAC must be unique.")
        self.rows.append({
            "index": index,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "usb_path": usb_path or "",
            "old_mac": old_mac or "",
            "old_serial": old_serial or "",
            "new_mac": new_mac or "",
            "new_serial": new_serial or "",
            "verify": verify,
        })
        self._flush()

    def _flush(self):
        with open(self.csv_path, "w", newline="") as f:
            w = csv.DictWriter(f, fieldnames=MANIFEST_FIELDS)
            w.writeheader()
            for r in self.rows:
                w.writerow({k: r.get(k, "") for k in MANIFEST_FIELDS})
        with open(self.json_path, "w") as f:
            json.dump(self.rows, f, indent=2)