File size: 2,756 Bytes
f5498f9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """Artifact provenance stamping and the generator/artifact registry."""
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Tuple
from .paths import REPO
@dataclass(frozen=True)
class ArtifactSpec:
"""Owning script, source pool, and required top-level payload keys."""
generator: Optional[str]
pool: Optional[str]
payload_keys: Tuple[str, ...]
note: str = ''
NO_GENERATOR = 'discovery sweep not committed; no producer for this file in the repo'
REGISTRY = {
'eval.json': ArtifactSpec('verify.py', 'VAL5000', ('metrics',)),
'eval_tight_fpr.json': ArtifactSpec(
'verify.py', None,
('metrics', 'prop_false_positive_rate', 'baseline')),
'discovery/dim_selection.json': ArtifactSpec(None, None, (), NO_GENERATOR),
'discovery/dim48_characterization.json': ArtifactSpec(None, None, (), NO_GENERATOR),
'discovery/prop_specificity.json': ArtifactSpec(None, None, (), NO_GENERATOR),
'discovery/prop_image_manifest.json': ArtifactSpec(None, None, (), NO_GENERATOR),
'discovery/variant_leaderboard.json': ArtifactSpec(None, None, (), NO_GENERATOR),
'per_dim_thresholds.json': ArtifactSpec(
'calibrate.py', 'BALANCED_VAL',
('quant_scale', 'per_dim_thresholds', 'popcount', 'additive',
'F1_delta_popcount_vs_additive')),
'circuit.json': ArtifactSpec('synth.py', None, ('variants', 'accuracy')),
}
def sha256_of(path) -> str:
"""Content hash of a file."""
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def provenance(generator: str, classifier=None, pool_info: Optional[dict] = None,
**extra) -> dict:
"""Assemble a provenance block from repository-recoverable fields only."""
block = {'generator': generator}
if classifier is not None:
path = Path(classifier)
block['classifier'] = str(path.resolve().relative_to(REPO)).replace('\\', '/')
block['classifier_sha256'] = sha256_of(path)
if pool_info:
block.update(pool_info)
block.update(extra)
return block
def write_artifact(path, payload: dict, *, generator: str, classifier=None,
pool_info: Optional[dict] = None, compact: bool = False, **extra):
"""Write `payload` beneath a provenance block and return the document."""
doc = {'provenance': provenance(generator, classifier, pool_info, **extra)}
doc.update(payload)
sep = (',', ':') if compact else None
text = json.dumps(doc, indent=None if compact else 2, separators=sep)
Path(path).write_text(text + ('' if compact else '\n'), encoding='utf-8')
return doc
def read_artifact(path) -> dict:
return json.loads(Path(path).read_text(encoding='utf-8'))
|