Digital Drosophila Archive
64 lines
3.5 KiB
Python
64 lines
3.5 KiB
Python
"""Post-seal descriptive interpretation; no trajectory rerun or gate change."""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
ROOT = Path('/data/digital-drosophila')
|
|
MODEL = ROOT / 'models/male-cns-lif/pn-recruitment-audit-v1'
|
|
EXP = ROOT / 'experiments/phase4ai-pn-recruitment-audit'
|
|
REGISTRY = ROOT / 'registry/phase4ai-pn-recruitment-audit'
|
|
sys.path.insert(0, str(MODEL))
|
|
from context import read, save, digest, GROUPS
|
|
from publication import verify_seal
|
|
|
|
|
|
def main():
|
|
verified = {'model': verify_seal(MODEL), 'experiment': verify_seal(EXP)}
|
|
status = read(EXP / 'audit/final-status.json')
|
|
structure = read(EXP / 'audit/PN-odor-structure.json')
|
|
output = {}
|
|
for mode, aggregate in status['modes'].items():
|
|
rows = [r for r in read(EXP / 'protocol/inventory.json') if r['repeat'] == 0 and r['kind'] == 'calibration'
|
|
and read(EXP / 'trials' / (r['case'] + '-r0') / 'summary.json')['mode'] == mode]
|
|
assert len(rows) == 30
|
|
pn = pd.concat([pd.read_parquet(EXP / 'trials' / (r['case'] + '-r0') / 'per-pn.parquet') for r in rows], ignore_index=True)
|
|
chosen = pn[(~pn.is_design_target) & pn.first_spike_observed]
|
|
positives = {g: float(chosen[g + '_accepted_positive_before_first_V'].sum()) for g in GROUPS}
|
|
total = sum(positives.values())
|
|
source = {g: dict(pre_first_accepted_positive_V=positives[g],
|
|
share=None if total == 0 else positives[g] / total,
|
|
pre_first_accepted_negative_abs_V=float(chosen[g + '_accepted_negative_before_first_abs_V'].sum()),
|
|
mean_integrated_positive_V=float(chosen['first_integrated_V_from_' + g + '/positive_weight'].mean()),
|
|
mean_integrated_negative_V=float(chosen['first_integrated_V_from_' + g + '/negative_weight'].mean())) for g in GROUPS}
|
|
view_metrics = {}
|
|
for name in ('fixed_686', 'common_non_design'):
|
|
view = structure['modes'][mode][name]
|
|
view_metrics[name] = dict(dimensions=view['dimensions'],
|
|
metrics={m: dict(within=v['mean'], between=view['between'][m]['mean'],
|
|
pair_mean_between_greater_than_within=sum(p['mean_between_minus_within'][m] is not None and
|
|
p['mean_between_minus_within'][m] > 0 for p in view['pairs'])) for m, v in view['within'].items()})
|
|
output[mode] = dict(frozen_aggregate=aggregate,
|
|
active_non_design_PN_neuron_trials=len(chosen), unique_active_non_design_PN=int(chosen.neuron_idx.nunique()),
|
|
active_non_design_PN_prefirst_source_accounting=source,
|
|
source_shares_are_nominal_positive_increment_sums_not_necessity_or_instantaneous_current=True,
|
|
odor_structure_descriptive=view_metrics)
|
|
result = dict(status=status['status'], modes=output,
|
|
exact_PN_trials=status['exact_PN_trials'], repeat_accounting_pass=status['repeat_accounting_pass'],
|
|
source_script_sha256=digest(Path(__file__)),
|
|
model_manifest_sha256=verified['model']['manifest_sha256'],
|
|
experiment_manifest_sha256=verified['experiment']['manifest_sha256'],
|
|
descriptive_postseal_interpretation=True, new_scientific_gate=False, learning_achieved=False,
|
|
original_CAL_failure_unchanged=True)
|
|
path = REGISTRY / 'interpretation-metrics.json'
|
|
save(path, result)
|
|
with (REGISTRY / 'interpretation-metrics.sha256').open('x') as f:
|
|
f.write(digest(path) + ' interpretation-metrics.json\n')
|
|
print(json.dumps(result, ensure_ascii=False, allow_nan=False), flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|