Files
pxe-server/webapp/services/image_pages.py
cproudlock b9a4d0230d Engineer image: upload the Engineering Layer from the web app
New Engineering Layer tab on the engineer image page. A site uploads
Engineering_Install_v<ver>.exe (keeps prereqs) or a .zip of the whole
extra_engineering payload. The upload is streamed to disk, checked (MZ
header, no unsafe zip paths) and swapped in; the replaced payload is
archived (two kept) with a Restore button. Install-EngineeringLayer.ps1
stays repo-owned and now installs the newest Engineering_Install_v*.exe.

KB: Part 4 becomes Site Content (enrollment package, Engineering Layer,
Blancco site settings, Shop Floor apps) plus a new Adding Shop Floor
Apps section for preinstall.json/postppkg.json. WJ examples removed.
2026-10-16 20:48:25 -04:00

352 lines
14 KiB
Python

"""Shared state for the per-image pages (GE Aerospace Images in the sidebar).
Each image has one page with tabs, all drawn under templates/_image_header.html:
tab id label endpoint URL
overview Overview images.overview /images/<t>
drivers Drivers images_drivers.page /images/<t>/drivers
unattend Unattend unattend_editor /images/<t>/unattend
os OS and Packages images.os_packages /images/<t>/os
IMAGE_TABS is the single place that maps a tab to its endpoint: the header
partial, the sidebar and every redirect use it (tab_url). A page that moves
to another endpoint changes its row here and nothing else.
header_info() is what the header partial shows; overview() feeds the
Overview tab. Both only read files, nothing here writes (saves go through
services/control_files.py).
"""
import glob
import os
from pathlib import PurePosixPath
from flask import url_for
import config
from services import control_files, fs, hardware_coverage
IMAGE_TABS = (
{"id": "overview", "label": "Overview", "icon": "bi-clipboard-data", "endpoint": "images.overview"},
{"id": "drivers", "label": "Drivers", "icon": "bi-hdd-network", "endpoint": "images_drivers.page"},
{"id": "unattend", "label": "Unattend", "icon": "bi-file-earmark-code", "endpoint": "unattend_editor"},
{"id": "os", "label": "OS and Packages", "icon": "bi-windows", "endpoint": "images.os_packages"},
# Only on the images listed in "images" (the header skips it elsewhere).
{"id": "englayer", "label": "Engineering Layer", "icon": "bi-box-seam",
"endpoint": "engineering_layer.page", "images": ("gea-engineer",)},
)
TAB_IDS = tuple(t["id"] for t in IMAGE_TABS)
# PESetup loads the first of these that exists (docs/PESETUP-INTERNALS.md,
# "Paths it hardcodes"); the Unattend tab edits the fallback.
UNATTEND_PREFERRED = ("Deploy", "Tools", "GE", "XML", "FlatUnattendW10.xml")
# osId values that mean Windows 10 in the GE catalogue: 18 is the only id
# the Dell_10 packs carry; 19, 20 and 21 appear only on Win11 packs (20 =
# Win11 LTSC 24H2, 21 = Win11 Pro 24H2).
WIN10_OS_IDS = {"18"}
def tab_url(image_type, tab="overview", **kw):
"""URL of one tab of an image page."""
endpoint = next(t["endpoint"] for t in IMAGE_TABS if t["id"] == tab)
return url_for(endpoint, image_type=image_type, **kw)
def tab_for_endpoint(endpoint):
"""Tab id for a request endpoint, else None. Exact match first; then a
tab whose blueprint is its own (images_drivers.*) claims every endpoint
of that blueprint. The shared "images" blueprint only matches exactly."""
for t in IMAGE_TABS:
if endpoint == t["endpoint"]:
return t["id"]
bp = (endpoint or "").partition(".")[0]
for t in IMAGE_TABS:
own = t["endpoint"].partition(".")[0]
if "." in t["endpoint"] and own == bp and own != "images":
return t["id"]
return None
# --- file locations ----------------------------------------------------------
def control_file(image_type, name):
return os.path.join(fs.control_path(image_type), name)
def user_selections_path(image_type):
return os.path.join(fs.tools_path(image_type), "user_selections.json")
def preferred_unattend_path(image_type):
return os.path.join(fs.image_root(image_type), *UNATTEND_PREFERRED)
# --- small readers -------------------------------------------------------------
def _list(data):
return data if isinstance(data, list) else []
def _load(path):
"""fs.load_json that also survives a file that is not UTF-8 (a UTF-16
file saved by PowerShell 5.1 Out-File raises UnicodeDecodeError, which
fs.load_json lets through and which would fail every image tab)."""
try:
return fs.load_json(path)
except ValueError:
return []
def drivers(image_type):
"""HardwareDriver.json entries (the only driver file PESetup reads)."""
return _list(_load(control_file(image_type, "HardwareDriver.json")))
def user_selections(image_type):
"""(first object of user_selections.json, whole list) - the file is a
one-element JSON array written by Media Creator Lite."""
raw = _load(user_selections_path(image_type))
if isinstance(raw, list) and raw and isinstance(raw[0], dict):
return raw[0], raw
return {}, []
def driver_os(entry):
"""'Win11', 'Win10' or 'Other' for one driver entry. The pack file name
(win10_/win11_) decides first, then the Dell_10/Dell_11 folder, then
the osId list."""
name = str(entry.get("fileName") or entry.get("FileName") or "").lower()
dest = str(entry.get("destinationDir") or entry.get("DestinationDir") or "").lower()
if name.startswith("win10"):
return "Win10"
if name.startswith("win11"):
return "Win11"
if "\\dell_10\\" in dest or "/dell_10/" in dest:
return "Win10"
if "\\dell_11\\" in dest or "/dell_11/" in dest:
return "Win11"
ids = [i.strip() for i in str(entry.get("osId") or "").split(",") if i.strip()]
if not ids:
return "Other"
return "Win10" if all(i in WIN10_OS_IDS for i in ids) else "Win11"
def zip_path(image_type, entry):
"""Real path of an entry's zip (directories matched case-insensitively,
as on the live share), or None when it is not on disk."""
rel = hardware_coverage._zip_relpath(entry)
if not rel:
return None
return _find_ci(fs.image_root(image_type), PurePosixPath(rel).parts)
def _find_ci(cur, parts):
"""Depth-first case-insensitive lookup. Every case variant of a folder
is tried (Dell_11 can hold both Optiplex and OptiPlex as separate
folders), the exact spelling first."""
if not parts:
return cur if os.path.isfile(cur) else None
try:
names = os.listdir(cur)
except OSError:
return None
hits = sorted((n for n in names if n.lower() == parts[0].lower()), key=lambda n: n != parts[0])
for hit in hits:
found = _find_ci(os.path.join(cur, hit), parts[1:])
if found:
return found
return None
def _stat(path):
try:
st = os.stat(path)
return {"exists": True, "size": st.st_size, "mtime": st.st_mtime}
except OSError:
return {"exists": False, "size": None, "mtime": None}
def operating_systems(image_type):
"""OperatingSystem.json entries with the wim resolved on disk:
[{"entry", "osv", "wim_dir", "wim_path", "wim": _stat}]."""
out = []
for entry in _list(_load(control_file(image_type, "OperatingSystem.json"))):
osv = entry.get("operatingSystemVersion") or {}
wim = osv.get("wim") or {}
dest = wim.get("DestinationDir") or wim.get("destinationDir") or ""
wim_dir = fs.resolve_destination(dest, image_type)
wim_path = os.path.join(wim_dir, "install.wim") if wim_dir else ""
out.append({"entry": entry, "osv": osv, "dest": dest, "wim_path": wim_path,
"wim": _stat(wim_path) if wim_path else _stat("")})
return out
def packages(image_type):
"""packages.json entries with the file resolved on disk:
[{"entry", "path", "file": _stat}]."""
out = []
for entry in _list(_load(control_file(image_type, "packages.json"))):
name = entry.get("fileName") or entry.get("FileName") or ""
dest = entry.get("destinationDir") or entry.get("DestinationDir") or ""
folder = fs.resolve_destination(dest, image_type)
path = os.path.join(folder, name) if folder and name else ""
out.append({"entry": entry, "path": path, "file": _stat(path) if path else _stat("")})
return out
def _current_os(rows, selection):
"""The OS row imaging uses: the one user_selections.json selects, else
the first active one, else the first."""
want = str(selection if selection is not None else "")
return (next((r for r in rows if str(r["osv"].get("id")) == want), None)
or next((r for r in rows if r["osv"].get("isActive")), None)
or (rows[0] if rows else None))
def os_label(image_type):
"""'Windows 11 Professional 24H2 (osId 21)' for the selected OS, else
the first active entry, else ''."""
sel, _ = user_selections(image_type)
row = _current_os(operating_systems(image_type), sel.get("OperatingSystemSelection"))
if not row:
return ""
pick = row["osv"]
name = pick.get("productName") or pick.get("marketingName") or "Windows"
ver = pick.get("versionNumber") or ""
return f"{name} {ver} (osId {pick.get('id')})".replace(" ", " ")
# --- header + overview ---------------------------------------------------------------
def header_info(image_type):
"""What _image_header.html shows: name, OS, counts for the tab pills."""
entries = drivers(image_type)
return {
"image_type": image_type,
"friendly_name": config.FRIENDLY_NAMES.get(image_type, image_type),
"os_label": os_label(image_type),
"driver_count": len(entries),
"package_count": len(packages(image_type)),
"has_unattend": os.path.isfile(fs.unattend_path(image_type)),
}
def _lint_counts(image_type):
"""Findings of services/driver_lint (built by the Drivers tab) as
{"available", "total", "by_level"}; unavailable until that module
exists or when it fails."""
try:
from services import driver_lint # noqa: WPS433 (optional module)
findings = driver_lint.lint_image(image_type)
except Exception: # missing module or a bad map: show "not available"
return {"available": False, "total": 0, "by_level": {}, "map_errors": 0}
by_level = {}
map_errors = 0
for f in findings or []:
f = f if isinstance(f, dict) else vars(f)
level = str(f.get("level") or "INFO").upper()
by_level[level] = by_level.get(level, 0) + 1
# A missing zip is an ERROR finding too, but the Overview counts
# those on their own tile; map_errors is what is wrong in the map.
if level in ("CRITICAL", "ERROR") and f.get("check") != "missing-zip":
map_errors += 1
return {"available": True, "total": sum(by_level.values()), "by_level": by_level,
"map_errors": map_errors}
def _control_files(image_type):
"""The files the image pages edit, with size/mtime and backups."""
items = [
("HardwareDriver.json", control_file(image_type, "HardwareDriver.json"), "drivers",
"Driver map PESetup matches the PC model against"),
("OperatingSystem.json", control_file(image_type, "OperatingSystem.json"), "os",
"Windows build, install.wim location and index"),
("packages.json", control_file(image_type, "packages.json"), "os",
"Servicing packages applied offline (updates, .NET)"),
("user_selections.json", user_selections_path(image_type), "os",
"Media Creator Lite selections (OS and driver families)"),
("FlatUnattendW10.xml", fs.unattend_path(image_type), "unattend",
"Windows answer file (fallback location, the one this app edits)"),
]
out = []
for name, path, tab, what in items:
st = _stat(path)
backups = sorted(glob.glob(glob.escape(path) + ".bak-*"))
out.append(dict(st, name=name, path=path,
rel=os.path.relpath(path, fs.image_root(image_type)),
tab=tab, what=what, backups=len(backups)))
return out
def overview(image_type):
"""Everything the Overview tab shows, from data already on the server."""
entries = drivers(image_type)
by_os = {"Win11": 0, "Win10": 0, "Other": 0}
missing = []
for i, e in enumerate(entries, 1):
by_os[driver_os(e)] += 1
if not zip_path(image_type, e):
missing.append({"pos": i, "file": e.get("fileName") or e.get("FileName") or "",
"models": e.get("modelswminame") or e.get("models") or "",
"os": driver_os(e)})
sel, _ = user_selections(image_type)
oses = operating_systems(image_type)
pkgs = packages(image_type)
# Hardware seen by WinPE with no working driver entry in THIS image.
gaps = []
try:
for row in hardware_coverage.coverage(include_ok=False):
d = (row.get("drivers") or {}).get(image_type)
if d and d["status"] in ("no-entry", "zip-missing") and not row.get("dismissed"):
gaps.append({"model": row["model"], "status": d["status"],
"file": d.get("file", ""), "last_seen": row.get("last_seen"),
"skus": row.get("skus") or []})
except Exception:
pass
covered = image_type in dict(hardware_coverage.IMAGES)
# Newer Dell packs for files this map uses (last Updates > Dell scan).
dell = {"checked": None, "rows": []}
try:
from services import dell_updates
hints = dell_updates.update_hints()["drivers"]
dell["checked"] = (dell_updates.last_scan() or {}).get("checked")
seen = set()
for i, e in enumerate(entries, 1):
name = str(e.get("fileName") or e.get("FileName") or "")
if name.lower() in hints and name.lower() not in seen:
seen.add(name.lower())
dell["rows"].append({"pos": i, "file": name, "ver": hints[name.lower()],
"models": e.get("modelswminame") or ""})
except Exception:
pass
preferred = preferred_unattend_path(image_type)
hw_extra = control_file(image_type, "hw_drivers.json")
families = {str(e.get("family") or "") for e in entries}
models = sel.get("HardwareModelSelection") or []
return {
"drivers": {"total": len(entries), "by_os": by_os, "missing": missing},
"lint": _lint_counts(image_type),
"os": {"selection": sel.get("OperatingSystemSelection"), "rows": oses,
"current": _current_os(oses, sel.get("OperatingSystemSelection"))},
"packages": {
"total": len(pkgs),
"enabled": sum(1 for p in pkgs if p["entry"].get("enabled")),
"missing": [p for p in pkgs if p["entry"].get("enabled") and not p["file"]["exists"]],
},
"models": {"total": len(models),
"no_entry": [m for m in models if str(m.get("Id") or "") not in families]},
"files": _control_files(image_type),
"gaps": gaps,
"gaps_checked": covered,
"dell": dell,
"unattend_override": preferred if os.path.isfile(preferred) else None,
"hw_drivers_extra": hw_extra if os.path.isfile(hw_extra) else None,
"last_saves": {scope: control_files.last_save(image_type, scope)
for scope in ("drivers", "unattend", "os")},
}