scripts/align_genes.py

"""Align raw study cohorts to shared gene identifiers before fitting PCA."""
import argparse
from pathlib import Path
import anndata as ad

p = argparse.ArgumentParser()
p.add_argument('--inputs', nargs='+', required=True)
p.add_argument('--output', required=True)
p.add_argument('--gene-columns', nargs='+', help='One var column per input. Use index for var_names.')
p.add_argument('--drop-ambiguous', action='store_true', help='Drop all duplicated gene identifiers')
a = p.parse_args()
cohorts = [ad.read_h5ad(f) for f in a.inputs]
if a.gene_columns and len(a.gene_columns) != len(cohorts): raise ValueError('Supply one gene-column choice per input')
for i, cohort in enumerate(cohorts):
    field = a.gene_columns[i] if a.gene_columns else 'index'
    if field != 'index':
        if cohort.var[field].isna().any():
            cohort = cohort[:, cohort.var[field].notna()].copy()
        cohort.var_names = cohort.var[field].astype(str)
    if not cohort.var_names.is_unique:
        if not a.drop_ambiguous: raise ValueError('Duplicate gene identifiers. Use --drop-ambiguous to discard these genes.')
        cohort = cohort[:, ~cohort.var_names.duplicated(keep=False)].copy()
    cohorts[i] = cohort
shared = sorted(set.intersection(*(set(c.var_names) for c in cohorts)))
if len(shared) < 100:
    raise ValueError('Fewer than 100 shared genes. Check identifier conventions.')
out = Path(a.output); out.mkdir(parents=True, exist_ok=True)
for i, (path, cohort) in enumerate(zip(a.inputs, cohorts)):
    cohort[:, shared].copy().write_h5ad(out / f'{i}_{Path(path).stem}.h5ad', compression='gzip')
(out / 'genes.txt').write_text('\n'.join(shared)+'\n')
print(f'Aligned {len(cohorts)} cohorts on {len(shared)} genes')