38 lines
1.5 KiB
Python
38 lines
1.5 KiB
Python
import hashlib
|
|
import json
|
|
from typing import Dict, Any, List
|
|
|
|
|
|
class ReproBundle:
|
|
"""Compact, content-addressed bundle for reproducible runs.
|
|
|
|
The bundle is a mapping of named components (code_commit, environment,
|
|
datasets, model_refs, run_manifest). We compute a deterministic Merkle-like
|
|
root by hashing each component's canonical JSON (sorted keys) and then
|
|
hashing the concatenation of component hashes in a stable order.
|
|
"""
|
|
|
|
def __init__(self, components: Dict[str, Any]):
|
|
# Normalize components to dict of JSON-serializable structures
|
|
self.components = components
|
|
|
|
def _hash_component(self, name: str, value: Any) -> str:
|
|
serialized = json.dumps({"name": name, "value": value}, sort_keys=True, default=str)
|
|
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
|
|
|
|
def merkle_root(self) -> str:
|
|
# Compute per-component hashes, sort by component name to ensure order-independence
|
|
items: List[tuple] = []
|
|
for k, v in self.components.items():
|
|
h = self._hash_component(k, v)
|
|
items.append((k, h))
|
|
items.sort(key=lambda x: x[0])
|
|
concat = "".join(h for _, h in items).encode("utf-8")
|
|
return hashlib.sha256(concat).hexdigest()
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"components": self.components, "merkle_root": self.merkle_root()}
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict(), sort_keys=True, default=str)
|