diff --git a/docs/reference/output-layout.md b/docs/reference/output-layout.md index d2aee17cd..51f903b6c 100644 --- a/docs/reference/output-layout.md +++ b/docs/reference/output-layout.md @@ -40,7 +40,7 @@ output/ | `model.onnx.data` | External weight data (only if model ≥ 100 MiB). Must stay alongside `model.onnx`. | | `winml_build_config.json` | The complete pipeline config used for this build (includes auto-discovered optimization flags). This file is a **reproducible pipeline specification** — check it into version control or feed it directly to `winml build -c` in a CI/CD pipeline to guarantee identical model processing across machines and runs (set `"auto": false` for fully deterministic builds). | | `analyze_result.json` | Static analysis output: EP compatibility, operator classification, detected patterns. | -| `build_manifest.json` | Build provenance with stage timings. Only generated via the Python API (`build_hf_model`/`build_onnx_model`). | +| `build_manifest.json` | Build provenance with stage timings. Generated by `winml build` and the Python API (`build_hf_model`/`build_onnx_model`). | | `export_htp_metadata.json` | HTP export metadata: module hierarchy, tracing info, tagging coverage. | ### Intermediate Files (Can Delete After Build) diff --git a/src/winml/modelkit/build/hf.py b/src/winml/modelkit/build/hf.py index 3fe0f462c..888c65df8 100644 --- a/src/winml/modelkit/build/hf.py +++ b/src/winml/modelkit/build/hf.py @@ -31,6 +31,7 @@ from ..export import export_onnx from ..onnx import copy_onnx_model from ..quant import quantize_onnx +from ..utils import MANIFEST_FILENAME, ManifestStage, WinMLManifest from .common import ensure_pre_quantized_stamped, run_optimize_analyze_loop @@ -174,7 +175,7 @@ def _name(base: str) -> str: compiled_path = output_dir / _name("compiled.onnx") final_path = output_dir / _name("model.onnx") config_path = output_dir / _name("winml_build_config.json") - manifest_path = output_dir / _name("build_manifest.json") + manifest_path = output_dir / _name(MANIFEST_FILENAME) analyze_result_path = output_dir / _name("analyze_result.json") # Check for existing artifact (skip build if present and not rebuilding) @@ -369,21 +370,7 @@ def _name(base: str) -> str: # ========================================================================= # [7] BUILD MANIFEST — Machine-readable build provenance # ========================================================================= - manifest: dict[str, Any] = { - "schema_version": 1, - "model_id": model_label, - "task": task, - "cache_key": cache_key, - "config_hash": cache_key.rsplit("_", 1)[-1] if cache_key and "_" in cache_key else None, - "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), - "elapsed_seconds": round(elapsed, 3), - "stages": [], - "final_artifact": final_path.name, - "analyze_iterations": analyze_iterations, - "analyze_unsupported_node_count": analyze_unsupported_nodes, - "analyze_details": analyze_details, - } - + manifest_stages: list[ManifestStage] = [] stage_filenames = { "export": export_path.name, "optimize": optimized_path.name, @@ -392,33 +379,37 @@ def _name(base: str) -> str: } for stage_name in ["export", "optimize", "quantize", "compile"]: if stage_name in stages_completed: - entry: dict[str, Any] = { - "name": stage_name, - "status": "completed", - "filename": stage_filenames[stage_name], - "elapsed_seconds": round(stage_timings.get(stage_name, 0), 3), - } + stage = ManifestStage( + name=stage_name, + status="completed", + filename=stage_filenames[stage_name], + elapsed_seconds=round(stage_timings.get(stage_name, 0), 3), + ) # Thread QuantizeResult metrics into manifest if stage_name == "quantize" and quant_result is not None: - entry["nodes_quantized"] = quant_result.nodes_quantized - entry["nodes_skipped"] = quant_result.nodes_skipped - entry["calibration_time_seconds"] = round(quant_result.calibration_time_seconds, 3) - entry["qdq_insertion_time_seconds"] = round( - quant_result.qdq_insertion_time_seconds, 3 - ) - manifest["stages"].append(entry) + stage.nodes_quantized = quant_result.nodes_quantized + stage.nodes_skipped = quant_result.nodes_skipped + stage.calibration_time_seconds = round(quant_result.calibration_time_seconds, 3) + stage.qdq_insertion_time_seconds = round(quant_result.qdq_insertion_time_seconds, 3) + manifest_stages.append(stage) elif stage_name in stages_skipped: - manifest["stages"].append( - { - "name": stage_name, - "status": "skipped", - "filename": None, - "elapsed_seconds": None, - } - ) + manifest_stages.append(ManifestStage(name=stage_name, status="skipped")) - manifest_path.write_text(json.dumps(manifest, indent=2)) - logger.debug("Build manifest persisted: %s", manifest_path) + manifest = WinMLManifest( + source="hf", + model_id=model_label, + task=task, + cache_key=cache_key, + config_hash=cache_key.rsplit("_", 1)[-1] if cache_key and "_" in cache_key else None, + timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), + elapsed_seconds=round(elapsed, 3), + final_artifact=final_path.name, + stages=manifest_stages, + analyze_iterations=analyze_iterations, + analyze_unsupported_node_count=analyze_unsupported_nodes, + analyze_details=analyze_details, + ) + manifest.save(manifest_path) return BuildResult( output_dir=output_dir, diff --git a/src/winml/modelkit/build/onnx.py b/src/winml/modelkit/build/onnx.py index 58521c73c..8e0095e5e 100644 --- a/src/winml/modelkit/build/onnx.py +++ b/src/winml/modelkit/build/onnx.py @@ -23,6 +23,7 @@ from ..compiler import compile_onnx from ..onnx import copy_onnx_model from ..quant import quantize_onnx +from ..utils import MANIFEST_FILENAME, ManifestStage, WinMLManifest from .common import ensure_pre_quantized_stamped, run_optimize_analyze_loop from .hf import BuildResult @@ -116,7 +117,7 @@ def _name(base: str) -> str: compiled_path = output_dir / _name("compiled.onnx") final_path = output_dir / _name("model.onnx") config_path = output_dir / _name("winml_build_config.json") - manifest_path = output_dir / _name("build_manifest.json") + manifest_path = output_dir / _name(MANIFEST_FILENAME) analyze_result_path = output_dir / _name("analyze_result.json") # Check for existing artifact (skip build if present and not rebuilding) @@ -276,19 +277,7 @@ def _name(base: str) -> str: # ========================================================================= # [5] BUILD MANIFEST — Machine-readable build provenance # ========================================================================= - manifest: dict[str, Any] = { - "schema_version": 1, - "source": "onnx", - "input_onnx": str(onnx_path), - "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), - "elapsed_seconds": round(elapsed, 3), - "stages": [], - "final_artifact": final_path.name, - "analyze_iterations": analyze_iters, - "analyze_unsupported_node_count": analyze_unsupported, - "analyze_details": analyze_details, - } - + manifest_stages: list[ManifestStage] = [] stage_filenames = { "optimize": optimized_path.name, "quantize": quantized_path.name, @@ -296,33 +285,34 @@ def _name(base: str) -> str: } for stage_name in ["optimize", "quantize", "compile"]: if stage_name in stages_completed: - entry: dict[str, Any] = { - "name": stage_name, - "status": "completed", - "filename": stage_filenames[stage_name], - "elapsed_seconds": round(stage_timings.get(stage_name, 0), 3), - } + stage = ManifestStage( + name=stage_name, + status="completed", + filename=stage_filenames[stage_name], + elapsed_seconds=round(stage_timings.get(stage_name, 0), 3), + ) # Thread QuantizeResult metrics into manifest if stage_name == "quantize" and quant_result is not None: - entry["nodes_quantized"] = quant_result.nodes_quantized - entry["nodes_skipped"] = quant_result.nodes_skipped - entry["calibration_time_seconds"] = round(quant_result.calibration_time_seconds, 3) - entry["qdq_insertion_time_seconds"] = round( - quant_result.qdq_insertion_time_seconds, 3 - ) - manifest["stages"].append(entry) + stage.nodes_quantized = quant_result.nodes_quantized + stage.nodes_skipped = quant_result.nodes_skipped + stage.calibration_time_seconds = round(quant_result.calibration_time_seconds, 3) + stage.qdq_insertion_time_seconds = round(quant_result.qdq_insertion_time_seconds, 3) + manifest_stages.append(stage) elif stage_name in stages_skipped: - manifest["stages"].append( - { - "name": stage_name, - "status": "skipped", - "filename": None, - "elapsed_seconds": None, - } - ) - - manifest_path.write_text(json.dumps(manifest, indent=2)) - logger.debug("Build manifest persisted: %s", manifest_path) + manifest_stages.append(ManifestStage(name=stage_name, status="skipped")) + + manifest = WinMLManifest( + source="onnx", + input_onnx=str(onnx_path), + timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), + elapsed_seconds=round(elapsed, 3), + final_artifact=final_path.name, + stages=manifest_stages, + analyze_iterations=analyze_iters, + analyze_unsupported_node_count=analyze_unsupported, + analyze_details=analyze_details, + ) + manifest.save(manifest_path) return BuildResult( output_dir=output_dir, diff --git a/src/winml/modelkit/inference/engine.py b/src/winml/modelkit/inference/engine.py index 9a9f41197..c38ad9c02 100644 --- a/src/winml/modelkit/inference/engine.py +++ b/src/winml/modelkit/inference/engine.py @@ -36,6 +36,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, cast +from ..utils.manifest import MANIFEST_FILENAME, WinMLManifest from .tasks import BINARY_TYPES, TASK_REGISTRY, InputField, PipelineMapping from .types import Prediction, PredictionResult @@ -150,7 +151,7 @@ def _find_build_artifacts(build_dir: Path, *, task: str | None = None) -> tuple[ """ # Try plain layout first (bare build output) plain_onnx = build_dir / "model.onnx" - plain_manifest = build_dir / "build_manifest.json" + plain_manifest = build_dir / MANIFEST_FILENAME if plain_onnx.exists(): manifest = json.loads(plain_manifest.read_text()) if plain_manifest.exists() else None if task is None or manifest is None or manifest.get("task") == task: @@ -161,7 +162,7 @@ def _find_build_artifacts(build_dir: Path, *, task: str | None = None) -> tuple[ candidates: list[tuple[Path, dict | None]] = [] for onnx_path in sorted(build_dir.glob("*_model.onnx")): prefix = onnx_path.name.rsplit("_model.onnx", 1)[0] - manifest_path = build_dir / f"{prefix}_build_manifest.json" + manifest_path = build_dir / f"{prefix}_{MANIFEST_FILENAME}" manifest = json.loads(manifest_path.read_text()) if manifest_path.exists() else None if task is not None: if manifest is None or manifest.get("task") == task: @@ -977,11 +978,9 @@ def _load_from_build_dir( @staticmethod def _resolve_model_id_from_dir(build_dir: Path) -> str | None: """Extract model_id from any manifest in the directory (task-agnostic).""" - for manifest_path in build_dir.glob("*build_manifest.json"): - manifest: dict[str, Any] = json.loads(manifest_path.read_text()) - model_id = manifest.get("model_id") - if model_id: - return str(model_id) + for m in WinMLManifest.find(build_dir): + if m.model_id: + return m.model_id return None def _load_from_onnx( diff --git a/src/winml/modelkit/inspect/resolver.py b/src/winml/modelkit/inspect/resolver.py index c5a6e2c01..36f47cdd7 100644 --- a/src/winml/modelkit/inspect/resolver.py +++ b/src/winml/modelkit/inspect/resolver.py @@ -493,6 +493,7 @@ def resolve_cache(model_id: str) -> CacheInfo: import json from ..cache import get_cache_dir, get_model_dir + from ..utils.manifest import MANIFEST_FILENAME, WinMLManifest cache_dir = get_cache_dir() model_dir = get_model_dir(model_id, cache_dir=cache_dir) @@ -508,18 +509,20 @@ def resolve_cache(model_id: str) -> CacheInfo: # PRIMARY: Manifest-based resolution # ------------------------------------------------------------------------- if model_dir.exists(): - manifests = list(model_dir.glob("*build_manifest.json")) - if manifests: - # Use the most recent manifest (by mtime) when multiple variants exist - manifest_path = max(manifests, key=lambda p: p.stat().st_mtime) + manifest_paths = sorted( + model_dir.glob(f"*{MANIFEST_FILENAME}"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + if manifest_paths: try: - manifest = json.loads(manifest_path.read_text()) - manifest_stages = {s["name"]: s for s in manifest.get("stages", [])} + manifest = WinMLManifest.load(manifest_paths[0]) + manifest_stages = {s.name: s for s in manifest.stages} for stage in pipeline_stages: ms = manifest_stages.get(stage) - if ms and ms.get("status") == "completed": - filename = ms.get("filename") + if ms and ms.status == "completed": + filename = ms.filename artifact = model_dir / filename if filename else None size_bytes = ( artifact.stat().st_size if artifact and artifact.exists() else 0 @@ -543,8 +546,8 @@ def resolve_cache(model_id: str) -> CacheInfo: total_cached=total_cached, total_size_mb=round(total_size_mb, 2), ) - except (json.JSONDecodeError, KeyError, OSError) as exc: - logger.debug("Failed to read manifest %s: %s", manifest_path, exc) + except (json.JSONDecodeError, KeyError, TypeError, ValueError, OSError) as exc: + logger.debug("Failed to read manifest %s: %s", manifest_paths[0], exc) # Fall through to filename scanning stages = [] total_cached = 0 diff --git a/src/winml/modelkit/serve/app.py b/src/winml/modelkit/serve/app.py index a666c237f..509fc395c 100644 --- a/src/winml/modelkit/serve/app.py +++ b/src/winml/modelkit/serve/app.py @@ -741,11 +741,13 @@ async def cli_command(command: str, request: CliRequest) -> CliResponse: def _manifest_from_engine(engine: InferenceEngine) -> dict[str, Any]: """Build manifest dict from engine, trying build_manifest.json first.""" if engine.model_path: - manifest_file = Path(engine.model_path) / "build_manifest.json" + from ..utils.manifest import MANIFEST_FILENAME, WinMLManifest + + manifest_file = Path(engine.model_path) / MANIFEST_FILENAME if manifest_file.exists(): try: - return cast("dict[str, Any]", json.loads(manifest_file.read_text())) - except (json.JSONDecodeError, OSError) as e: + return WinMLManifest.load(manifest_file).to_dict() + except (json.JSONDecodeError, TypeError, ValueError, OSError) as e: logger.warning("Failed to load manifest: %s", e) return { diff --git a/src/winml/modelkit/utils/__init__.py b/src/winml/modelkit/utils/__init__.py index 90332584c..10a2c420a 100644 --- a/src/winml/modelkit/utils/__init__.py +++ b/src/winml/modelkit/utils/__init__.py @@ -12,6 +12,7 @@ load_hf_components_from_onnx, save_local_model_configs, ) +from .manifest import MANIFEST_FILENAME, ManifestStage, WinMLManifest from .model_input import ( ModelInput, ModelInputKind, @@ -21,8 +22,11 @@ __all__ = [ + "MANIFEST_FILENAME", + "ManifestStage", "ModelInput", "ModelInputKind", + "WinMLManifest", "classify_model_input", "inject_hub_metadata", "is_hub_model", diff --git a/src/winml/modelkit/utils/manifest.py b/src/winml/modelkit/utils/manifest.py new file mode 100644 index 000000000..c4dc4b475 --- /dev/null +++ b/src/winml/modelkit/utils/manifest.py @@ -0,0 +1,205 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +# -------------------------------------------------------------------------- +"""WinML manifest: machine-readable provenance for build and export outputs. + +``build_manifest.json`` sits alongside every ONNX artifact produced by +``winml build`` or the Python-API equivalents. It records *what* was +built, *how* (which pipeline stages ran), and *when*, so downstream +tools (``winml inspect``, ``winml serve``, the inference engine) can +discover model metadata without re-running the pipeline. + +The :class:`WinMLManifest` dataclass is the single source of truth for the +manifest schema. All producers **must** construct a ``WinMLManifest`` +instance and call :meth:`save`; all consumers **should** use +:meth:`load` / :meth:`find` instead of hand-parsing the JSON. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import asdict, dataclass, field +from typing import TYPE_CHECKING, Any + +import numpy as np + + +if TYPE_CHECKING: + from pathlib import Path + + +logger = logging.getLogger(__name__) + +MANIFEST_FILENAME = "build_manifest.json" +"""Bare manifest filename (no cache-key prefix).""" + +SCHEMA_VERSION = 1 + + +def _sanitize_value(value: Any) -> Any: + """Coerce non-JSON-native types to JSON-safe primitives. + + Handles ``Path`` (→ ``str``) and numpy scalars (→ native Python + numbers/bools) so that numeric metrics are never accidentally + serialised as strings by the ``default=str`` fallback. + """ + from pathlib import PurePath + + if isinstance(value, PurePath): + return str(value) + if isinstance(value, dict): + return {k: _sanitize_value(v) for k, v in value.items()} + if isinstance(value, list): + return [_sanitize_value(v) for v in value] + if isinstance(value, np.generic): + return value.item() + return value + + +def _sanitize_stage_dict(stage: dict[str, Any]) -> dict[str, Any]: + """Compact a stage dict: drop ``None``, merge ``extras``, sanitize values.""" + extras = stage.pop("extras", {}) + d = {k: _sanitize_value(v) for k, v in stage.items() if v is not None} + d.update({k: _sanitize_value(v) for k, v in extras.items()}) + return d + + +@dataclass +class ManifestStage: + """One pipeline stage entry inside the manifest.""" + + name: str + status: str # "completed" | "skipped" + filename: str | None = None + elapsed_seconds: float | None = None + + # Optional quantize metrics (populated only for the "quantize" stage). + nodes_quantized: int | None = None + nodes_skipped: int | None = None + calibration_time_seconds: float | None = None + qdq_insertion_time_seconds: float | None = None + + extras: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class WinMLManifest: + """Machine-readable provenance for a single ONNX artifact. + + Attributes: + source: How the artifact was produced (``"hf"``, ``"onnx"``, + ``"export"``). + final_artifact: Filename of the deployment-ready ONNX + (e.g. ``"model.onnx"``). + stages: Ordered list of pipeline stages that ran (or were skipped). + model_id: HuggingFace model ID or user-supplied label. + task: Pipeline task (e.g. ``"image-classification"``). + cache_key: Build-cache key (build pipelines only). + config_hash: Trailing hash portion of *cache_key*. + input_onnx: Path to the source ONNX file (``source="onnx"`` only). + elapsed_seconds: Total wall-clock time in seconds. + timestamp: ISO-8601 UTC timestamp of manifest creation. + analyze_iterations: Number of optimize-analyze loop iterations. + analyze_unsupported_node_count: Unsupported-node count from the + final analysis pass. + analyze_details: Free-form analysis metadata. + export_stats: Statistics dict from HTPExporter (``source="export"`` + only). + extras: Catch-all for forward-compatible fields that this version + of the code does not model explicitly. + """ + + source: str | None = None + final_artifact: str | None = None + stages: list[ManifestStage] = field(default_factory=list) + schema_version: int = SCHEMA_VERSION + + model_id: str | None = None + task: str | None = None + cache_key: str | None = None + config_hash: str | None = None + input_onnx: str | None = None + elapsed_seconds: float | None = None + timestamp: str | None = None + + # Analyze loop metadata + analyze_iterations: int | None = None + analyze_unsupported_node_count: int | None = None + analyze_details: dict[str, Any] | None = None + + # Export-only + export_stats: dict[str, Any] | None = None + + extras: dict[str, Any] = field(default_factory=dict) + + # ------------------------------------------------------------------ + # Serialisation helpers + # ------------------------------------------------------------------ + + def to_dict(self) -> dict[str, Any]: + """Serialise to a JSON-safe dict (``schema_version`` injected).""" + raw = asdict(self) + extras = raw.pop("extras", {}) + # schema_version is placed first for readability; remaining fields follow. + d: dict[str, Any] = {"schema_version": raw.pop("schema_version")} + # Drop None values for a compact JSON representation. + d.update({k: _sanitize_value(v) for k, v in raw.items() if v is not None}) + # Stage entries: also drop None fields inside each stage, merge extras. + if "stages" in d: + d["stages"] = [_sanitize_stage_dict(s) for s in d["stages"]] + d.update({k: _sanitize_value(v) for k, v in extras.items()}) + return d + + def save(self, path: Path) -> None: + """Write the manifest as indented JSON to *path*.""" + path.write_text(json.dumps(self.to_dict(), indent=2, default=str)) + logger.debug("Manifest persisted: %s", path) + + # ------------------------------------------------------------------ + # Factory helpers + # ------------------------------------------------------------------ + + @classmethod + def load(cls, path: Path) -> WinMLManifest: + """Deserialise from a JSON file on disk.""" + return cls.from_dict(json.loads(path.read_text())) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> WinMLManifest: + """Construct from a raw JSON dict (as stored on disk).""" + data = dict(data) # shallow copy — don't mutate caller's dict + + stages_raw = data.pop("stages", []) + stage_known = set(ManifestStage.__dataclass_fields__) - {"extras"} + stages = [] + for s in stages_raw: + s = dict(s) # shallow copy + known_stage = {k: s.pop(k) for k in list(s) if k in stage_known} + stages.append(ManifestStage(**known_stage, extras=s)) + + known = set(cls.__dataclass_fields__) - {"stages", "extras"} + known_kwargs = {k: data.pop(k) for k in list(data) if k in known} + extras = data # anything left is forward-compat extras + + return cls(stages=stages, extras=extras, **known_kwargs) + + @classmethod + def find(cls, directory: Path) -> list[WinMLManifest]: + """Discover and load all ``*build_manifest.json`` in *directory*.""" + results: list[WinMLManifest] = [] + for p in sorted(directory.glob(f"*{MANIFEST_FILENAME}")): + m = cls._try_load(p) + if m is not None: + results.append(m) + return results + + @classmethod + def _try_load(cls, path: Path) -> WinMLManifest | None: + """Load a manifest, returning ``None`` on read/parse failure.""" + try: + return cls.load(path) + except (json.JSONDecodeError, TypeError, ValueError, OSError) as exc: + logger.warning("Skipping unreadable manifest %s: %s", path, exc) + return None diff --git a/tests/unit/utils/test_manifest.py b/tests/unit/utils/test_manifest.py new file mode 100644 index 000000000..fc5d443c0 --- /dev/null +++ b/tests/unit/utils/test_manifest.py @@ -0,0 +1,219 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. +# -------------------------------------------------------------------------- +"""Tests for WinMLManifest dataclass (utils/manifest.py).""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pathlib import Path + +import pytest + +from winml.modelkit.utils.manifest import ( + MANIFEST_FILENAME, + ManifestStage, + WinMLManifest, +) + + +class TestManifestStage: + """ManifestStage construction and serialisation.""" + + def test_minimal(self) -> None: + s = ManifestStage(name="export", status="completed") + assert s.name == "export" + assert s.filename is None + + def test_with_quant_metrics(self) -> None: + s = ManifestStage( + name="quantize", + status="completed", + filename="quantized.onnx", + elapsed_seconds=1.5, + nodes_quantized=42, + nodes_skipped=3, + ) + assert s.nodes_quantized == 42 + assert s.nodes_skipped == 3 + + +class TestWinMLManifestRoundTrip: + """Serialise → deserialise round-trip.""" + + def test_to_dict_injects_schema_version(self) -> None: + m = WinMLManifest(source="hf", final_artifact="model.onnx") + d = m.to_dict() + assert d["schema_version"] == 1 + assert d["source"] == "hf" + + def test_none_fields_omitted(self) -> None: + m = WinMLManifest(source="export", final_artifact="model.onnx") + d = m.to_dict() + assert "model_id" not in d + assert "cache_key" not in d + + def test_stage_none_fields_omitted(self) -> None: + m = WinMLManifest( + source="hf", + final_artifact="model.onnx", + stages=[ManifestStage(name="optimize", status="skipped")], + ) + d = m.to_dict() + stage = d["stages"][0] + assert "filename" not in stage + assert "elapsed_seconds" not in stage + + def test_round_trip(self) -> None: + original = WinMLManifest( + source="hf", + model_id="microsoft/resnet-50", + task="image-classification", + final_artifact="model.onnx", + elapsed_seconds=12.345, + stages=[ + ManifestStage( + name="export", + status="completed", + filename="export.onnx", + elapsed_seconds=5.0, + ), + ManifestStage(name="quantize", status="skipped"), + ], + ) + d = original.to_dict() + restored = WinMLManifest.from_dict(d) + assert restored.source == original.source + assert restored.model_id == original.model_id + assert restored.task == original.task + assert restored.elapsed_seconds == original.elapsed_seconds + assert len(restored.stages) == 2 + assert restored.stages[0].name == "export" + assert restored.stages[0].filename == "export.onnx" + assert restored.stages[1].status == "skipped" + + def test_forward_compat_extras(self) -> None: + """Unknown keys in JSON land in ``extras`` instead of crashing.""" + d = { + "schema_version": 1, + "source": "hf", + "final_artifact": "model.onnx", + "future_field": "hello", + } + m = WinMLManifest.from_dict(d) + assert m.extras["future_field"] == "hello" + + def test_forward_compat_stage_extras(self) -> None: + """Unknown keys inside a stage land in stage ``extras``.""" + d = { + "schema_version": 1, + "source": "hf", + "final_artifact": "model.onnx", + "stages": [ + { + "name": "export", + "status": "completed", + "future_metric": 42, + } + ], + } + m = WinMLManifest.from_dict(d) + assert m.stages[0].extras["future_metric"] == 42 + # Round-trip preserves the extra + rt = WinMLManifest.from_dict(m.to_dict()) + assert rt.stages[0].extras["future_metric"] == 42 + + +class TestWinMLManifestIO: + """File I/O helpers.""" + + def test_save_and_load(self, tmp_path: Path) -> None: + m = WinMLManifest( + source="onnx", + final_artifact="model.onnx", + input_onnx="input.onnx", + stages=[ + ManifestStage( + name="optimize", + status="completed", + filename="optimized.onnx", + elapsed_seconds=2.0, + ) + ], + ) + path = tmp_path / MANIFEST_FILENAME + m.save(path) + assert path.exists() + + loaded = WinMLManifest.load(path) + assert loaded.source == "onnx" + assert loaded.input_onnx == "input.onnx" + assert len(loaded.stages) == 1 + + def test_save_sanitizes_path_objects(self, tmp_path: Path) -> None: + """Path values in export_stats are coerced to strings, not dropped.""" + m = WinMLManifest( + source="export", + final_artifact="model.onnx", + export_stats={"output_path": tmp_path / "model.onnx", "count": 5}, + ) + path = tmp_path / MANIFEST_FILENAME + m.save(path) + data = json.loads(path.read_text()) + assert isinstance(data["export_stats"]["output_path"], str) + assert data["export_stats"]["count"] == 5 # numeric stays numeric + + def test_save_sanitizes_numpy_scalars(self, tmp_path: Path) -> None: + """Numpy scalars in export_stats are coerced to native Python types.""" + import numpy as np + + m = WinMLManifest( + source="export", + final_artifact="model.onnx", + export_stats={ + "accuracy": np.float32(0.95), + "count": np.int64(42), + "flag": np.bool_(True), + }, + ) + path = tmp_path / MANIFEST_FILENAME + m.save(path) + data = json.loads(path.read_text()) + assert data["export_stats"]["accuracy"] == pytest.approx(0.95, abs=1e-5) + assert data["export_stats"]["count"] == 42 + assert isinstance(data["export_stats"]["count"], int) + assert data["export_stats"]["flag"] is True + + def test_find_discovers_multiple(self, tmp_path: Path) -> None: + for name in ["build_manifest.json", "feat_aaa_build_manifest.json"]: + (tmp_path / name).write_text( + json.dumps({"schema_version": 1, "source": "hf", "final_artifact": "model.onnx"}) + ) + found = WinMLManifest.find(tmp_path) + assert len(found) == 2 + + def test_find_skips_corrupt(self, tmp_path: Path) -> None: + (tmp_path / MANIFEST_FILENAME).write_text("not json{{{") + found = WinMLManifest.find(tmp_path) + assert found == [] + + def test_json_on_disk_matches_schema(self, tmp_path: Path) -> None: + """Verify the on-disk JSON has schema_version and is valid.""" + m = WinMLManifest( + source="export", + model_id="test/model", + task="image-classification", + final_artifact="model.onnx", + export_stats={"tagged_nodes": 10}, + ) + path = tmp_path / MANIFEST_FILENAME + m.save(path) + raw = json.loads(path.read_text()) + assert raw["schema_version"] == 1 + assert raw["source"] == "export" + assert raw["export_stats"]["tagged_nodes"] == 10