DERBACH's picture
download
raw
3.02 kB
import json,subprocess,datetime,hashlib,shutil
from pathlib import Path
import pandas as pd
FIELDS='experiment_id timestamp git_commit dataset split backbone model_revision representation_layer pooling_method latent_architecture seed latent_dim criteria_count residual_enabled n_interventions intervention_filter lambda_target lambda_leak lambda_residual preference_accuracy axis_purity axis_stability subspace_stability target_effect criterion_leakage composition_accuracy composition_kendall_tau composition_spearman'.split()
def commit():
try: return subprocess.check_output(['git','rev-parse','HEAD'],text=True,stderr=subprocess.DEVNULL).strip()
except subprocess.CalledProcessError: return 'uncommitted'
def save_evaluation_provenance(path):
"""Pin the evaluator that produced the current raw evaluation and metrics.
Training snapshots intentionally remain unchanged when evaluation is rerun.
Call before evaluation and again after a final cross-seed stability pass.
"""
path=Path(path);source={}
folders=['eval','models','intervention','representations','data','analysis']
files=[Path('scripts/evaluate_heads.py')]
for folder in folders:files.extend(Path(folder).rglob('*.py'))
for f in files:
content=f.read_bytes();source[str(f)]=hashlib.sha256(content).hexdigest()
dest=path/'evaluation_source_snapshot'/f;dest.parent.mkdir(parents=True,exist_ok=True);dest.write_bytes(content)
record={'timestamp':datetime.datetime.now(datetime.timezone.utc).isoformat(),'git_commit':commit(),'source_sha256':source,'training_source_manifest':'source_manifest.json','purpose':'current evaluation code; separate from original fitting snapshot'}
(path/'evaluation_manifest.json').write_text(json.dumps(record,indent=2)+'\n')
return record
def save_run(path,config,metrics):
path=Path(path); path.mkdir(parents=True,exist_ok=True)
record=dict.fromkeys(FIELDS); record.update(config); record.update(metrics)
record.update(timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(),git_commit=commit())
record['git_dirty']=bool(subprocess.check_output(['git','status','--porcelain'],text=True).strip())
source={}
for folder in ['data','representations','intervention','models','losses','eval','analysis','scripts','configs']:
for f in Path(folder).rglob('*'):
if f.is_file() and f.suffix in ['.py','.yaml']:
source[str(f)]=hashlib.sha256(f.read_bytes()).hexdigest()
dest=path/'source_snapshot'/f;dest.parent.mkdir(parents=True,exist_ok=True);shutil.copyfile(f,dest)
(path/'source_manifest.json').write_text(json.dumps(source,indent=2)+'\n')
(path/'config.json').write_text(json.dumps(config,indent=2)+'\n')
(path/'metrics.json').write_text(json.dumps(record,indent=2)+'\n')
return record
def rebuild():
rows=[json.loads(p.read_text()) for p in Path('outputs').glob('**/metrics.json')]
pd.DataFrame(rows).to_csv('outputs/results.csv',index=False)

Xet Storage Details

Size:
3.02 kB
·
Xet hash:
cda9d3c93044143e4feb41a20f208b90f130688b0bcba06d362aa6ec1f594ecc

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.