| """Read-only training watcher + final English report synthesis; no GPU or tuning. |
| |
| Waits for the actual study worker to finish. Never promotes a model, changes |
| training, reads microphone data, deletes caches, or treats partial runs as final. |
| """ |
| import argparse,json,os,shutil,time |
| from pathlib import Path |
| import numpy as np |
| from .common import ROOT,WORK,FEATURES,atomic,digest,report,STATE |
|
|
| def summarize(): |
| from .train import aggregates,objective |
| results=json.loads((WORK/'results.json').read_text()) |
| if not results.get('finished'):raise RuntimeError('Training not complete; refusing final claims') |
| for stage in ('stage1','factorial'):results[stage+'_aggregates']=aggregates(results[stage]) |
| conclusions=[];baseline=results['original_frozen_320ms_control']['test'] |
| control=results['original_frozen_320ms_control'] |
| candidates=[(objective(control['validation']),'original_frozen','query_gru_legacy',[control])] |
| for stage in ('stage1','factorial'): |
| for arm in results[stage+'_aggregates']: |
| rows=[r for r in results[stage] if r['arm']==arm] |
| candidates.append((float(np.mean([objective(r['validation']) for r in rows])),stage,arm,rows)) |
| _,stage,best,rows=max(candidates,key=lambda x:x[0]) |
| results['best_overall_by_validation']={'stage':stage,'arm':best,'note':'Operational selection, not a causal full-data/subset comparison'} |
| test_mean=float(np.mean([r['test']['macro_speaker_accuracy'] for r in rows])) |
| conclusions.append(f"Original frozen 320-ms control: {100*baseline['macro_speaker_accuracy']:.2f}% test macro identification. Best overall option by validation, including that original control: {best} ({stage}), {100*test_mean:.2f}% test macro identification, a descriptive change of {100*(test_mean-baseline['macro_speaker_accuracy']):+.2f} percentage points. New arms average three seeds; the original control is one frozen checkpoint. Test is among ten previously benchmarked synthetic identities.") |
| first=results['stage1_aggregates'];special=first['specialize_only']['validation'] |
| for arm in sorted(a for a in first if a.startswith('stable')): |
| other=first[arm]['validation'] |
| conclusions.append(f"{arm} versus specialization alone (validation means): macro accuracy {100*(other['macro_speaker_accuracy']['mean']-special['macro_speaker_accuracy']['mean']):+.2f} pp; genuine enrollment cosine {other['positive_cosine']['mean']-special['positive_cosine']['mean']:+.4f}; impostor cosine {other['impostor_cosine']['mean']-special['impostor_cosine']['mean']:+.4f}; EER {100*(other['eer']['mean']-special['eer']['mean']):+.2f} pp. A higher genuine cosine is not automatically better discrimination.") |
| base=[r for r in results['factorial'] if r['arm']=='base12'];paired={} |
| for arm in sorted({r['arm'] for r in results['factorial']} - {'base12'}): |
| treatment=[r for r in results['factorial'] if r['arm']==arm];paired[arm]={} |
| for split in ('validation','test'): |
| speakers=sorted(base[0][split]['per_speaker']);differences=[] |
| for speaker in speakers: |
| a=np.mean([r[split]['per_speaker'][speaker]['accuracy'] for r in treatment]) |
| b=np.mean([r[split]['per_speaker'][speaker]['accuracy'] for r in base]);differences.append(a-b) |
| draws=np.random.default_rng(943).choice(differences,(5000,len(differences))).mean(1) |
| paired[arm][split]={'macro_accuracy_delta_pp':float(np.mean(differences)*100), |
| 'paired_speaker_bootstrap_95ci_pp':[float(v*100) for v in np.quantile(draws,[.025,.975])], |
| 'identities':len(speakers),'seeds_averaged_before_bootstrap':len(base)} |
| point=paired[arm]['test'];lo,hi=point['paired_speaker_bootstrap_95ci_pp'] |
| conclusions.append(f"Matched factorial {arm} versus base12: test macro accuracy {point['macro_accuracy_delta_pp']:+.2f} pp; paired identity bootstrap interval [{lo:+.2f}, {hi:+.2f}] pp. This interval uses only ten synthetic identities; it is not evidence of real-human reliability.") |
| results['paired_factorial_comparisons']=paired |
| cleaned=json.loads((WORK/'cleanup_manifest.json').read_text()) |
| resources={'surplus_downloads_removed':len(cleaned['files']),'bytes_removed':sum(r['bytes'] for r in cleaned['files']), |
| 'cleanup_manifest_sha256':digest(WORK/'cleanup_manifest.json'),'feature_cache_files':{},'checkpoints':{}, |
| 'working_volume_free_bytes':shutil.disk_usage(WORK).free,'nvme_volume_free_bytes':shutil.disk_usage(ROOT.parent).free, |
| 'feature_cache_volume_free_bytes':shutil.disk_usage(FEATURES).free,'feature_cache_location':str(FEATURES.resolve()), |
| 'feature_cache_is_ephemeral':FEATURES.resolve().is_relative_to(Path('/dev/shm'))} |
| if (WORK/'cache_migration.json').exists():resources['cache_migration']=json.loads((WORK/'cache_migration.json').read_text()) |
| for path in FEATURES.glob('*.npy'):resources['feature_cache_files'][path.name]={'bytes':path.stat().st_size,'sha256':digest(path)} |
| for path in (WORK/'models').glob('*-best.pt'):resources['checkpoints'][path.name]={'bytes':path.stat().st_size,'sha256':digest(path)} |
| results['resources_and_integrity']=resources |
| conclusions.append('No model was automatically enabled as the live user filter. Pooling spans of 320 ms already include past encoder context. Cold-start real-user verification and secure authentication remain untested.') |
| results['conclusions']=conclusions;STATE.update(json.loads((ROOT/'progress.json').read_text())) |
| report(results,True);atomic(WORK/'finalized.json',{'completed':time.time(),'best_group':best,'stage':stage}) |
|
|
| def main(wait=False): |
| if wait: |
| atomic(ROOT/'report_watcher.json',{'pid':os.getpid(),'started':time.time()}) |
| dead_checks=0 |
| while True: |
| path=WORK/'results.json' |
| if path.exists() and json.loads(path.read_text()).get('finished'):break |
| status=json.loads((ROOT/'progress.json').read_text()) if (ROOT/'progress.json').exists() else {} |
| if status.get('error'): |
| STATE.update(status);report();return |
| record=json.loads((ROOT/'runner.json').read_text()) if (ROOT/'runner.json').exists() else {} |
| worker=Path('/proc',str(record.get('pid',0)),'cmdline') |
| if not worker.exists(): |
| dead_checks+=1 |
| if dead_checks>=6:return |
| time.sleep(5);continue |
| dead_checks=0 |
| time.sleep(20) |
| summarize() |
|
|
| if __name__=='__main__': |
| parser=argparse.ArgumentParser();parser.add_argument('--wait',action='store_true');main(parser.parse_args().wait) |
|
|
