| """Derive atlas assembly-presence coverage directly from a full accession index.""" |
| import argparse |
| from collections import Counter |
| from datetime import datetime,timezone |
| import json |
| from pathlib import Path |
| import sqlite3 |
|
|
| if __name__=='__main__': |
| p=argparse.ArgumentParser(description=__doc__) |
| p.add_argument('--index',type=Path,required=True) |
| p.add_argument('--output',type=Path,required=True) |
| args=p.parse_args() |
| with sqlite3.connect(args.index) as c: |
| m=json.loads(c.execute("SELECT value FROM metadata WHERE key='manifest'").fetchone()[0]) |
| if not m.get('full_snapshot') or m.get('schema_version')!=3: raise ValueError('A complete compact index is required') |
| accessions=[r[0] for r in c.execute('SELECT DISTINCT c.assembly_accession FROM contexts c WHERE EXISTS (SELECT 1 FROM segment_data s WHERE s.context_id=c.id) ORDER BY c.assembly_accession')] |
| shards=Counter('/'.join(r[0].split('/')[1:3]) for r in c.execute('SELECT path FROM files')) |
| fetched=c.execute('SELECT sum(bytes_read) FROM files').fetchone()[0] |
| inventory={'created_at':datetime.now(timezone.utc).isoformat(),'listing_observed_at':m['inventory_at'], |
| 'annotation_bucket':m['bucket_id'],'scope':'Assemblies with published annotations; includes partial assemblies. Exact accession versions; no base completeness claim.', |
| 'method':'Distinct assembly IDs read directly from scalar metadata of every indexed published annotation Parquet object. No completion-marker or packed-input inference.', |
| 'annotation_object_count':m['source_count'],'shard_count':len(shards),'metadata_bytes_read':fetched, |
| 'inventory_fingerprint':m['inventory_sha256'],'accessions':accessions, |
| 'shards':[{'shard':s,'annotation_objects':n,'method':'annotation_metadata'} for s,n in sorted(shards.items())]} |
| temp=args.output.with_suffix('.json.tmp'); temp.write_text(json.dumps(inventory)); temp.replace(args.output) |
| print('Direct annotation assembly versions:',len(accessions),flush=True) |
|
|
