fetch_loci.py

"""Read measured gene_count values from published per-assembly metadata."""
from pathlib import Path
from huggingface_hub import HfApi
import json,hashlib,time
R=Path(__file__).parent; CACHE=Path('/home/node/local/carbon-a-strand-metadata');CACHE.mkdir(exist_ok=True)
D=json.loads((R/'data/tree-data.json').read_text());ids={r[0] for r in D['assemblies']}
inv=json.loads((R/'data/cds-inventory.json').read_text());want=[r for r in inv if r['accession'] in ids]
api=HfApi();records={};target=R/'data/locus-counts.json'
if target.exists():records=json.loads(target.read_text())['records']
remaining=[r for r in want if r['accession'] not in records];read_bytes=0;start=time.time()
for k in range(0,len(remaining),500):
 batch=remaining[k:k+500];lookup={r['path']+'/metadata.json':r for r in batch}
 entries=list(api.get_bucket_paths_info('HuggingFaceBio/genbank-annotations',list(lookup)))
 read_bytes+=sum(e.size for e in entries)
 api.download_bucket_files('HuggingFaceBio/genbank-annotations',[(e,str(CACHE/(lookup[e.path]['accession']+'.json'))) for e in entries],raise_on_missing_files=True)
 for e in entries:
  info=lookup[e.path];local=CACHE/(info['accession']+'.json');raw=local.read_bytes();m=json.loads(raw)
  assert m['genbank_accession']==info['accession']
  assert isinstance(m.get('gene_count'),int) and m['gene_count']>=0
  records[info['accession']]={'loci':m['gene_count'],'bases':m.get('annotated_bases'),'path':e.path,'sha256':hashlib.sha256(raw).hexdigest(),'pipeline_revision':m.get('postprocessing',{}).get('revision')}
  local.unlink()
 out={'source_bucket':'HuggingFaceBio/genbank-annotations','field':'gene_count','retrieved_at':'2026-10-09','inventory':'cds-inventory.json from 2026-10-09','records':records,'summary':{'available_assemblies':len(records),'shown_assemblies':len(ids),'missing_counts':len(ids)-len(records),'sum_available_loci':sum(r['loci'] for r in records.values())}}
 target.write_text(json.dumps(out,separators=(',',':')))
 print('counts',len(records),'/',len(want),'downloaded_MB',round(read_bytes/1e6,1),'elapsed_s',round(time.time()-start),flush=True)
print(out['summary'],flush=True)