pxe-server/webapp/services/hardware_coverage.py

"""Which hardware models the PXE server can image, and which it cannot.

WinPE (playbook/winpe-status-push.ps1) reports each machine's WMI model,
manufacturer, Dell system SKU and BIOS version to /hardware/seen before
imaging starts. This module records the models seen and checks each one
against what the server actually has:

  * drivers: every image's Deploy/Control/HardwareDriver.json, matched the
    way PESetup's GetDriverByModel matches (scripts/lint-driver-catalogue.py
    reimplements it and documents why), plus whether the pack's zip is on
    disk;
  * BIOS: the WinPE BIOS manifest (models.txt, substring match like
    check-bios.cmd) and whether the machine is already on that version.

A model missing drivers for an image, or with no BIOS entry, becomes a
technician to-do until it is fixed (the check re-runs on every page load,
so adding the pack clears it) or dismissed.

State: one JSON file, config.HARDWARE_STATE, written atomically:
  {"models": {model: {manufacturer, skus[], serials[], bios_versions[],
                      pctypes[], first_seen, last_seen, count}},
   "dismissed": {model: {"ts"}}}
"""
from __future__ import annotations

import json
import os
import tempfile
import time
from datetime import datetime, timezone
from pathlib import PurePosixPath

import config

# Images whose HardwareDriver.json is checked, with the Windows they deploy.
# All three deploy Windows 11 now (engineer moved off Win10), so every one
# counts toward a model's gaps.
IMAGES = (
    ("gea-standard", "Win11"),
    ("gea-shopfloor", "Win11 LTSC"),
    ("gea-engineer", "Win11"),
)
KEEP = 10  # serials / versions remembered per model


# --- PESetup's matcher (see scripts/lint-driver-catalogue.py) ---------------

def _model_family(model):
    upper = model.upper()
    family = ""
    if "LATITUDE" in upper:
        family = "Latitude"
    if "OPTIPLEX" in upper:
        family = "Optiplex"
    if "PRECISION" in upper:
        family = "Precision"
    return family


def _entry_matches(entry, model):
    family = _model_family(model)
    if family and family.lower() not in str(entry.get("family") or "").lower():
        return False
    # Not trimmed, case-insensitive substring: what the C# does.
    return any(t.lower() in model.lower()
               for t in str(entry.get("modelswminame") or "").split(","))


def _zip_relpath(entry):
    dest = (str(entry.get("destinationDir") or entry.get("DestinationDir") or "")
            .replace("*destinationdir*\\", "").replace("*destinationdir*", "")
            .replace("\\", "/").strip("/"))
    name = str(entry.get("fileName") or entry.get("FileName") or "")
    return str(PurePosixPath(dest) / name) if name else None


def _exists_ci(root, relpath):
    """Path exists, matching each component case-insensitively (Dell_11/
    Optiplex vs OptiPlex has split driver trees before)."""
    cur = root
    for part in PurePosixPath(relpath).parts:
        try:
            names = os.listdir(cur)
        except OSError:
            return False
        hit = next((n for n in names if n.lower() == part.lower()), None)
        if hit is None:
            return False
        cur = os.path.join(cur, hit)
    return os.path.isfile(cur)


_catalog_cache = {"at": 0.0, "value": None}


def _catalogs(ttl=60.0):
    now = time.monotonic()
    if _catalog_cache["value"] is None or now - _catalog_cache["at"] > ttl:
        cats = {}
        for image, _ in IMAGES:
            path = os.path.join(config.SAMBA_SHARE, image, "Deploy", "Control", "HardwareDriver.json")
            try:
                with open(path, encoding="utf-8-sig") as f:
                    data = json.load(f)
                cats[image] = data if isinstance(data, list) else []
            except (OSError, ValueError):
                cats[image] = None  # image not present on this server
        _catalog_cache.update(at=now, value=cats)
    return _catalog_cache["value"]


def driver_status(model):
    """Per image: {"status": ok|no-entry|zip-missing|no-image, "file": ...}."""
    out = {}
    for image, os_label in IMAGES:
        entries = _catalogs().get(image)
        if entries is None:
            out[image] = {"status": "no-image", "os": os_label}
            continue
        entry = next((e for e in entries if _entry_matches(e, model)), None)
        if entry is None:
            out[image] = {"status": "no-entry", "os": os_label}
            continue
        rel = _zip_relpath(entry)
        root = os.path.join(config.SAMBA_SHARE, image)
        ok = bool(rel) and _exists_ci(root, rel)
        out[image] = {"status": "ok" if ok else "zip-missing", "os": os_label,
                      "file": entry.get("fileName") or entry.get("FileName") or ""}
    return out


def _ver(v):
    parts = []
    for p in str(v or "").replace("A", "").split("."):
        try:
            parts.append(int(p))
        except ValueError:
            parts.append(0)
    return tuple(parts)


def bios_status(model, current_version=""):
    """{"status": none|update|current|file-missing, "target", "file"}."""
    manifest = os.path.join(config.BIOS_DIR, "models.txt")
    try:
        with open(manifest, encoding="utf-8", errors="replace") as f:
            lines = f.read().splitlines()
    except OSError:
        return {"status": "none", "target": "", "file": ""}
    for line in lines:
        line = line.strip()
        if not line or line.startswith("#") or line.count("|") < 2:
            continue
        sub, bfile, target = (x.strip() for x in line.split("|")[:3])
        if sub and sub.lower() in model.lower():
            if not os.path.isfile(os.path.join(config.BIOS_DIR, bfile)):
                return {"status": "file-missing", "target": target, "file": bfile}
            if current_version and _ver(current_version) >= _ver(target):
                return {"status": "current", "target": target, "file": bfile}
            return {"status": "update", "target": target, "file": bfile}
    return {"status": "none", "target": "", "file": ""}


# --- state ------------------------------------------------------------------

def _load():
    try:
        with open(config.HARDWARE_STATE) as f:
            state = json.load(f)
    except (OSError, ValueError):
        state = {}
    state.setdefault("models", {})
    state.setdefault("dismissed", {})
    return state


def _save(state):
    folder = os.path.dirname(config.HARDWARE_STATE)
    os.makedirs(folder, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=folder, prefix=".hardware-")
    with os.fdopen(fd, "w") as f:
        json.dump(state, f, indent=1, sort_keys=True)
    os.replace(tmp, config.HARDWARE_STATE)


def _now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _push(lst, value):
    if value and value not in lst:
        lst.append(value)
    del lst[:-KEEP]


def record(payload):
    """Remember one machine sighting from WinPE."""
    model = " ".join(str(payload.get("model") or "").split())
    if not model:
        raise ValueError("missing model")
    model = model[:120]
    state = _load()
    rec = state["models"].setdefault(model, {
        "first_seen": _now(), "skus": [], "serials": [], "bios_versions": [],
        "pctypes": [], "count": 0})
    rec["manufacturer"] = str(payload.get("manufacturer") or rec.get("manufacturer") or "")[:60]
    rec["last_seen"] = _now()
    rec["count"] = int(rec.get("count") or 0) + 1
    _push(rec["skus"], str(payload.get("sku") or "").strip()[:40])
    _push(rec["serials"], str(payload.get("serial") or "").strip()[:64])
    _push(rec["bios_versions"], str(payload.get("bios_version") or "").strip()[:40])
    _push(rec["pctypes"], str(payload.get("pctype") or "").strip()[:60])
    rec["last_bios_version"] = str(payload.get("bios_version") or "").strip()[:40]
    _save(state)
    invalidate()


def dismiss(model, done=True):
    state = _load()
    if model not in state["models"]:
        raise ValueError("unknown model")
    if done:
        state["dismissed"][model] = {"ts": _now()}
    else:
        state["dismissed"].pop(model, None)
    _save(state)
    invalidate()


# Virtual machines and other non-Dell hardware have no packs by design.
_SKIP = ("virtual", "vmware", "qemu", "hyper-v", "kvm")


def coverage(include_ok=True):
    """Every model seen, newest first, with driver and BIOS status."""
    state = _load()
    rows = []
    for model, rec in state["models"].items():
        if any(s in model.lower() for s in _SKIP):
            continue
        drivers = driver_status(model)
        bios = bios_status(model, rec.get("last_bios_version", ""))
        missing = [img for img, d in drivers.items()
                   if d["status"] in ("no-entry", "zip-missing")]
        gap = bool(missing) or bios["status"] in ("none", "file-missing")
        row = dict(rec, model=model, drivers=drivers, bios=bios, missing=missing,
                   gap=gap, dismissed=model in state["dismissed"])
        if include_ok or gap:
            rows.append(row)
    rows.sort(key=lambda r: r.get("last_seen") or "", reverse=True)
    return rows


def open_gaps():
    return [r for r in coverage(include_ok=False) if not r["dismissed"]]


_count_cache = {"at": 0.0, "value": None}


def gap_count(ttl=60.0):
    now = time.monotonic()
    if _count_cache["value"] is None or now - _count_cache["at"] > ttl:
        try:
            _count_cache["value"] = len(open_gaps())
        except Exception:
            _count_cache["value"] = 0
        _count_cache["at"] = now
    return _count_cache["value"]


def invalidate():
    _count_cache["value"] = None
    _catalog_cache["value"] = None