| |
| """Export lightweight, reproducible views of the paper's frozen million-point Atlas. |
| |
| Requires numpy and Pillow in the research environment. No embeddings are refitted, |
| no remote dataset is downloaded, and full-corpus ID search is independent of these assets. |
| """ |
| from export_atlas_coverage import export_coverage |
|
|
| import csv |
| import gzip |
| import hashlib |
| import json |
| from pathlib import Path |
|
|
| import numpy as np |
| from PIL import Image, ImageChops |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
| SOURCE = ROOT / 'Ouroboros_Atlas/visualization/20261107-atlas-million-r1' |
| LAYOUT = SOURCE / 'research/20261107-category-layout-r1' |
| PANEL = SOURCE / 'research/20261108-prot2text-panel-r5' |
| CASES = SOURCE / 'research/20261107-integrated-showcases-r1' |
| PUBLIC = ROOT / 'Ouroboros_Showcase/public' |
| SEED = 20261110 |
| AA = dict(zip('ALA ARG ASN ASP CYS GLN GLU GLY HIS ILE LEU LYS MET PHE PRO SER THR TRP TYR VAL SEC PYL ASX GLX UNK'.split(), 'ARNDCQEGHILKMFPSTWYVUOBZX')) |
|
|
|
|
| def sha(data): |
| return hashlib.sha256(data).hexdigest() |
|
|
|
|
| def pdb_residues(data): |
| residues, seen = [], set() |
| for line in data.decode().splitlines(): |
| if line.startswith('ENDMDL'): |
| break |
| if not line.startswith('ATOM ') or line[12:16].strip() != 'CA' or line[16] not in ' A': |
| continue |
| key = (line[21], line[22:27]) |
| if key in seen: |
| continue |
| seen.add(key) |
| residues.append(dict(index=int(line[22:26]), chain=line[21].strip(), |
| code=AA[line[17:20].strip()], confidence=float(line[60:66]))) |
| assert residues, 'No CA residues' |
| return residues |
|
|
|
|
| def sample(pool, labels, count, required): |
| """Community-stratified sample; retain small communities and every paper case.""" |
| rng = np.random.default_rng(SEED) |
| chosen = set(required) |
| for label in np.unique(labels[pool]): |
| group = pool[labels[pool] == label] |
| chosen.update(rng.choice(group, min(8, len(group)), replace=False).tolist()) |
| remainder = np.array(sorted(set(pool.tolist()) - chosen)) |
| chosen.update(rng.choice(remainder, count - len(chosen), replace=False).tolist()) |
| |
| return rng.permutation(sorted(chosen)) |
|
|
|
|
| def thumbnail(case): |
| data = Path(case['image_path']).read_bytes() |
| assert sha(data) == case['image_sha256'] |
| im = Image.open(case['image_path']).convert('RGBA') |
| white = Image.new('RGBA', im.size, 'white') |
| white.alpha_composite(im) |
| im = white.convert('RGB') |
| mask = ImageChops.difference(im, Image.new('RGB', im.size, 'white')).convert('L').point(lambda x: 255 if x > 12 else 0) |
| bbox = mask.getbbox() |
| if bbox: |
| im = im.crop(bbox) |
| im.thumbnail((320, 220)) |
| result = Image.new('RGB', (352, 244), 'white') |
| result.paste(im, ((352-im.width)//2, (244-im.height)//2)) |
| name = 'atlas-' + case['accession'] + '.webp' |
| result.save(PUBLIC / 'images' / name, quality=86, method=6) |
| return '/images/' + name |
|
|
|
|
| def main(): |
| export_coverage() |
| cluster = json.loads((PANEL / 'cluster_manifest.json').read_text()) |
| coordinates = LAYOUT / 'coordinates_landmark.npy' |
| labels_path = PANEL / 'cluster_labels.npy' |
| assert sha(labels_path.read_bytes()) == cluster['labels_sha256'] |
| assert sha(coordinates.read_bytes()) == json.loads((LAYOUT/'landmark_map_manifest.json').read_text())['coordinates_sha256'] |
| assert sha((SOURCE/'map_points.csv').read_bytes()) == cluster['inputs']['map_points_sha256'] |
| xy, labels = np.load(coordinates), np.load(labels_path) |
| assert xy.shape == (1000000, 2) and np.isfinite(xy).all() |
| assert np.bincount(labels).tolist() == cluster['cluster_sizes'] |
| selected = json.loads((CASES/'main_cases.json').read_text()) + json.loads((CASES/'supplementary_cases.json').read_text()) |
| assert len(selected) == 8 |
| required = [c['index'] for c in selected] |
| main = sample(np.arange(len(xy)), labels, 12000, required) |
| preview = sample(main, labels, 4000, required) |
| keep = set(main.tolist()) |
| rows = {} |
| with (SOURCE/'map_points.csv').open() as handle: |
| for i, row in enumerate(csv.DictReader(handle)): |
| if i in keep: |
| rows[i] = row |
| assert len(rows) == 12000 |
| (PUBLIC/'images').mkdir(exist_ok=True) |
| output_cases = [] |
| titles = ['C2H2 zinc finger', 'RACK1 scaffold', 'PPE family', 'Lysine transporter', |
| 'C2H2 zinc finger', 'C2H2 zinc finger', 'PPE family', 'WXG100 family'] |
| for number, case in enumerate(selected, 1): |
| row = rows[case['index']] |
| assert row['uniref_id'] == case['uniref_id'] |
| assert np.array_equal(xy[case['index']], case['map_xy']) |
| pdb = Path(case['pdb_path']).read_bytes() |
| assert sha(pdb) == case['pdb_sha256'] |
| residues = pdb_residues(pdb) |
| sequence = ''.join(r['code'] for r in residues) |
| assert sha(sequence.encode()) == case['sequence_sha256'] |
| assert len(sequence) == case['sequence_length'] == int(row['sequence_length']) |
| assert abs(sum(r['confidence'] for r in residues)/len(residues) - case['mean_plddt']) < 1e-8 |
| filename = 'atlas-' + case['accession'] + '.pdb' |
| (PUBLIC/'structures'/filename).write_bytes(pdb) |
| fields = ('uniref_id', 'accession', 'pred_text', 'sequence_length', 'map_xy', 'organism', |
| 'source_url', 'afdb_entry', 'mean_plddt', 'sequence_sha256', 'pdb_sha256', |
| 'original_name', 'domain_support_display') |
| output_cases.append({**{k: case[k] for k in fields}, 'sequence': sequence, |
| 'residues': residues, 'pdbUrl': '/structures/'+filename, 'thumbnail': thumbnail(case), |
| 'number': number, 'title': titles[number-1], |
| |
| 'prediction_status': row['prediction_status'], 'output_hit_limit': row['output_hit_limit'].lower() == 'true', |
| 'model': 'Ouroboros FIX large', |
| 'selection_note': 'Paper illustration; generated function has not been experimentally validated.'}) |
| colors = json.loads((PANEL/'figures/cluster_colors.json').read_text())['cluster_colors'] |
| sources = {str(p.relative_to(ROOT)): sha(p.read_bytes()) for p in [coordinates, labels_path, |
| PANEL/'figures/cluster_colors.json', CASES/'main_cases.json', CASES/'supplementary_cases.json']} |
| angle = -45 * np.pi / 180 |
| display_xy = xy @ np.array([[np.cos(angle), np.sin(angle)], [-np.sin(angle), np.cos(angle)]]) |
| def asset(indices): |
| return dict(ids=[rows[i]['uniref_id'] for i in indices], |
| xy=np.round(xy[indices].astype(float), 6).tolist(), |
| reference=[rows[i]['reference_status']=='swissprot2024_exact_match' for i in indices], |
| lengths=[int(rows[i]['sequence_length']) for i in indices], |
| clusters=labels[indices].tolist(), colors=colors, |
| bounds=[xy.min(axis=0).tolist(), xy.max(axis=0).tolist()], |
| display_bounds=[display_xy.min(axis=0).tolist(), display_xy.max(axis=0).tolist()], |
| rotation_degrees=-45, records=len(indices), population=51321839, |
| source_records=1000000, sample_seed=SEED, |
| sampling='Community-stratified display sample with all eight paper cases retained; display proportions are not corpus estimates.', |
| reference_label='Exact sequence match to Swiss-Prot 2024_01', |
| source_sample_sha256=cluster['inputs']['sample_sha256'], sources=sources) |
| files = {} |
| for name, value in [('atlas-map.json', asset(main)), ('atlas-preview.json', asset(preview)), ('atlas-cases.json', output_cases)]: |
| payload = json.dumps(value, ensure_ascii=False, separators=(',', ':')).encode() |
| (PUBLIC/'data'/name).write_bytes(payload) |
| if name != 'atlas-cases.json': |
| (PUBLIC/'data'/(name+'.gz')).write_bytes(gzip.compress(payload, compresslevel=9, mtime=0)) |
| files[name] = dict(bytes=len(payload), sha256=sha(payload)) |
| report = dict(status='passed', records=len(main), preview_records=len(preview), cases=len(output_cases), |
| communities=len(colors), full_lookup_records=51321839, files=files, sources=sources) |
| (PUBLIC/'data/atlas-provenance.json').write_text(json.dumps(report, indent=2)+'\n') |
| print(json.dumps(report)) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
