From Genome Assemblies to Glycan Structures & Vaccine Formulations¶

from seroverse.client.ncbi import NCBIClient
from seroverse.core.io.fasta import GenomeAssembly
from seroverse.core.seq.clustering import Clusterer

# 1. Load a multi-contig FASTA assembly into a zero-copy 2-bit DNA SoA container
assembly = GenomeAssembly.from_fasta("klebsiella_isolate.fasta")
print(f"Loaded {len(assembly)} contigs ({assembly.total_length:,} bp)")

# 2. Cluster protein sequences using SIMD Smith-Waterman-Gotoh + sparse Markov Clustering (MCL)
clusterer = Clusterer(mode="mcl", min_identity=0.30)

# 3. Query NCBI Entrez / Datasets asynchronously with token-bucket rate limiting
async with NCBIClient() as ncbi:
    summaries = await ncbi.fetch_nuccore_summaries(["CP014225.1"])

from seroref.loci.matcher import LocusMatcher
from seroref.loci.models import LocusDatabase
from seroref.loci.predictor import LocusPredictor
from seroverse.core.io.fasta import GenomeAssembly

# 1. Load a bacterial genome assembly (single or fragmented multi-contig FASTA)
assembly = GenomeAssembly.from_fasta("klebsiella_isolate.fasta")

# 2. Predict ORFs (bacterial + prophage), classify 29 CNN gene roles, and segment loci via ACP-DP
predictor = LocusPredictor()
loci = predictor.predict(assembly)

for locus in loci:
    print(f"{locus.archetype.name}: {locus.start}..{locus.end} ({len(locus.genes)} genes)")

# 3. Match segmented loci against a compiled reference database with bipartite synteny alignment
db = LocusDatabase.load("spneumo_cps.npz")
matcher = LocusMatcher.from_database(db)
best = matcher.match(loci)[0]
print(f"Serotype: {best.serotype} (score={best.score:.3f}, AA identity={best.mean_aa_identity:.1%})")

from kbank.glycan import Glycan
from serogen.biosynthesis.align import BiosyntheticAssigner
from serogen.biosynthesis.predictor import GlycanPredictor
from serogen.db.records import SeroGenRecords
from seroref.loci.predictor import LocusPredictor
from seroverse.core.io.fasta import GenomeAssembly

# 1. Parse a linear CSDB/IUPAC polysaccharide structure into a directed CSR chemical graph
glycan = Glycan.from_csdb("-3)[aDGlcp(1-2)]bDGalp(1-")

# 2. Segment a capsular polysaccharide locus from raw assembly DNA via SeroRef
assembly = GenomeAssembly.from_fasta("KL106.fasta")
locus = LocusPredictor().predict(assembly)[0]

# 3. Predict the repeating-unit structure de novo or align enzymes to linkages via SeroGen
records = SeroGenRecords.load()
predictor = GlycanPredictor.from_records(records)
top_hit = predictor.predict(locus, top_k=1)[0]
assigner = BiosyntheticAssigner()
assignment = assigner.assign_record(top_hit.glycan, locus)

from seroepi.aggregate import Aggregator
from seroepi.estimators.formulation import FormulationEstimator
from seroepi.estimators.hierarchical import HierarchicalSVIEstimator
from seroepi.examples import KlebnetNeonatalSepsis
from seroepi.plotting.choropleth import ChoroplethPlotter

# 1. Load the KlebNET Neonatal Sepsis surveillance cohort (981 K. pneumoniae isolates)
ds = KlebnetNeonatalSepsis.load_built()

# 2. Aggregate K-locus serotype counts stratified by country/site
agg = Aggregator(target=ds.var("K_locus"), strata=[ds.var("country")]).fit(ds)

# 3. Fit a NumPyro Hierarchical Bayesian estimator & 10-valent vaccine formulation
est = HierarchicalSVIEstimator().fit(agg)
vaccine = FormulationEstimator(max_valency=10).fit(est)
chart = ChoroplethPlotter().plot(est)