Digital Drosophila Archive
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
"""Phase 4AI: read-only PN/APL accounting of sealed Phase 4AG-r2 records."""
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path('/data/digital-drosophila')
|
|
MODEL = Path(__file__).resolve().parent
|
|
EXP = ROOT / 'experiments/phase4ai-pn-recruitment-audit'
|
|
AH_MODEL = ROOT / 'models/male-cns-lif/kc-drive-audit-v1'
|
|
AH = ROOT / 'experiments/phase4ah-kc-drive-audit'
|
|
AG_MODEL = ROOT / 'models/male-cns-lif/author-apl-lab-v2'
|
|
AG = ROOT / 'experiments/phase4ag-author-apl-r2'
|
|
META = ROOT / 'datasets/male-cns-v1.0/processed/neuron-only/neuron-index/unified-metadata.parquet'
|
|
COEFF = ROOT / 'experiments/phase4af-author-fullgraph/oracle/no_input/coefficients.json'
|
|
GROUPS = ('KC', 'APL', 'ALPN', 'MBON', 'anatomical_DAN', 'dopamine_annotation_only',
|
|
'ALLN', 'ALIN', 'ALON', 'olfactory_sensory', 'other', 'unclassified')
|
|
|
|
|
|
def load_module(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def digest(path):
|
|
h = hashlib.sha256()
|
|
with Path(path).open('rb') as f:
|
|
for b in iter(lambda: f.read(8 << 20), b''):
|
|
h.update(b)
|
|
return h.hexdigest()
|
|
|
|
|
|
def read(path):
|
|
return json.loads(Path(path).read_text(encoding='utf-8'))
|
|
|
|
|
|
def save(path, value):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open('x', encoding='utf-8') as f:
|
|
json.dump(value, f, indent=2, ensure_ascii=False, allow_nan=False)
|
|
f.write('\n')
|
|
|
|
|
|
def sources():
|
|
return {p.name: digest(p) for p in sorted(MODEL.iterdir()) if p.is_file()
|
|
and p.name not in ('manifest.json', 'checksums.sha256')}
|
|
|
|
|
|
def integrity():
|
|
prior = load_module('sealed_ah_context_ai', AH_MODEL / 'context.py')
|
|
result = prior.integrity()
|
|
sys.path.insert(0, str(ROOT / 'models/male-cns-lif/terminal-scope-diagnostic-v1'))
|
|
from diagnostic import manifest_check
|
|
for path in (AH_MODEL, AH):
|
|
result[str(path.relative_to(ROOT))] = manifest_check(path, digest)
|
|
assert all(row['pass'] for row in result.values())
|
|
return result
|
|
|
|
|
|
def check():
|
|
record = read(EXP / 'protocol/preregistered.json')
|
|
assert record['sources'] == sources(), 'Registered source changed'
|
|
for row in record['inputs']:
|
|
assert digest(ROOT / row['path']) == row['sha256'], row['path']
|
|
for row in read(EXP / 'protocol/prepared-hashes.json'):
|
|
assert digest(EXP / row['path']) == row['sha256'], row['path']
|
|
return record
|
|
|
|
|
|
def seal(root):
|
|
rows = [dict(path=str(p.relative_to(root)), bytes=p.stat().st_size, sha256=digest(p))
|
|
for p in sorted(root.rglob('*')) if p.is_file() and '__pycache__' not in p.parts]
|
|
save(root / 'manifest.json', dict(root=str(root), files=rows))
|
|
with (root / 'checksums.sha256').open('x') as f:
|
|
for row in rows:
|
|
f.write(row['sha256'] + ' ' + row['path'] + '\n')
|
|
f.write(digest(root / 'manifest.json') + ' manifest.json\n')
|
|
|
|
|
|
def bit_equal(a, b):
|
|
return a.dtype == b.dtype and a.shape == b.shape and a.tobytes() == b.tobytes()
|