`seroref.db`¶

Data-oriented species specifications (BacterialSpecies, SpeciesSpec, SpeciesSpecs), deduplicated reference strain isolates (ReferenceStrain, ReferenceStrains), and relational Structure-of-Arrays (SoA) locus records (SeroRefRecord, SeroRefRecords).

Species Specifications & Serotype Rules (`seroref.db.species`)¶

species ¶

Data-oriented bacterial species ontology, serotype grammars, and SoA SpeciesSpecs container.

Classes:

Name	Description
`GramStain`	Cell-envelope Gram stain classification (`0..3`) for bacterial species.
`SerotypeRule`	Declarative rule mapping a serotype token prefix to its `ArchetypeCategory` and canonical prefix.
`SpeciesSpec`	Immutable biological specification for a canonical `BacterialSpecies`.
`SpeciesSpecs`	Ragged Structure-of-Arrays (SoA) batch container over `SpeciesSpec` specifications.
`BacterialSpecies`	Canonical bacterial species ontology (`0..51`) for `SeroRef` reference loci and strains.

Functions:

Name	Description
`archetypes_to_mask`	Pack a collection of `LocusArchetype` values into a 32-bit integer presence bitmask.
`mask_to_archetypes`	Unpack a 32-bit integer presence bitmask into a sorted tuple of `LocusArchetype` enums.

GramStain ¶

Bases: IntEnum

Cell-envelope Gram stain classification (0..3) for bacterial species.

Methods:

Name	Description
`from_name`	Resolve a `GramStain` by enum ID, enum name, or label.
`resolve`	Resolve a `GramStain` identifier, falling back to `default` (or `GramStain.UNKNOWN`).

Attributes:

Name	Type	Description
`label`	`bytes`	Canonical ASCII byte-string label for this Gram stain class.
`label_str`	`str`	Canonical UTF-8 string label for this Gram stain class.

Source code in src/seroref/db/species.py

class GramStain(IntEnum):
    """Cell-envelope Gram stain classification (`0..3`) for bacterial species."""

    NEGATIVE = 0
    POSITIVE = 1
    VARIABLE = 2
    UNKNOWN = 3

    @property
    def label(self) -> bytes:
        """Canonical ASCII byte-string label for this Gram stain class."""
        return _GRAM_STAIN_LABELS[int(self)]

    @property
    def label_str(self) -> str:
        """Canonical UTF-8 string label for this Gram stain class."""
        return self.label.decode("ascii")

    @classmethod
    def from_name(cls, key: str | bytes | int | GramStain) -> GramStain:
        """Resolve a `GramStain` by enum ID, enum name, or label."""
        if isinstance(key, cls):
            return key
        if isinstance(key, (int, np.integer)):
            return cls(int(key))
        b = key.encode("utf-8") if isinstance(key, str) else bytes(key)
        norm = b.strip().lower()
        for member in cls:
            if norm in (member.name.encode("ascii").lower(), member.label.lower()):
                return member
        raise KeyError(f"Unknown GramStain identifier: {key!r}")

    @classmethod
    def resolve(
        cls,
        key: str | bytes | int | GramStain,
        default: GramStain | None = None,
    ) -> GramStain:
        """Resolve a `GramStain` identifier, falling back to `default` (or `GramStain.UNKNOWN`)."""
        fallback = cls.UNKNOWN if default is None else default
        try:
            return cls.from_name(key)
        except KeyError, ValueError:
            return fallback

label `property` ¶

label: bytes

Canonical ASCII byte-string label for this Gram stain class.

label_str `property` ¶

label_str: str

Canonical UTF-8 string label for this Gram stain class.

from_name `classmethod` ¶

from_name(key: str | bytes | int | GramStain) -> GramStain

Resolve a GramStain by enum ID, enum name, or label.

Source code in src/seroref/db/species.py

@classmethod
def from_name(cls, key: str | bytes | int | GramStain) -> GramStain:
    """Resolve a `GramStain` by enum ID, enum name, or label."""
    if isinstance(key, cls):
        return key
    if isinstance(key, (int, np.integer)):
        return cls(int(key))
    b = key.encode("utf-8") if isinstance(key, str) else bytes(key)
    norm = b.strip().lower()
    for member in cls:
        if norm in (member.name.encode("ascii").lower(), member.label.lower()):
            return member
    raise KeyError(f"Unknown GramStain identifier: {key!r}")

resolve `classmethod` ¶

resolve(
    key: str | bytes | int | GramStain,
    default: GramStain | None = None,
) -> GramStain

Resolve a GramStain identifier, falling back to default (or GramStain.UNKNOWN).

Source code in src/seroref/db/species.py

@classmethod
def resolve(
    cls,
    key: str | bytes | int | GramStain,
    default: GramStain | None = None,
) -> GramStain:
    """Resolve a `GramStain` identifier, falling back to `default` (or `GramStain.UNKNOWN`)."""
    fallback = cls.UNKNOWN if default is None else default
    try:
        return cls.from_name(key)
    except KeyError, ValueError:
        return fallback

SerotypeRule `dataclass` ¶

Bases: ScalarView

Declarative rule mapping a serotype token prefix to its ArchetypeCategory and canonical prefix.

Source code in src/seroref/db/species.py

@dataclass(frozen=True, slots=True)
class SerotypeRule(ScalarView):
    """Declarative rule mapping a serotype token prefix to its `ArchetypeCategory` and canonical prefix."""

    prefix: bytes
    category: ArchetypeCategory
    canonical_prefix: bytes = b""

SpeciesSpec `dataclass` ¶

Bases: ScalarView

Immutable biological specification for a canonical BacterialSpecies.

Encapsulates taxonomic identity, cell-envelope architecture, expected LocusArchetype bitmask, declarative serotype formula rules, serovar-to-O-group mappings, and landmark BGC accession series.

Methods:

Name	Description
`supports_archetype`	Return `True` if `archetype` is enabled in `self.archetype_mask`.
`is_strain_or_mutant_token`	Return `True` if `token` is a laboratory strain, culture-collection ID, or genetic mutant.
`canonicalize_token`	Normalize a single serotype token into `(ArchetypeCategory, canonical_bytes)` or `None`.
`parse_serotype`	Parse a single or multi-antigen serotype formula (e.g. `b"O1:K1(L1):H7"`) into typed tokens.
`normalize_strain`	Normalize a strain isolate designation, rescuing strain/serovar tokens from `raw_serotype` when needed.

Attributes:

Name	Type	Description
`scientific_name`	`bytes`	Canonical scientific name as ASCII bytes (alias for `self.name`).
`synonyms`	`tuple[bytes, ...]`	Taxonomic synonyms and genus aliases as ASCII bytes (alias for `self.aliases`).
`name_str`	`str`	UTF-8 scientific name of this bacterial species.
`scientific_name_str`	`str`	UTF-8 scientific name of this bacterial species.
`genus`	`bytes`	Genus byte string extracted from `self.name`.
`archetypes`	`tuple[LocusArchetype, ...]`	Sorted tuple of `LocusArchetype`s compatible with this species.

Source code in src/seroref/db/species.py

@dataclass(frozen=True, slots=True)
class SpeciesSpec(ScalarView):
    """Immutable biological specification for a canonical `BacterialSpecies`.

    Encapsulates taxonomic identity, cell-envelope architecture, expected `LocusArchetype` bitmask,
    declarative serotype formula rules, serovar-to-O-group mappings, and landmark BGC accession series.
    """

    name: bytes
    ncbi_taxid: int
    gram_stain: GramStain
    default_category: ArchetypeCategory
    archetype_mask: int
    serotype_rules: tuple[SerotypeRule, ...] = ()
    serovar_map: tuple[tuple[bytes, bytes], ...] = ()
    aliases: tuple[bytes, ...] = ()
    landmark_accessions: tuple[tuple[bytes, bytes], ...] = ()
    known_serotypes: tuple[bytes, ...] = ()

    @property
    def scientific_name(self) -> bytes:
        """Canonical scientific name as ASCII bytes (alias for `self.name`)."""
        return self.name

    @property
    def synonyms(self) -> tuple[bytes, ...]:
        """Taxonomic synonyms and genus aliases as ASCII bytes (alias for `self.aliases`)."""
        return self.aliases

    @property
    def name_str(self) -> str:
        """UTF-8 scientific name of this bacterial species."""
        return self.name.decode("utf-8", errors="replace")

    @property
    def scientific_name_str(self) -> str:
        """UTF-8 scientific name of this bacterial species."""
        return self.name_str

    @property
    def genus(self) -> bytes:
        """Genus byte string extracted from `self.name`."""
        return self.name.split(b" ", 1)[0]

    @property
    def archetypes(self) -> tuple[LocusArchetype, ...]:
        """Sorted tuple of `LocusArchetype`s compatible with this species."""
        return mask_to_archetypes(self.archetype_mask)

    def supports_archetype(self, archetype: LocusArchetype) -> bool:
        """Return `True` if `archetype` is enabled in `self.archetype_mask`."""
        return bool(self.archetype_mask & (1 << int(archetype)))

    def _serovar_lookup(self) -> dict[bytes, bytes]:
        return {k.lower(): v for k, v in self.serovar_map}

    def is_strain_or_mutant_token(self, token: str | bytes) -> bool:
        """Return `True` if `token` is a laboratory strain, culture-collection ID, or genetic mutant."""
        b = token.encode("utf-8").strip() if isinstance(token, str) else bytes(token).strip()
        if not b:
            return False
        if _MUTANT_OR_LAB_STRAIN_RE.search(b):
            return True
        return b.lower() in self._serovar_lookup()

    def canonicalize_token(
        self,
        raw_token: str | bytes,
        *,
        archetype: LocusArchetype | None = None,
    ) -> tuple[ArchetypeCategory, bytes] | None:
        """Normalize a single serotype token into `(ArchetypeCategory, canonical_bytes)` or `None`."""
        b = raw_token.encode("utf-8").strip() if isinstance(raw_token, str) else bytes(raw_token).strip()
        low = b.lower()
        if not b or low in _BAD_PLACEHOLDER_TOKENS or _FLAGELLAR_ONLY_RE.match(b):
            return None

        serovars = self._serovar_lookup()
        if low in serovars:
            return (ArchetypeCategory.O_ANTIGEN, serovars[low])

        if _MUTANT_OR_LAB_STRAIN_RE.search(b):
            return None

        if m_num := _TYPED_NUM_RE.search(b):
            num_tok = m_num.group(1)
            cat = (
                archetype.category
                if (archetype is not None and archetype != LocusArchetype.UNKNOWN)
                else self.default_category
            )
            return (cat, self._format_bare_token(num_tok, cat))

        if m_zero := _LEADING_ZERO_O_RE.match(b):
            return (ArchetypeCategory.O_ANTIGEN, b"O%s" % m_zero.group(1))
        if m_col := _COLON_O_RE.match(b):
            return (ArchetypeCategory.O_ANTIGEN, b"O%s" % m_col.group(1))

        up = b.upper()
        for rule in self.serotype_rules:
            pref_up = rule.prefix.upper()
            if up.startswith(pref_up) and len(b) > len(pref_up):
                rest = b[len(pref_up) :].lstrip(b":- ")
                if rest and (rest[:1].isdigit() or len(rest) <= 4):
                    canon = b"%s%s" % (rule.canonical_prefix, rest) if rule.canonical_prefix else b
                    return (rule.category, canon)

        target_cat = (
            archetype.category
            if (archetype is not None and archetype != LocusArchetype.UNKNOWN)
            else self.default_category
        )
        if _BARE_NUMBER_SERO_RE.match(b):
            return (target_cat, self._format_bare_token(b, target_cat))

        return (target_cat, b)

    def _format_bare_token(self, bare: bytes, category: ArchetypeCategory) -> bytes:
        for rule in self.serotype_rules:
            if rule.category == category and rule.canonical_prefix:
                if not bare.upper().startswith(rule.canonical_prefix.upper()):
                    return b"%s%s" % (rule.canonical_prefix, bare)
                return bare
        return bare

    def parse_serotype(
        self,
        raw: str | bytes,
        *,
        archetype: LocusArchetype | None = None,
    ) -> tuple[tuple[ArchetypeCategory, bytes], ...]:
        """Parse a single or multi-antigen serotype formula (e.g. `b"O1:K1(L1):H7"`) into typed tokens."""
        b = raw.encode("utf-8").strip() if isinstance(raw, str) else bytes(raw).strip()
        if not b or b.lower() in _BAD_PLACEHOLDER_TOKENS:
            return ()

        cleaned = re.sub(rb"\([Ll]\d*\)", b"", b).strip()
        ok_hits = _OK_TOKEN_RE.findall(cleaned)
        if len(ok_hits) >= 2 or (len(ok_hits) == 1 and b":" in cleaned):
            results: list[tuple[ArchetypeCategory, bytes]] = []
            seen: set[tuple[ArchetypeCategory, bytes]] = set()
            for tok in ok_hits:
                parsed = self.canonicalize_token(tok, archetype=archetype)
                if parsed is not None and parsed not in seen:
                    seen.add(parsed)
                    results.append(parsed)
            if results:
                return tuple(results)

        single = self.canonicalize_token(cleaned, archetype=archetype)
        return (single,) if single is not None else ()

    def normalize_strain(
        self,
        raw_strain: str | bytes,
        raw_serotype: str | bytes = b"",
    ) -> bytes:
        """Normalize a strain isolate designation, rescuing strain/serovar tokens from `raw_serotype` when needed."""
        s_b = raw_strain.encode("utf-8").strip() if isinstance(raw_strain, str) else bytes(raw_strain).strip()
        sero_b = raw_serotype.encode("utf-8").strip() if isinstance(raw_serotype, str) else bytes(raw_serotype).strip()

        # Rescue mutant/lab strain or Salmonella serovar mistakenly placed in raw_serotype
        if not s_b and sero_b and self.is_strain_or_mutant_token(sero_b):
            s_b = sero_b

        if not s_b or s_b.lower() in _BAD_PLACEHOLDER_TOKENS or _FLAGELLAR_ONLY_RE.match(s_b):
            return b""

        # Reject if strain is identical to the canonicalized serotype token or raw serotype formula
        if sero_b and s_b.lower() == sero_b.lower() and not self.is_strain_or_mutant_token(s_b):
            return b""
        for _, canon_b in self.parse_serotype(sero_b):
            if s_b.lower() == canon_b.lower():
                return b""

        # Reject bare O/K antigen formulas masquerading as strain names unless accompanied by an isolate token
        if not self.is_strain_or_mutant_token(s_b):
            ok_tokens = _OK_TOKEN_RE.findall(s_b)
            if (
                ok_tokens
                and b":".join(ok_tokens).lower() == re.sub(rb":H\d+", b"", s_b, flags=re.I).strip(b":").lower()
            ):
                return b""
            if s_b in self.known_serotypes:
                return b""

        return s_b

scientific_name `property` ¶

scientific_name: bytes

Canonical scientific name as ASCII bytes (alias for self.name).

synonyms `property` ¶

synonyms: tuple[bytes, ...]

Taxonomic synonyms and genus aliases as ASCII bytes (alias for self.aliases).

name_str `property` ¶

name_str: str

UTF-8 scientific name of this bacterial species.

scientific_name_str `property` ¶

scientific_name_str: str

UTF-8 scientific name of this bacterial species.

genus `property` ¶

genus: bytes

Genus byte string extracted from self.name.

archetypes `property` ¶

archetypes: tuple[LocusArchetype, ...]

Sorted tuple of LocusArchetypes compatible with this species.

supports_archetype ¶

supports_archetype(archetype: LocusArchetype) -> bool

Return True if archetype is enabled in self.archetype_mask.

Source code in src/seroref/db/species.py

def supports_archetype(self, archetype: LocusArchetype) -> bool:
    """Return `True` if `archetype` is enabled in `self.archetype_mask`."""
    return bool(self.archetype_mask & (1 << int(archetype)))

is_strain_or_mutant_token ¶

is_strain_or_mutant_token(token: str | bytes) -> bool

Return True if token is a laboratory strain, culture-collection ID, or genetic mutant.

Source code in src/seroref/db/species.py

def is_strain_or_mutant_token(self, token: str | bytes) -> bool:
    """Return `True` if `token` is a laboratory strain, culture-collection ID, or genetic mutant."""
    b = token.encode("utf-8").strip() if isinstance(token, str) else bytes(token).strip()
    if not b:
        return False
    if _MUTANT_OR_LAB_STRAIN_RE.search(b):
        return True
    return b.lower() in self._serovar_lookup()

canonicalize_token ¶

canonicalize_token(
    raw_token: str | bytes,
    *,
    archetype: LocusArchetype | None = None,
) -> tuple[ArchetypeCategory, bytes] | None

Normalize a single serotype token into (ArchetypeCategory, canonical_bytes) or None.

Source code in src/seroref/db/species.py

def canonicalize_token(
    self,
    raw_token: str | bytes,
    *,
    archetype: LocusArchetype | None = None,
) -> tuple[ArchetypeCategory, bytes] | None:
    """Normalize a single serotype token into `(ArchetypeCategory, canonical_bytes)` or `None`."""
    b = raw_token.encode("utf-8").strip() if isinstance(raw_token, str) else bytes(raw_token).strip()
    low = b.lower()
    if not b or low in _BAD_PLACEHOLDER_TOKENS or _FLAGELLAR_ONLY_RE.match(b):
        return None

    serovars = self._serovar_lookup()
    if low in serovars:
        return (ArchetypeCategory.O_ANTIGEN, serovars[low])

    if _MUTANT_OR_LAB_STRAIN_RE.search(b):
        return None

    if m_num := _TYPED_NUM_RE.search(b):
        num_tok = m_num.group(1)
        cat = (
            archetype.category
            if (archetype is not None and archetype != LocusArchetype.UNKNOWN)
            else self.default_category
        )
        return (cat, self._format_bare_token(num_tok, cat))

    if m_zero := _LEADING_ZERO_O_RE.match(b):
        return (ArchetypeCategory.O_ANTIGEN, b"O%s" % m_zero.group(1))
    if m_col := _COLON_O_RE.match(b):
        return (ArchetypeCategory.O_ANTIGEN, b"O%s" % m_col.group(1))

    up = b.upper()
    for rule in self.serotype_rules:
        pref_up = rule.prefix.upper()
        if up.startswith(pref_up) and len(b) > len(pref_up):
            rest = b[len(pref_up) :].lstrip(b":- ")
            if rest and (rest[:1].isdigit() or len(rest) <= 4):
                canon = b"%s%s" % (rule.canonical_prefix, rest) if rule.canonical_prefix else b
                return (rule.category, canon)

    target_cat = (
        archetype.category
        if (archetype is not None and archetype != LocusArchetype.UNKNOWN)
        else self.default_category
    )
    if _BARE_NUMBER_SERO_RE.match(b):
        return (target_cat, self._format_bare_token(b, target_cat))

    return (target_cat, b)

parse_serotype ¶

parse_serotype(
    raw: str | bytes,
    *,
    archetype: LocusArchetype | None = None,
) -> tuple[tuple[ArchetypeCategory, bytes], ...]

Parse a single or multi-antigen serotype formula (e.g. b"O1:K1(L1):H7") into typed tokens.

Source code in src/seroref/db/species.py

def parse_serotype(
    self,
    raw: str | bytes,
    *,
    archetype: LocusArchetype | None = None,
) -> tuple[tuple[ArchetypeCategory, bytes], ...]:
    """Parse a single or multi-antigen serotype formula (e.g. `b"O1:K1(L1):H7"`) into typed tokens."""
    b = raw.encode("utf-8").strip() if isinstance(raw, str) else bytes(raw).strip()
    if not b or b.lower() in _BAD_PLACEHOLDER_TOKENS:
        return ()

    cleaned = re.sub(rb"\([Ll]\d*\)", b"", b).strip()
    ok_hits = _OK_TOKEN_RE.findall(cleaned)
    if len(ok_hits) >= 2 or (len(ok_hits) == 1 and b":" in cleaned):
        results: list[tuple[ArchetypeCategory, bytes]] = []
        seen: set[tuple[ArchetypeCategory, bytes]] = set()
        for tok in ok_hits:
            parsed = self.canonicalize_token(tok, archetype=archetype)
            if parsed is not None and parsed not in seen:
                seen.add(parsed)
                results.append(parsed)
        if results:
            return tuple(results)

    single = self.canonicalize_token(cleaned, archetype=archetype)
    return (single,) if single is not None else ()

normalize_strain ¶

normalize_strain(
    raw_strain: str | bytes, raw_serotype: str | bytes = b""
) -> bytes

Normalize a strain isolate designation, rescuing strain/serovar tokens from raw_serotype when needed.

Source code in src/seroref/db/species.py

def normalize_strain(
    self,
    raw_strain: str | bytes,
    raw_serotype: str | bytes = b"",
) -> bytes:
    """Normalize a strain isolate designation, rescuing strain/serovar tokens from `raw_serotype` when needed."""
    s_b = raw_strain.encode("utf-8").strip() if isinstance(raw_strain, str) else bytes(raw_strain).strip()
    sero_b = raw_serotype.encode("utf-8").strip() if isinstance(raw_serotype, str) else bytes(raw_serotype).strip()

    # Rescue mutant/lab strain or Salmonella serovar mistakenly placed in raw_serotype
    if not s_b and sero_b and self.is_strain_or_mutant_token(sero_b):
        s_b = sero_b

    if not s_b or s_b.lower() in _BAD_PLACEHOLDER_TOKENS or _FLAGELLAR_ONLY_RE.match(s_b):
        return b""

    # Reject if strain is identical to the canonicalized serotype token or raw serotype formula
    if sero_b and s_b.lower() == sero_b.lower() and not self.is_strain_or_mutant_token(s_b):
        return b""
    for _, canon_b in self.parse_serotype(sero_b):
        if s_b.lower() == canon_b.lower():
            return b""

    # Reject bare O/K antigen formulas masquerading as strain names unless accompanied by an isolate token
    if not self.is_strain_or_mutant_token(s_b):
        ok_tokens = _OK_TOKEN_RE.findall(s_b)
        if (
            ok_tokens
            and b":".join(ok_tokens).lower() == re.sub(rb":H\d+", b"", s_b, flags=re.I).strip(b":").lower()
        ):
            return b""
        if s_b in self.known_serotypes:
            return b""

    return s_b

SpeciesSpecs `dataclass` ¶

Bases: RaggedArrayContainer[SpeciesSpec, 'SpeciesSpecs']

Ragged Structure-of-Arrays (SoA) batch container over SpeciesSpec specifications.

All 1D fields share leading dimension len(self) (52 for the canonical BacterialSpecies ontology). Variable-length per-species metadata (aliases, serotype_rules, serovar_map, landmark_accessions, and known_serotypes) are CSR-packed into contiguous NumPy arrays without Python object tuples.

Methods:

Name	Description
`empty`	Create an empty 0-length `SpeciesSpecs` container.
`from_list`	Compile a sequence of `SpeciesSpec` scalar views into a `SpeciesSpecs` SoA container.
`concat`	Concatenate multiple `SpeciesSpecs` containers.

Attributes:

Name	Type	Description
`scientific_names`	`ndarray[tuple[int], dtype[bytes_]]`	1D byte-string array of canonical scientific names (alias for `self.names`).
`alias_offsets`	`ndarray[tuple[int], dtype[int32]]`	Alias for `self.offsets` (starting indices into `self.alias_values`).
`alias_lengths`	`ndarray[tuple[int], dtype[int32]]`	Alias for `self.lengths` (number of taxonomic aliases per species).

Source code in src/seroref/db/species.py

@dataclass(frozen=True, slots=True)
class SpeciesSpecs(RaggedArrayContainer[SpeciesSpec, "SpeciesSpecs"]):
    """Ragged Structure-of-Arrays (SoA) batch container over `SpeciesSpec` specifications.

    All 1D fields share leading dimension `len(self)` (`52` for the canonical `BacterialSpecies` ontology).
    Variable-length per-species metadata (`aliases`, `serotype_rules`, `serovar_map`, `landmark_accessions`,
    and `known_serotypes`) are CSR-packed into contiguous NumPy arrays without Python object tuples.
    """

    names: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    ncbi_taxids: np.ndarray[tuple[int], np.dtype[np.int32]]
    gram_stains: np.ndarray[tuple[int], np.dtype[np.int8]]
    default_categories: np.ndarray[tuple[int], np.dtype[np.int8]]
    archetype_masks: np.ndarray[tuple[int], np.dtype[np.uint32]]
    alias_values: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    rule_prefixes: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    rule_categories: np.ndarray[tuple[int], np.dtype[np.int8]]
    rule_canonical_prefixes: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    rule_offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    rule_lengths: np.ndarray[tuple[int], np.dtype[np.int32]]
    serovar_keys: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    serovar_targets: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    serovar_offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    serovar_lengths: np.ndarray[tuple[int], np.dtype[np.int32]]
    landmark_starts: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    landmark_ends: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    landmark_offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    landmark_lengths: np.ndarray[tuple[int], np.dtype[np.int32]]
    known_serotype_values: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    known_serotype_offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    known_serotype_lengths: np.ndarray[tuple[int], np.dtype[np.int32]]
    offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    lengths: np.ndarray[tuple[int], np.dtype[np.int32]]

    def __len__(self) -> int:
        return len(self.names)

    def __iter__(self) -> Iterator[SpeciesSpec]:
        for i in range(len(self)):
            yield self[i]

    @property
    def scientific_names(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """1D byte-string array of canonical scientific names (alias for `self.names`)."""
        return self.names

    @property
    def alias_offsets(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """Alias for `self.offsets` (starting indices into `self.alias_values`)."""
        return self.offsets

    @property
    def alias_lengths(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """Alias for `self.lengths` (number of taxonomic aliases per species)."""
        return self.lengths

    @overload
    def __getitem__(self, item: int | np.integer[Any]) -> SpeciesSpec: ...

    @overload
    def __getitem__(self, item: slice | np.ndarray[tuple[int], Any] | list[Any]) -> Self: ...

    @overload
    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> SpeciesSpec | Self: ...

    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> SpeciesSpec | Self:
        if isinstance(item, (int, np.integer)):
            idx = int(item)
            n = len(self)
            if idx < 0:
                idx += n
            if idx < 0 or idx >= n:
                raise IndexError(f"SpeciesSpecs index {item} out of range for length {n}.")

            a_off, a_ln = int(self.offsets[idx]), int(self.lengths[idx])
            aliases = tuple(bytes(self.alias_values[k]) for k in range(a_off, a_off + a_ln))

            r_off, r_ln = int(self.rule_offsets[idx]), int(self.rule_lengths[idx])
            rules = tuple(
                SerotypeRule(
                    prefix=bytes(self.rule_prefixes[k]),
                    category=ArchetypeCategory(int(self.rule_categories[k])),
                    canonical_prefix=bytes(self.rule_canonical_prefixes[k]),
                )
                for k in range(r_off, r_off + r_ln)
            )

            s_off, s_ln = int(self.serovar_offsets[idx]), int(self.serovar_lengths[idx])
            serovars = tuple(
                (bytes(self.serovar_keys[k]), bytes(self.serovar_targets[k])) for k in range(s_off, s_off + s_ln)
            )

            l_off, l_ln = int(self.landmark_offsets[idx]), int(self.landmark_lengths[idx])
            landmarks = tuple(
                (bytes(self.landmark_starts[k]), bytes(self.landmark_ends[k])) for k in range(l_off, l_off + l_ln)
            )

            k_off, k_ln = int(self.known_serotype_offsets[idx]), int(self.known_serotype_lengths[idx])
            known_seros = tuple(bytes(self.known_serotype_values[k]) for k in range(k_off, k_off + k_ln))

            return SpeciesSpec(
                name=bytes(self.names[idx]),
                ncbi_taxid=int(self.ncbi_taxids[idx]),
                gram_stain=GramStain(int(self.gram_stains[idx])),
                default_category=ArchetypeCategory(int(self.default_categories[idx])),
                archetype_mask=int(self.archetype_masks[idx]),
                serotype_rules=rules,
                serovar_map=serovars,
                aliases=aliases,
                landmark_accessions=landmarks,
                known_serotypes=known_seros,
            )

        alias_idx_pool = np.arange(len(self.alias_values), dtype=np.int32)
        indices, new_offsets, new_lengths, (gathered_alias_idx,) = self._slice_ragged(item, alias_idx_pool)
        if len(indices) == 0:
            return self.empty()

        alias_v = self.alias_values[gathered_alias_idx] if len(gathered_alias_idx) > 0 else np.empty(0, dtype="S1")

        rule_idx_pool = np.arange(len(self.rule_prefixes), dtype=np.int32)
        rule_idx, r_off, r_ln = _gather_ragged_secondary(
            rule_idx_pool,
            self.rule_offsets,
            self.rule_lengths,
            indices,
        )
        r_pref = self.rule_prefixes[rule_idx] if len(rule_idx) > 0 else np.empty(0, dtype="S1")
        r_cats = self.rule_categories[rule_idx] if len(rule_idx) > 0 else np.empty(0, dtype=np.int8)
        r_canon = self.rule_canonical_prefixes[rule_idx] if len(rule_idx) > 0 else np.empty(0, dtype="S1")

        sero_idx_pool = np.arange(len(self.serovar_keys), dtype=np.int32)
        sero_idx, s_off, s_ln = _gather_ragged_secondary(
            sero_idx_pool,
            self.serovar_offsets,
            self.serovar_lengths,
            indices,
        )
        s_keys = self.serovar_keys[sero_idx] if len(sero_idx) > 0 else np.empty(0, dtype="S1")
        s_targs = self.serovar_targets[sero_idx] if len(sero_idx) > 0 else np.empty(0, dtype="S1")

        lm_idx_pool = np.arange(len(self.landmark_starts), dtype=np.int32)
        lm_idx, l_off, l_ln = _gather_ragged_secondary(
            lm_idx_pool,
            self.landmark_offsets,
            self.landmark_lengths,
            indices,
        )
        l_starts = self.landmark_starts[lm_idx] if len(lm_idx) > 0 else np.empty(0, dtype="S1")
        l_ends = self.landmark_ends[lm_idx] if len(lm_idx) > 0 else np.empty(0, dtype="S1")

        ks_v, ks_off, ks_ln = _gather_ragged_secondary(
            self.known_serotype_values,
            self.known_serotype_offsets,
            self.known_serotype_lengths,
            indices,
        )

        return type(self)(
            names=self.names[indices],
            ncbi_taxids=self.ncbi_taxids[indices],
            gram_stains=self.gram_stains[indices],
            default_categories=self.default_categories[indices],
            archetype_masks=self.archetype_masks[indices],
            alias_values=alias_v,
            offsets=new_offsets,
            lengths=new_lengths,
            rule_prefixes=r_pref,
            rule_categories=r_cats,
            rule_canonical_prefixes=r_canon,
            rule_offsets=r_off,
            rule_lengths=r_ln,
            serovar_keys=s_keys,
            serovar_targets=s_targs,
            serovar_offsets=s_off,
            serovar_lengths=s_ln,
            landmark_starts=l_starts,
            landmark_ends=l_ends,
            landmark_offsets=l_off,
            landmark_lengths=l_ln,
            known_serotype_values=ks_v,
            known_serotype_offsets=ks_off,
            known_serotype_lengths=ks_ln,
        )

    @classmethod
    def empty(cls) -> Self:
        """Create an empty 0-length `SpeciesSpecs` container."""
        return cls(
            names=np.empty(0, dtype="S1"),
            ncbi_taxids=np.empty(0, dtype=np.int32),
            gram_stains=np.empty(0, dtype=np.int8),
            default_categories=np.empty(0, dtype=np.int8),
            archetype_masks=np.empty(0, dtype=np.uint32),
            alias_values=np.empty(0, dtype="S1"),
            offsets=np.empty(0, dtype=np.int32),
            lengths=np.empty(0, dtype=np.int32),
            rule_prefixes=np.empty(0, dtype="S1"),
            rule_categories=np.empty(0, dtype=np.int8),
            rule_canonical_prefixes=np.empty(0, dtype="S1"),
            rule_offsets=np.empty(0, dtype=np.int32),
            rule_lengths=np.empty(0, dtype=np.int32),
            serovar_keys=np.empty(0, dtype="S1"),
            serovar_targets=np.empty(0, dtype="S1"),
            serovar_offsets=np.empty(0, dtype=np.int32),
            serovar_lengths=np.empty(0, dtype=np.int32),
            landmark_starts=np.empty(0, dtype="S1"),
            landmark_ends=np.empty(0, dtype="S1"),
            landmark_offsets=np.empty(0, dtype=np.int32),
            landmark_lengths=np.empty(0, dtype=np.int32),
            known_serotype_values=np.empty(0, dtype="S1"),
            known_serotype_offsets=np.empty(0, dtype=np.int32),
            known_serotype_lengths=np.empty(0, dtype=np.int32),
        )

    @classmethod
    def from_list(cls, specs: Sequence[SpeciesSpec]) -> Self:
        """Compile a sequence of `SpeciesSpec` scalar views into a `SpeciesSpecs` SoA container."""
        n = len(specs)
        if n == 0:
            return cls.empty()

        names = np.array([s.name for s in specs], dtype="S64")
        taxids = np.array([s.ncbi_taxid for s in specs], dtype=np.int32)
        grams = np.array([int(s.gram_stain) for s in specs], dtype=np.int8)
        cats = np.array([int(s.default_category) for s in specs], dtype=np.int8)
        masks = np.array([s.archetype_mask for s in specs], dtype=np.uint32)

        alias_flat: list[bytes] = []
        alias_lens = np.empty(n, dtype=np.int32)
        r_pref_flat: list[bytes] = []
        r_cat_flat: list[int] = []
        r_canon_flat: list[bytes] = []
        r_lens = np.empty(n, dtype=np.int32)
        s_key_flat: list[bytes] = []
        s_targ_flat: list[bytes] = []
        s_lens = np.empty(n, dtype=np.int32)
        l_start_flat: list[bytes] = []
        l_end_flat: list[bytes] = []
        l_lens = np.empty(n, dtype=np.int32)
        ks_flat: list[bytes] = []
        ks_lens = np.empty(n, dtype=np.int32)

        for i, s in enumerate(specs):
            alias_lens[i] = len(s.aliases)
            alias_flat.extend(s.aliases)

            r_lens[i] = len(s.serotype_rules)
            for rule in s.serotype_rules:
                r_pref_flat.append(rule.prefix)
                r_cat_flat.append(int(rule.category))
                r_canon_flat.append(rule.canonical_prefix)

            s_lens[i] = len(s.serovar_map)
            for k, v in s.serovar_map:
                s_key_flat.append(k)
                s_targ_flat.append(v)

            l_lens[i] = len(s.landmark_accessions)
            for st, en in s.landmark_accessions:
                l_start_flat.append(st)
                l_end_flat.append(en)

            ks_lens[i] = len(s.known_serotypes)
            ks_flat.extend(s.known_serotypes)

        alias_values = np.array(alias_flat, dtype="S") if alias_flat else np.empty(0, dtype="S1")
        offsets = lengths_to_offsets(alias_lens)

        rule_prefixes = np.array(r_pref_flat, dtype="S") if r_pref_flat else np.empty(0, dtype="S1")
        rule_categories = np.array(r_cat_flat, dtype=np.int8) if r_cat_flat else np.empty(0, dtype=np.int8)
        rule_canonical_prefixes = np.array(r_canon_flat, dtype="S") if r_canon_flat else np.empty(0, dtype="S1")
        rule_offsets = lengths_to_offsets(r_lens)

        serovar_keys = np.array(s_key_flat, dtype="S") if s_key_flat else np.empty(0, dtype="S1")
        serovar_targets = np.array(s_targ_flat, dtype="S") if s_targ_flat else np.empty(0, dtype="S1")
        serovar_offsets = lengths_to_offsets(s_lens)

        landmark_starts = np.array(l_start_flat, dtype="S") if l_start_flat else np.empty(0, dtype="S1")
        landmark_ends = np.array(l_end_flat, dtype="S") if l_end_flat else np.empty(0, dtype="S1")
        landmark_offsets = lengths_to_offsets(l_lens)

        known_serotype_values = np.array(ks_flat, dtype="S") if ks_flat else np.empty(0, dtype="S1")
        known_serotype_offsets = lengths_to_offsets(ks_lens)

        for arr in (
            names,
            taxids,
            grams,
            cats,
            masks,
            alias_values,
            offsets,
            alias_lens,
            rule_prefixes,
            rule_categories,
            rule_canonical_prefixes,
            rule_offsets,
            r_lens,
            serovar_keys,
            serovar_targets,
            serovar_offsets,
            s_lens,
            landmark_starts,
            landmark_ends,
            landmark_offsets,
            l_lens,
            known_serotype_values,
            known_serotype_offsets,
            ks_lens,
        ):
            arr.setflags(write=False)

        return cls(
            names=names,
            ncbi_taxids=taxids,
            gram_stains=grams,
            default_categories=cats,
            archetype_masks=masks,
            alias_values=alias_values,
            offsets=offsets,
            lengths=alias_lens,
            rule_prefixes=rule_prefixes,
            rule_categories=rule_categories,
            rule_canonical_prefixes=rule_canonical_prefixes,
            rule_offsets=rule_offsets,
            rule_lengths=r_lens,
            serovar_keys=serovar_keys,
            serovar_targets=serovar_targets,
            serovar_offsets=serovar_offsets,
            serovar_lengths=s_lens,
            landmark_starts=landmark_starts,
            landmark_ends=landmark_ends,
            landmark_offsets=landmark_offsets,
            landmark_lengths=l_lens,
            known_serotype_values=known_serotype_values,
            known_serotype_offsets=known_serotype_offsets,
            known_serotype_lengths=ks_lens,
        )

    @classmethod
    def concat(cls, batches: Iterable[Self]) -> Self:
        """Concatenate multiple `SpeciesSpecs` containers."""
        items: list[SpeciesSpec] = []
        for b in batches:
            items.extend(b)
        return cls.from_list(items)

scientific_names `property` ¶

scientific_names: ndarray[tuple[int], dtype[bytes_]]

1D byte-string array of canonical scientific names (alias for self.names).

alias_offsets `property` ¶

alias_offsets: ndarray[tuple[int], dtype[int32]]

Alias for self.offsets (starting indices into self.alias_values).

alias_lengths `property` ¶

alias_lengths: ndarray[tuple[int], dtype[int32]]

Alias for self.lengths (number of taxonomic aliases per species).

empty `classmethod` ¶

empty() -> Self

Create an empty 0-length SpeciesSpecs container.

Source code in src/seroref/db/species.py

@classmethod
def empty(cls) -> Self:
    """Create an empty 0-length `SpeciesSpecs` container."""
    return cls(
        names=np.empty(0, dtype="S1"),
        ncbi_taxids=np.empty(0, dtype=np.int32),
        gram_stains=np.empty(0, dtype=np.int8),
        default_categories=np.empty(0, dtype=np.int8),
        archetype_masks=np.empty(0, dtype=np.uint32),
        alias_values=np.empty(0, dtype="S1"),
        offsets=np.empty(0, dtype=np.int32),
        lengths=np.empty(0, dtype=np.int32),
        rule_prefixes=np.empty(0, dtype="S1"),
        rule_categories=np.empty(0, dtype=np.int8),
        rule_canonical_prefixes=np.empty(0, dtype="S1"),
        rule_offsets=np.empty(0, dtype=np.int32),
        rule_lengths=np.empty(0, dtype=np.int32),
        serovar_keys=np.empty(0, dtype="S1"),
        serovar_targets=np.empty(0, dtype="S1"),
        serovar_offsets=np.empty(0, dtype=np.int32),
        serovar_lengths=np.empty(0, dtype=np.int32),
        landmark_starts=np.empty(0, dtype="S1"),
        landmark_ends=np.empty(0, dtype="S1"),
        landmark_offsets=np.empty(0, dtype=np.int32),
        landmark_lengths=np.empty(0, dtype=np.int32),
        known_serotype_values=np.empty(0, dtype="S1"),
        known_serotype_offsets=np.empty(0, dtype=np.int32),
        known_serotype_lengths=np.empty(0, dtype=np.int32),
    )

from_list `classmethod` ¶

from_list(specs: Sequence[SpeciesSpec]) -> Self

Compile a sequence of SpeciesSpec scalar views into a SpeciesSpecs SoA container.

Source code in src/seroref/db/species.py

@classmethod
def from_list(cls, specs: Sequence[SpeciesSpec]) -> Self:
    """Compile a sequence of `SpeciesSpec` scalar views into a `SpeciesSpecs` SoA container."""
    n = len(specs)
    if n == 0:
        return cls.empty()

    names = np.array([s.name for s in specs], dtype="S64")
    taxids = np.array([s.ncbi_taxid for s in specs], dtype=np.int32)
    grams = np.array([int(s.gram_stain) for s in specs], dtype=np.int8)
    cats = np.array([int(s.default_category) for s in specs], dtype=np.int8)
    masks = np.array([s.archetype_mask for s in specs], dtype=np.uint32)

    alias_flat: list[bytes] = []
    alias_lens = np.empty(n, dtype=np.int32)
    r_pref_flat: list[bytes] = []
    r_cat_flat: list[int] = []
    r_canon_flat: list[bytes] = []
    r_lens = np.empty(n, dtype=np.int32)
    s_key_flat: list[bytes] = []
    s_targ_flat: list[bytes] = []
    s_lens = np.empty(n, dtype=np.int32)
    l_start_flat: list[bytes] = []
    l_end_flat: list[bytes] = []
    l_lens = np.empty(n, dtype=np.int32)
    ks_flat: list[bytes] = []
    ks_lens = np.empty(n, dtype=np.int32)

    for i, s in enumerate(specs):
        alias_lens[i] = len(s.aliases)
        alias_flat.extend(s.aliases)

        r_lens[i] = len(s.serotype_rules)
        for rule in s.serotype_rules:
            r_pref_flat.append(rule.prefix)
            r_cat_flat.append(int(rule.category))
            r_canon_flat.append(rule.canonical_prefix)

        s_lens[i] = len(s.serovar_map)
        for k, v in s.serovar_map:
            s_key_flat.append(k)
            s_targ_flat.append(v)

        l_lens[i] = len(s.landmark_accessions)
        for st, en in s.landmark_accessions:
            l_start_flat.append(st)
            l_end_flat.append(en)

        ks_lens[i] = len(s.known_serotypes)
        ks_flat.extend(s.known_serotypes)

    alias_values = np.array(alias_flat, dtype="S") if alias_flat else np.empty(0, dtype="S1")
    offsets = lengths_to_offsets(alias_lens)

    rule_prefixes = np.array(r_pref_flat, dtype="S") if r_pref_flat else np.empty(0, dtype="S1")
    rule_categories = np.array(r_cat_flat, dtype=np.int8) if r_cat_flat else np.empty(0, dtype=np.int8)
    rule_canonical_prefixes = np.array(r_canon_flat, dtype="S") if r_canon_flat else np.empty(0, dtype="S1")
    rule_offsets = lengths_to_offsets(r_lens)

    serovar_keys = np.array(s_key_flat, dtype="S") if s_key_flat else np.empty(0, dtype="S1")
    serovar_targets = np.array(s_targ_flat, dtype="S") if s_targ_flat else np.empty(0, dtype="S1")
    serovar_offsets = lengths_to_offsets(s_lens)

    landmark_starts = np.array(l_start_flat, dtype="S") if l_start_flat else np.empty(0, dtype="S1")
    landmark_ends = np.array(l_end_flat, dtype="S") if l_end_flat else np.empty(0, dtype="S1")
    landmark_offsets = lengths_to_offsets(l_lens)

    known_serotype_values = np.array(ks_flat, dtype="S") if ks_flat else np.empty(0, dtype="S1")
    known_serotype_offsets = lengths_to_offsets(ks_lens)

    for arr in (
        names,
        taxids,
        grams,
        cats,
        masks,
        alias_values,
        offsets,
        alias_lens,
        rule_prefixes,
        rule_categories,
        rule_canonical_prefixes,
        rule_offsets,
        r_lens,
        serovar_keys,
        serovar_targets,
        serovar_offsets,
        s_lens,
        landmark_starts,
        landmark_ends,
        landmark_offsets,
        l_lens,
        known_serotype_values,
        known_serotype_offsets,
        ks_lens,
    ):
        arr.setflags(write=False)

    return cls(
        names=names,
        ncbi_taxids=taxids,
        gram_stains=grams,
        default_categories=cats,
        archetype_masks=masks,
        alias_values=alias_values,
        offsets=offsets,
        lengths=alias_lens,
        rule_prefixes=rule_prefixes,
        rule_categories=rule_categories,
        rule_canonical_prefixes=rule_canonical_prefixes,
        rule_offsets=rule_offsets,
        rule_lengths=r_lens,
        serovar_keys=serovar_keys,
        serovar_targets=serovar_targets,
        serovar_offsets=serovar_offsets,
        serovar_lengths=s_lens,
        landmark_starts=landmark_starts,
        landmark_ends=landmark_ends,
        landmark_offsets=landmark_offsets,
        landmark_lengths=l_lens,
        known_serotype_values=known_serotype_values,
        known_serotype_offsets=known_serotype_offsets,
        known_serotype_lengths=ks_lens,
    )

concat `classmethod` ¶

concat(batches: Iterable[Self]) -> Self

Concatenate multiple SpeciesSpecs containers.

Source code in src/seroref/db/species.py

@classmethod
def concat(cls, batches: Iterable[Self]) -> Self:
    """Concatenate multiple `SpeciesSpecs` containers."""
    items: list[SpeciesSpec] = []
    for b in batches:
        items.extend(b)
    return cls.from_list(items)

BacterialSpecies ¶

Bases: IntEnum

Canonical bacterial species ontology (0..51) for SeroRef reference loci and strains.

Follows the same data-oriented IntEnum + frozen SpeciesSpec + SpeciesSpecs pattern as LocusArchetype and GeneRole.

Methods:

Name	Description
`parse_serotype`	Parse and normalize a single or multi-antigen serotype formula using this species' grammar.
`normalize_strain`	Normalize a strain isolate designation for this species.
`from_name`	Resolve a `BacterialSpecies` by enum ID, scientific name, enum name, or taxonomic alias.
`resolve`	Resolve a species identifier, falling back to `default` (or `BacterialSpecies.UNKNOWN`).
`from_taxid`	Resolve a `BacterialSpecies` from an NCBI Taxonomy ID, or `BacterialSpecies.UNKNOWN`.
`specs`	Return the cached singleton `SpeciesSpecs` SoA container over all `BacterialSpecies`.

Attributes:

Name	Type	Description
`spec`	`SpeciesSpec`	Return the immutable `SpeciesSpec` metadata for this species.
`name_bytes`	`bytes`	Canonical scientific name as ASCII bytes (e.g. `b"Escherichia coli"`).
`scientific_name`	`bytes`	Canonical scientific name as ASCII bytes (e.g. `b"Escherichia coli"`).
`name_str`	`str`	Canonical scientific name as UTF-8 `str`.
`scientific_name_str`	`str`	Canonical scientific name as UTF-8 `str`.
`ncbi_taxid`	`int`	Species-level NCBI Taxonomy ID.
`gram_stain`	`GramStain`	Gram stain classification (`GramStain.NEGATIVE` or `GramStain.POSITIVE`).
`default_category`	`ArchetypeCategory`	Default surface polysaccharide `ArchetypeCategory` when serotype tokens lack an explicit prefix.
`archetype_mask`	`int`	32-bit bitmask of `LocusArchetype`s compatible with this species.
`aliases`	`tuple[bytes, ...]`	Taxonomic synonyms and genus aliases for this species.
`synonyms`	`tuple[bytes, ...]`	Taxonomic synonyms and genus aliases for this species (alias for `self.aliases`).
`landmark_accessions`	`tuple[tuple[bytes, bytes], ...]`	Landmark INSDC BGC accession intervals for this species.

Source code in src/seroref/db/species.py

class BacterialSpecies(IntEnum):
    """Canonical bacterial species ontology (`0..51`) for `SeroRef` reference loci and strains.

    Follows the same data-oriented `IntEnum` + frozen `SpeciesSpec` + `SpeciesSpecs` pattern as
    [`LocusArchetype`][seroref.loci.archetypes.LocusArchetype] and [`GeneRole`][seroref.genes.roles.GeneRole].
    """

    ACINETOBACTER_BAUMANNII = 0
    ACINETOBACTER_CALCOACETICUS = 1
    ACTINOBACILLUS_PLEUROPNEUMONIAE = 2
    AEROMONAS_HYDROPHILA = 3
    AEROMONAS_SALMONICIDA = 4
    AGGREGATIBACTER_ACTINOMYCETEMCOMITANS = 5
    BACILLUS_CEREUS = 6
    BACTEROIDES_FRAGILIS = 7
    BURKHOLDERIA_CEPACIA = 8
    BURKHOLDERIA_MULTIVORANS = 9
    CAMPYLOBACTER_JEJUNI = 10
    CITROBACTER_BRAAKII = 11
    CITROBACTER_FREUNDII = 12
    CITROBACTER_WERKMANII = 13
    CRONOBACTER_DUBLINENSIS = 14
    CRONOBACTER_MALONATICUS = 15
    CRONOBACTER_SAKAZAKII = 16
    CRONOBACTER_TURICENSIS = 17
    ENTEROBACTER_CLOACAE = 18
    ENTEROCOCCUS_FAECALIS = 19
    ESCHERICHIA_ALBERTII = 20
    ESCHERICHIA_COLI = 21
    HAEMOPHILUS_INFLUENZAE = 22
    HELICOBACTER_PYLORI = 23
    KLEBSIELLA_PNEUMONIAE = 24
    LACTOBACILLUS_DELBRUECKII = 25
    NEISSERIA_MENINGITIDIS = 26
    PLESIOMONAS_SHIGELLOIDES = 27
    PROTEUS_MIRABILIS = 28
    PROTEUS_MYXOFACIENS = 29
    PROTEUS_PENNERI = 30
    PROTEUS_VULGARIS = 31
    PROVIDENCIA_RUSTIGIANII = 32
    PSEUDOMONAS_AERUGINOSA = 33
    RAOULTELLA_TERRIGENA = 34
    SALMONELLA_ENTERICA = 35
    SHIGELLA_BOYDII = 36
    SHIGELLA_DYSENTERIAE = 37
    SHIGELLA_FLEXNERI = 38
    SHIGELLA_SONNEI = 39
    STAPHYLOCOCCUS_AUREUS = 40
    STREPTOCOCCUS_ORALIS = 41
    STREPTOCOCCUS_PNEUMONIAE = 42
    STREPTOCOCCUS_SUIS = 43
    VIBRIO_CHOLERAE = 44
    YERSINIA_BERCOVIERI = 45
    YERSINIA_ENTEROCOLITICA = 46
    YERSINIA_FREDERIKSENII = 47
    YERSINIA_KRISTENSENII = 48
    YERSINIA_PSEUDOTUBERCULOSIS = 49
    YERSINIA_SIMILIS = 50
    UNKNOWN = 51

    @property
    def spec(self) -> SpeciesSpec:
        """Return the immutable `SpeciesSpec` metadata for this species."""
        return _get_species_specs()[int(self)]

    @property
    def name_bytes(self) -> bytes:
        """Canonical scientific name as ASCII bytes (e.g. `b"Escherichia coli"`)."""
        return self.spec.name

    @property
    def scientific_name(self) -> bytes:
        """Canonical scientific name as ASCII bytes (e.g. `b"Escherichia coli"`)."""
        return self.spec.name

    @property
    def name_str(self) -> str:
        """Canonical scientific name as UTF-8 `str`."""
        return self.spec.name_str

    @property
    def scientific_name_str(self) -> str:
        """Canonical scientific name as UTF-8 `str`."""
        return self.spec.name_str

    @property
    def ncbi_taxid(self) -> int:
        """Species-level NCBI Taxonomy ID."""
        return self.spec.ncbi_taxid

    @property
    def gram_stain(self) -> GramStain:
        """Gram stain classification (`GramStain.NEGATIVE` or `GramStain.POSITIVE`)."""
        return self.spec.gram_stain

    @property
    def default_category(self) -> ArchetypeCategory:
        """Default surface polysaccharide `ArchetypeCategory` when serotype tokens lack an explicit prefix."""
        return self.spec.default_category

    @property
    def archetype_mask(self) -> int:
        """32-bit bitmask of `LocusArchetype`s compatible with this species."""
        return self.spec.archetype_mask

    @property
    def aliases(self) -> tuple[bytes, ...]:
        """Taxonomic synonyms and genus aliases for this species."""
        return self.spec.aliases

    @property
    def synonyms(self) -> tuple[bytes, ...]:
        """Taxonomic synonyms and genus aliases for this species (alias for `self.aliases`)."""
        return self.spec.aliases

    @property
    def landmark_accessions(self) -> tuple[tuple[bytes, bytes], ...]:
        """Landmark INSDC BGC accession intervals for this species."""
        return self.spec.landmark_accessions

    def parse_serotype(
        self,
        raw: str | bytes,
        *,
        archetype: LocusArchetype | None = None,
    ) -> tuple[tuple[ArchetypeCategory, bytes], ...]:
        """Parse and normalize a single or multi-antigen serotype formula using this species' grammar."""
        return self.spec.parse_serotype(raw, archetype=archetype)

    def normalize_strain(
        self,
        raw_strain: str | bytes,
        raw_serotype: str | bytes = b"",
    ) -> bytes:
        """Normalize a strain isolate designation for this species."""
        return self.spec.normalize_strain(raw_strain, raw_serotype)

    @classmethod
    def from_name(cls, key: str | bytes | int | BacterialSpecies) -> BacterialSpecies:
        """Resolve a `BacterialSpecies` by enum ID, scientific name, enum name, or taxonomic alias."""
        if isinstance(key, cls):
            return key
        if isinstance(key, (int, np.integer)):
            return cls(int(key))
        b = key.encode("utf-8") if isinstance(key, str) else bytes(key)
        b_clean = b.split(b";", 1)[0].strip()
        lookup = _get_species_lookup()
        low = b_clean.lower()
        if low in lookup:
            return lookup[low]
        parts = b_clean.split()
        if len(parts) >= 2:
            binomial = (b"%s %s" % (parts[0], parts[1])).lower()
            if binomial in lookup:
                return lookup[binomial]
        raise KeyError(f"Unknown BacterialSpecies identifier: {key!r}")

    @classmethod
    def resolve(
        cls,
        key: str | bytes | int | BacterialSpecies,
        default: BacterialSpecies | None = None,
    ) -> BacterialSpecies:
        """Resolve a species identifier, falling back to `default` (or `BacterialSpecies.UNKNOWN`)."""
        fallback = cls.UNKNOWN if default is None else default
        try:
            return cls.from_name(key)
        except KeyError, ValueError:
            return fallback

    @classmethod
    def from_taxid(cls, taxid: int) -> BacterialSpecies:
        """Resolve a `BacterialSpecies` from an NCBI Taxonomy ID, or `BacterialSpecies.UNKNOWN`."""
        return _get_taxid_lookup().get(int(taxid), cls.UNKNOWN)

    @classmethod
    def specs(cls) -> SpeciesSpecs:
        """Return the cached singleton `SpeciesSpecs` SoA container over all `BacterialSpecies`."""
        return _get_species_specs()

spec `property` ¶

spec: SpeciesSpec

Return the immutable SpeciesSpec metadata for this species.

name_bytes `property` ¶

name_bytes: bytes

Canonical scientific name as ASCII bytes (e.g. b"Escherichia coli").

scientific_name `property` ¶

scientific_name: bytes

Canonical scientific name as ASCII bytes (e.g. b"Escherichia coli").

name_str `property` ¶

name_str: str

Canonical scientific name as UTF-8 str.

scientific_name_str `property` ¶

scientific_name_str: str

Canonical scientific name as UTF-8 str.

ncbi_taxid `property` ¶

ncbi_taxid: int

Species-level NCBI Taxonomy ID.

gram_stain `property` ¶

gram_stain: GramStain

Gram stain classification (GramStain.NEGATIVE or GramStain.POSITIVE).

default_category `property` ¶

default_category: ArchetypeCategory

Default surface polysaccharide ArchetypeCategory when serotype tokens lack an explicit prefix.

archetype_mask `property` ¶

archetype_mask: int

32-bit bitmask of LocusArchetypes compatible with this species.

aliases `property` ¶

aliases: tuple[bytes, ...]

Taxonomic synonyms and genus aliases for this species.

synonyms `property` ¶

synonyms: tuple[bytes, ...]

Taxonomic synonyms and genus aliases for this species (alias for self.aliases).

landmark_accessions `property` ¶

landmark_accessions: tuple[tuple[bytes, bytes], ...]

Landmark INSDC BGC accession intervals for this species.

parse_serotype ¶

parse_serotype(
    raw: str | bytes,
    *,
    archetype: LocusArchetype | None = None,
) -> tuple[tuple[ArchetypeCategory, bytes], ...]

Parse and normalize a single or multi-antigen serotype formula using this species' grammar.

Source code in src/seroref/db/species.py

def parse_serotype(
    self,
    raw: str | bytes,
    *,
    archetype: LocusArchetype | None = None,
) -> tuple[tuple[ArchetypeCategory, bytes], ...]:
    """Parse and normalize a single or multi-antigen serotype formula using this species' grammar."""
    return self.spec.parse_serotype(raw, archetype=archetype)

normalize_strain ¶

normalize_strain(
    raw_strain: str | bytes, raw_serotype: str | bytes = b""
) -> bytes

Normalize a strain isolate designation for this species.

Source code in src/seroref/db/species.py

def normalize_strain(
    self,
    raw_strain: str | bytes,
    raw_serotype: str | bytes = b"",
) -> bytes:
    """Normalize a strain isolate designation for this species."""
    return self.spec.normalize_strain(raw_strain, raw_serotype)

from_name `classmethod` ¶

from_name(
    key: str | bytes | int | BacterialSpecies,
) -> BacterialSpecies

Resolve a BacterialSpecies by enum ID, scientific name, enum name, or taxonomic alias.

Source code in src/seroref/db/species.py

@classmethod
def from_name(cls, key: str | bytes | int | BacterialSpecies) -> BacterialSpecies:
    """Resolve a `BacterialSpecies` by enum ID, scientific name, enum name, or taxonomic alias."""
    if isinstance(key, cls):
        return key
    if isinstance(key, (int, np.integer)):
        return cls(int(key))
    b = key.encode("utf-8") if isinstance(key, str) else bytes(key)
    b_clean = b.split(b";", 1)[0].strip()
    lookup = _get_species_lookup()
    low = b_clean.lower()
    if low in lookup:
        return lookup[low]
    parts = b_clean.split()
    if len(parts) >= 2:
        binomial = (b"%s %s" % (parts[0], parts[1])).lower()
        if binomial in lookup:
            return lookup[binomial]
    raise KeyError(f"Unknown BacterialSpecies identifier: {key!r}")

resolve `classmethod` ¶

resolve(
    key: str | bytes | int | BacterialSpecies,
    default: BacterialSpecies | None = None,
) -> BacterialSpecies

Resolve a species identifier, falling back to default (or BacterialSpecies.UNKNOWN).

Source code in src/seroref/db/species.py

@classmethod
def resolve(
    cls,
    key: str | bytes | int | BacterialSpecies,
    default: BacterialSpecies | None = None,
) -> BacterialSpecies:
    """Resolve a species identifier, falling back to `default` (or `BacterialSpecies.UNKNOWN`)."""
    fallback = cls.UNKNOWN if default is None else default
    try:
        return cls.from_name(key)
    except KeyError, ValueError:
        return fallback

from_taxid `classmethod` ¶

from_taxid(taxid: int) -> BacterialSpecies

Resolve a BacterialSpecies from an NCBI Taxonomy ID, or BacterialSpecies.UNKNOWN.

Source code in src/seroref/db/species.py

@classmethod
def from_taxid(cls, taxid: int) -> BacterialSpecies:
    """Resolve a `BacterialSpecies` from an NCBI Taxonomy ID, or `BacterialSpecies.UNKNOWN`."""
    return _get_taxid_lookup().get(int(taxid), cls.UNKNOWN)

specs `classmethod` ¶

specs() -> SpeciesSpecs

Return the cached singleton SpeciesSpecs SoA container over all BacterialSpecies.

Source code in src/seroref/db/species.py

@classmethod
def specs(cls) -> SpeciesSpecs:
    """Return the cached singleton `SpeciesSpecs` SoA container over all `BacterialSpecies`."""
    return _get_species_specs()

archetypes_to_mask ¶

archetypes_to_mask(
    archetypes: Iterable[int | LocusArchetype],
) -> int

Pack a collection of LocusArchetype values into a 32-bit integer presence bitmask.

Source code in src/seroref/db/species.py

def archetypes_to_mask(archetypes: Iterable[int | LocusArchetype]) -> int:
    """Pack a collection of `LocusArchetype` values into a 32-bit integer presence bitmask."""
    mask = 0
    for arch in archetypes:
        val = int(arch)
        if 0 <= val < len(LocusArchetype):
            mask |= 1 << val
    return mask

mask_to_archetypes ¶

mask_to_archetypes(mask: int) -> tuple[LocusArchetype, ...]

Unpack a 32-bit integer presence bitmask into a sorted tuple of LocusArchetype enums.

Source code in src/seroref/db/species.py

def mask_to_archetypes(mask: int) -> tuple[LocusArchetype, ...]:
    """Unpack a 32-bit integer presence bitmask into a sorted tuple of `LocusArchetype` enums."""
    return tuple(arch for arch in LocusArchetype if (mask & (1 << int(arch))) != 0)

Scalar Views (`seroref.db.models`)¶

models ¶

OOP scalar data models for ReferenceStrain and SeroRefRecord.

Classes:

Name	Description
`ReferenceStrain`	Immutable scalar view of a reference bacterial strain isolate and genome assembly.
`SeroRefRecord`	Immutable scalar view of a reference bacterial antigen locus record in `SeroRefRecords`.

Functions:

Name	Description
`archetype_to_antigen_type`	Map a `LocusArchetype` to its canonical surface polysaccharide antigen class (`CPS`, `O-antigen`, `EPS`, `WTA`).

ReferenceStrain `dataclass` ¶

Bases: ScalarView

Immutable scalar view of a reference bacterial strain isolate and genome assembly.

Attributes:

Name	Type	Description
`species`	`BacterialSpecies`	Canonical `BacterialSpecies` enum member.
`name`	`bytes`	Normalized primary reference strain designation (e.g. `b"NCTC 9001"`, `b"ATCC 25238"`).
`synonyms`	`tuple[bytes, ...]`	Culture-collection and laboratory strain equivalence clique from DSMZ StrainInfo v2.
`assembly_id`	`bytes`	INSDC genome assembly or nucleotide accession (e.g. `b"GCA_900706755.1"`).
`ncbi_taxid`	`int`	Strain-specific or species-level NCBI Taxonomy ID integer.
`straininfo_id`	`int`	DSMZ StrainInfo `SI-ID` integer, or `0` if unknown.
`bacdive_id`	`int`	DSMZ BacDive ID integer, or `0` if unknown.
`pmids`	`tuple[int, ...]`	Tuple of PubMed IDs associated with this reference strain or assembly.

Methods:

Name	Description
`from_raw`	Construct a normalized `ReferenceStrain` using `BacterialSpecies` rules.
`with_passport`	Enrich this `ReferenceStrain` with culture-collection synonyms and IDs from a `StrainPassport`.
`merge`	Merge metadata from another `ReferenceStrain` describing the same isolate or assembly.
`bind_loci`	Bind a batch of predicted `Loci` on this reference assembly into `SeroRefRecord` rows.

Source code in src/seroref/db/models.py

@dataclass(frozen=True, slots=True)
class ReferenceStrain(ScalarView):
    """Immutable scalar view of a reference bacterial strain isolate and genome assembly.

    Attributes:
        species: Canonical [`BacterialSpecies`][seroref.db.species.BacterialSpecies] enum member.
        name: Normalized primary reference strain designation (e.g. `b"NCTC 9001"`, `b"ATCC 25238"`).
        synonyms: Culture-collection and laboratory strain equivalence clique from DSMZ StrainInfo v2.
        assembly_id: INSDC genome assembly or nucleotide accession (e.g. `b"GCA_900706755.1"`).
        ncbi_taxid: Strain-specific or species-level NCBI Taxonomy ID integer.
        straininfo_id: DSMZ StrainInfo `SI-ID` integer, or `0` if unknown.
        bacdive_id: DSMZ BacDive ID integer, or `0` if unknown.
        pmids: Tuple of PubMed IDs associated with this reference strain or assembly.
    """

    species: BacterialSpecies
    name: bytes = b""
    synonyms: tuple[bytes, ...] = ()
    assembly_id: bytes = b""
    ncbi_taxid: int = 0
    straininfo_id: int = 0
    bacdive_id: int = 0
    pmids: tuple[int, ...] = ()

    @property
    def organism(self) -> bytes:
        """Canonical scientific name byte string from `self.species`."""
        return self.species.scientific_name

    @property
    def effective_taxid(self) -> int:
        """Resolved NCBI Taxonomy ID, falling back to `self.species.ncbi_taxid` when `self.ncbi_taxid == 0`."""
        return self.ncbi_taxid if self.ncbi_taxid > 0 else self.species.ncbi_taxid

    @property
    def compound_key(self) -> bytes:
        """Unique composite strain-assembly key `b"<organism>|<name>|<assembly_id>"`."""
        return b"%s|%s|%s" % (self.organism, self.name, self.assembly_id)

    @classmethod
    def from_raw(
        cls,
        organism: str | bytes | int | BacterialSpecies,
        strain: str | bytes = b"",
        *,
        raw_serotype: str | bytes = b"",
        synonyms: Sequence[bytes | str] = (),
        assembly_id: bytes | str = b"",
        ncbi_taxid: int = 0,
        straininfo_id: int = 0,
        bacdive_id: int = 0,
        pmids: Sequence[int] = (),
    ) -> Self:
        """Construct a normalized `ReferenceStrain` using `BacterialSpecies` rules."""
        sp = BacterialSpecies.resolve(organism)
        if sp == BacterialSpecies.UNKNOWN and ncbi_taxid > 0:
            sp = BacterialSpecies.from_taxid(ncbi_taxid)
        norm_name = sp.normalize_strain(strain, raw_serotype)
        asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else bytes(assembly_id)
        syn_list: list[bytes] = []
        if norm_name:
            syn_list.append(norm_name)
        for s in synonyms:
            sb = s.encode("utf-8").strip() if isinstance(s, str) else bytes(s).strip()
            if sb and sb not in syn_list:
                syn_list.append(sb)
        pm_tuple = tuple(dict.fromkeys(int(p) for p in pmids if int(p) > 0))
        return cls(
            species=sp,
            name=norm_name,
            synonyms=tuple(syn_list),
            assembly_id=asm_b,
            ncbi_taxid=int(ncbi_taxid) if int(ncbi_taxid) > 0 else sp.ncbi_taxid,
            straininfo_id=max(0, int(straininfo_id)),
            bacdive_id=max(0, int(bacdive_id)),
            pmids=pm_tuple,
        )

    def with_passport(self, passport: Any, *, raw_serotype: str | bytes = b"") -> Self:
        """Enrich this `ReferenceStrain` with culture-collection synonyms and IDs from a `StrainPassport`."""
        if passport is None:
            return self
        des_bytes = [d.encode("utf-8") for d in getattr(passport, "designations", ()) if d]
        norm_name = self.name
        if not norm_name and des_bytes:
            norm_name = self.species.normalize_strain(des_bytes[0], raw_serotype)
        syns = tuple(dict.fromkeys((*((norm_name,) if norm_name else ()), *self.synonyms, *des_bytes)))
        p_pmids = [int(p) for p in getattr(passport, "pmids", ()) if int(p) > 0]
        merged_pmids = tuple(dict.fromkeys((*self.pmids, *p_pmids)))
        p_tax = int(getattr(passport, "ncbi_taxid", 0) or 0)
        p_si = int(getattr(passport, "si_id", 0) or 0)
        p_bd = int(getattr(passport, "bacdive_id", 0) or 0)
        return type(self)(
            species=self.species,
            name=norm_name,
            synonyms=syns,
            assembly_id=self.assembly_id,
            ncbi_taxid=self.ncbi_taxid or p_tax or self.species.ncbi_taxid,
            straininfo_id=self.straininfo_id or p_si,
            bacdive_id=self.bacdive_id or p_bd,
            pmids=merged_pmids,
        )

    def merge(self, other: ReferenceStrain) -> Self:
        """Merge metadata from another `ReferenceStrain` describing the same isolate or assembly."""
        sp = self.species if self.species != BacterialSpecies.UNKNOWN else other.species
        name = self.name or other.name
        syns = tuple(dict.fromkeys((*self.synonyms, *other.synonyms)))
        pmids = tuple(dict.fromkeys((*self.pmids, *other.pmids)))
        return type(self)(
            species=sp,
            name=name,
            synonyms=syns,
            assembly_id=self.assembly_id or other.assembly_id,
            ncbi_taxid=self.ncbi_taxid or other.ncbi_taxid or sp.ncbi_taxid,
            straininfo_id=self.straininfo_id or other.straininfo_id,
            bacdive_id=self.bacdive_id or other.bacdive_id,
            pmids=pmids,
        )

    def bind_loci(
        self,
        loci: Loci,
        raw_serotype: str | bytes = b"",
    ) -> list[SeroRefRecord]:
        """Bind a batch of predicted `Loci` on this reference assembly into `SeroRefRecord` rows.

        Splits multi-antigen formulas (e.g. `'O1:K1(L1):H7'`) across predicted `ArchetypeCategory`
        groups (`O-Antigen`, `Capsule`, `Exopolysaccharide`, `Other`), assigns invariant glycan names
        (`Pel`, `PNAG`, `Alginate`, `Cellulose`) to exopolysaccharide archetypes, and synchronizes
        `locus.name` with the resolved `serotype`.
        """
        if len(loci) == 0:
            return []

        strain_rec = (
            self
            if self.name or not raw_serotype
            else self.from_raw(
                self.species,
                self.name,
                raw_serotype=raw_serotype,
                synonyms=self.synonyms,
                assembly_id=self.assembly_id,
                ncbi_taxid=self.ncbi_taxid,
                straininfo_id=self.straininfo_id,
                bacdive_id=self.bacdive_id,
                pmids=self.pmids,
            )
        )

        # Keep highest-scoring locus per (ArchetypeCategory, LocusArchetype, start, end)
        best_by_cat: dict[ArchetypeCategory, list[Locus]] = defaultdict(list)
        seen_intervals: set[tuple[LocusArchetype, int, int]] = set()
        for i in range(len(loci)):
            loc = loci[i].canonicalize_strand()
            if loc.archetype == LocusArchetype.UNKNOWN or len(loc.nodes) == 0:
                continue
            if not strain_rec.species.spec.supports_archetype(loc.archetype):
                continue
            key = (loc.archetype, int(loc.start), int(loc.end))
            if key in seen_intervals:
                continue
            seen_intervals.add(key)
            best_by_cat[loc.archetype.category].append(loc)

        for cat_loci in best_by_cat.values():
            cat_loci.sort(key=lambda loc: (float(loc.completeness), float(loc.score), len(loc.nodes)), reverse=True)

        # Parse multi-antigen or single-antigen serotype tokens using species grammar
        primary_arch = next(
            (locs[0].archetype for locs in best_by_cat.values() if locs),
            LocusArchetype.UNKNOWN,
        )
        token_by_cat: dict[ArchetypeCategory, bytes] = {}
        for cat, tok_b in strain_rec.species.parse_serotype(raw_serotype, archetype=primary_arch):
            token_by_cat.setdefault(cat, tok_b)

        out: list[SeroRefRecord] = []
        for cat, cat_loci in best_by_cat.items():
            matched_sero = token_by_cat.get(cat, b"")
            for rank, loc in enumerate(cat_loci):
                sero_for_loc = matched_sero if rank == 0 else b""
                if not sero_for_loc and loc.archetype in INVARIANT_ARCHETYPE_GLYCAN_NAMES:
                    sero_for_loc = INVARIANT_ARCHETYPE_GLYCAN_NAMES[loc.archetype]
                if sero_for_loc:
                    parsed = strain_rec.species.parse_serotype(sero_for_loc, archetype=loc.archetype)
                    if parsed:
                        sero_for_loc = parsed[0][1]
                loc_named = loc.with_name(sero_for_loc)
                out.append(
                    SeroRefRecord(
                        strain_record=strain_rec,
                        serotype=sero_for_loc,
                        locus=loc_named,
                    )
                )
        return out

organism `property` ¶

organism: bytes

Canonical scientific name byte string from self.species.

effective_taxid `property` ¶

effective_taxid: int

Resolved NCBI Taxonomy ID, falling back to self.species.ncbi_taxid when self.ncbi_taxid == 0.

compound_key `property` ¶

compound_key: bytes

Unique composite strain-assembly key b"<organism>|<name>|<assembly_id>".

from_raw `classmethod` ¶

from_raw(
    organism: str | bytes | int | BacterialSpecies,
    strain: str | bytes = b"",
    *,
    raw_serotype: str | bytes = b"",
    synonyms: Sequence[bytes | str] = (),
    assembly_id: bytes | str = b"",
    ncbi_taxid: int = 0,
    straininfo_id: int = 0,
    bacdive_id: int = 0,
    pmids: Sequence[int] = (),
) -> Self

Construct a normalized ReferenceStrain using BacterialSpecies rules.

Source code in src/seroref/db/models.py

@classmethod
def from_raw(
    cls,
    organism: str | bytes | int | BacterialSpecies,
    strain: str | bytes = b"",
    *,
    raw_serotype: str | bytes = b"",
    synonyms: Sequence[bytes | str] = (),
    assembly_id: bytes | str = b"",
    ncbi_taxid: int = 0,
    straininfo_id: int = 0,
    bacdive_id: int = 0,
    pmids: Sequence[int] = (),
) -> Self:
    """Construct a normalized `ReferenceStrain` using `BacterialSpecies` rules."""
    sp = BacterialSpecies.resolve(organism)
    if sp == BacterialSpecies.UNKNOWN and ncbi_taxid > 0:
        sp = BacterialSpecies.from_taxid(ncbi_taxid)
    norm_name = sp.normalize_strain(strain, raw_serotype)
    asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else bytes(assembly_id)
    syn_list: list[bytes] = []
    if norm_name:
        syn_list.append(norm_name)
    for s in synonyms:
        sb = s.encode("utf-8").strip() if isinstance(s, str) else bytes(s).strip()
        if sb and sb not in syn_list:
            syn_list.append(sb)
    pm_tuple = tuple(dict.fromkeys(int(p) for p in pmids if int(p) > 0))
    return cls(
        species=sp,
        name=norm_name,
        synonyms=tuple(syn_list),
        assembly_id=asm_b,
        ncbi_taxid=int(ncbi_taxid) if int(ncbi_taxid) > 0 else sp.ncbi_taxid,
        straininfo_id=max(0, int(straininfo_id)),
        bacdive_id=max(0, int(bacdive_id)),
        pmids=pm_tuple,
    )

with_passport ¶

with_passport(
    passport: Any, *, raw_serotype: str | bytes = b""
) -> Self

Enrich this ReferenceStrain with culture-collection synonyms and IDs from a StrainPassport.

Source code in src/seroref/db/models.py

def with_passport(self, passport: Any, *, raw_serotype: str | bytes = b"") -> Self:
    """Enrich this `ReferenceStrain` with culture-collection synonyms and IDs from a `StrainPassport`."""
    if passport is None:
        return self
    des_bytes = [d.encode("utf-8") for d in getattr(passport, "designations", ()) if d]
    norm_name = self.name
    if not norm_name and des_bytes:
        norm_name = self.species.normalize_strain(des_bytes[0], raw_serotype)
    syns = tuple(dict.fromkeys((*((norm_name,) if norm_name else ()), *self.synonyms, *des_bytes)))
    p_pmids = [int(p) for p in getattr(passport, "pmids", ()) if int(p) > 0]
    merged_pmids = tuple(dict.fromkeys((*self.pmids, *p_pmids)))
    p_tax = int(getattr(passport, "ncbi_taxid", 0) or 0)
    p_si = int(getattr(passport, "si_id", 0) or 0)
    p_bd = int(getattr(passport, "bacdive_id", 0) or 0)
    return type(self)(
        species=self.species,
        name=norm_name,
        synonyms=syns,
        assembly_id=self.assembly_id,
        ncbi_taxid=self.ncbi_taxid or p_tax or self.species.ncbi_taxid,
        straininfo_id=self.straininfo_id or p_si,
        bacdive_id=self.bacdive_id or p_bd,
        pmids=merged_pmids,
    )

merge ¶

merge(other: ReferenceStrain) -> Self

Merge metadata from another ReferenceStrain describing the same isolate or assembly.

Source code in src/seroref/db/models.py

def merge(self, other: ReferenceStrain) -> Self:
    """Merge metadata from another `ReferenceStrain` describing the same isolate or assembly."""
    sp = self.species if self.species != BacterialSpecies.UNKNOWN else other.species
    name = self.name or other.name
    syns = tuple(dict.fromkeys((*self.synonyms, *other.synonyms)))
    pmids = tuple(dict.fromkeys((*self.pmids, *other.pmids)))
    return type(self)(
        species=sp,
        name=name,
        synonyms=syns,
        assembly_id=self.assembly_id or other.assembly_id,
        ncbi_taxid=self.ncbi_taxid or other.ncbi_taxid or sp.ncbi_taxid,
        straininfo_id=self.straininfo_id or other.straininfo_id,
        bacdive_id=self.bacdive_id or other.bacdive_id,
        pmids=pmids,
    )

bind_loci ¶

bind_loci(
    loci: Loci, raw_serotype: str | bytes = b""
) -> list[SeroRefRecord]

Bind a batch of predicted Loci on this reference assembly into SeroRefRecord rows.

Splits multi-antigen formulas (e.g. 'O1:K1(L1):H7') across predicted ArchetypeCategory groups (O-Antigen, Capsule, Exopolysaccharide, Other), assigns invariant glycan names (Pel, PNAG, Alginate, Cellulose) to exopolysaccharide archetypes, and synchronizes locus.name with the resolved serotype.

Source code in src/seroref/db/models.py

def bind_loci(
    self,
    loci: Loci,
    raw_serotype: str | bytes = b"",
) -> list[SeroRefRecord]:
    """Bind a batch of predicted `Loci` on this reference assembly into `SeroRefRecord` rows.

    Splits multi-antigen formulas (e.g. `'O1:K1(L1):H7'`) across predicted `ArchetypeCategory`
    groups (`O-Antigen`, `Capsule`, `Exopolysaccharide`, `Other`), assigns invariant glycan names
    (`Pel`, `PNAG`, `Alginate`, `Cellulose`) to exopolysaccharide archetypes, and synchronizes
    `locus.name` with the resolved `serotype`.
    """
    if len(loci) == 0:
        return []

    strain_rec = (
        self
        if self.name or not raw_serotype
        else self.from_raw(
            self.species,
            self.name,
            raw_serotype=raw_serotype,
            synonyms=self.synonyms,
            assembly_id=self.assembly_id,
            ncbi_taxid=self.ncbi_taxid,
            straininfo_id=self.straininfo_id,
            bacdive_id=self.bacdive_id,
            pmids=self.pmids,
        )
    )

    # Keep highest-scoring locus per (ArchetypeCategory, LocusArchetype, start, end)
    best_by_cat: dict[ArchetypeCategory, list[Locus]] = defaultdict(list)
    seen_intervals: set[tuple[LocusArchetype, int, int]] = set()
    for i in range(len(loci)):
        loc = loci[i].canonicalize_strand()
        if loc.archetype == LocusArchetype.UNKNOWN or len(loc.nodes) == 0:
            continue
        if not strain_rec.species.spec.supports_archetype(loc.archetype):
            continue
        key = (loc.archetype, int(loc.start), int(loc.end))
        if key in seen_intervals:
            continue
        seen_intervals.add(key)
        best_by_cat[loc.archetype.category].append(loc)

    for cat_loci in best_by_cat.values():
        cat_loci.sort(key=lambda loc: (float(loc.completeness), float(loc.score), len(loc.nodes)), reverse=True)

    # Parse multi-antigen or single-antigen serotype tokens using species grammar
    primary_arch = next(
        (locs[0].archetype for locs in best_by_cat.values() if locs),
        LocusArchetype.UNKNOWN,
    )
    token_by_cat: dict[ArchetypeCategory, bytes] = {}
    for cat, tok_b in strain_rec.species.parse_serotype(raw_serotype, archetype=primary_arch):
        token_by_cat.setdefault(cat, tok_b)

    out: list[SeroRefRecord] = []
    for cat, cat_loci in best_by_cat.items():
        matched_sero = token_by_cat.get(cat, b"")
        for rank, loc in enumerate(cat_loci):
            sero_for_loc = matched_sero if rank == 0 else b""
            if not sero_for_loc and loc.archetype in INVARIANT_ARCHETYPE_GLYCAN_NAMES:
                sero_for_loc = INVARIANT_ARCHETYPE_GLYCAN_NAMES[loc.archetype]
            if sero_for_loc:
                parsed = strain_rec.species.parse_serotype(sero_for_loc, archetype=loc.archetype)
                if parsed:
                    sero_for_loc = parsed[0][1]
            loc_named = loc.with_name(sero_for_loc)
            out.append(
                SeroRefRecord(
                    strain_record=strain_rec,
                    serotype=sero_for_loc,
                    locus=loc_named,
                )
            )
    return out

SeroRefRecord `dataclass` ¶

Bases: ScalarView

Immutable scalar view of a reference bacterial antigen locus record in SeroRefRecords.

Attributes:

Name	Type	Description
`strain_record`	`ReferenceStrain`	Parent `ReferenceStrain` isolate and assembly record.
`serotype`	`bytes`	Canonical serotype or glycan name (e.g. `b"O1"`, `b"K1"`, `b"35B"`, `b"Pel"`, `b"PNAG"`).
`locus`	`Locus \| None`	Predicted `Locus` biosynthetic gene cluster.

Methods:

Name	Description
`from_fields`	Construct a `SeroRefRecord` from flat organism, strain, serotype, and locus fields.
`merge`	Merge metadata from another `SeroRefRecord` describing the same physical locus.

Source code in src/seroref/db/models.py

@dataclass(frozen=True, slots=True)
class SeroRefRecord(ScalarView):
    """Immutable scalar view of a reference bacterial antigen locus record in `SeroRefRecords`.

    Attributes:
        strain_record: Parent [`ReferenceStrain`][seroref.db.models.ReferenceStrain] isolate and assembly record.
        serotype: Canonical serotype or glycan name (e.g. `b"O1"`, `b"K1"`, `b"35B"`, `b"Pel"`, `b"PNAG"`).
        locus: Predicted [`Locus`][seroref.loci.models.Locus] biosynthetic gene cluster.
    """

    strain_record: ReferenceStrain
    serotype: bytes = b""
    locus: Locus | None = None

    @classmethod
    def from_fields(
        cls,
        organism: str | bytes | int | BacterialSpecies,
        *,
        serotype: str | bytes = b"",
        strain: str | bytes = b"",
        strain_synonyms: Sequence[bytes | str] = (),
        assembly_id: bytes | str = b"",
        locus: Locus | None = None,
        ncbi_taxid: int = 0,
        straininfo_id: int = 0,
        bacdive_id: int = 0,
        pmids: Sequence[int] = (),
    ) -> Self:
        """Construct a `SeroRefRecord` from flat organism, strain, serotype, and locus fields."""
        strain_rec = ReferenceStrain.from_raw(
            organism,
            strain,
            raw_serotype=serotype,
            synonyms=strain_synonyms,
            assembly_id=assembly_id,
            ncbi_taxid=ncbi_taxid,
            straininfo_id=straininfo_id,
            bacdive_id=bacdive_id,
            pmids=pmids,
        )
        arch = locus.archetype if locus is not None else None
        parsed = strain_rec.species.parse_serotype(serotype, archetype=arch)
        sero_b = parsed[0][1] if parsed else b""
        if not sero_b and arch is not None and arch in INVARIANT_ARCHETYPE_GLYCAN_NAMES:
            sero_b = INVARIANT_ARCHETYPE_GLYCAN_NAMES[arch]
        loc_named = locus.with_name(sero_b) if locus is not None else None
        return cls(
            strain_record=strain_rec,
            serotype=sero_b,
            locus=loc_named,
        )

    @property
    def species(self) -> BacterialSpecies:
        """Canonical `BacterialSpecies` enum member of this record."""
        return self.strain_record.species

    @property
    def organism(self) -> bytes:
        """Canonical bacterial species name (e.g. `b"Escherichia coli"`)."""
        return self.strain_record.organism

    @property
    def strain(self) -> bytes:
        """Primary reference strain designation (e.g. `b"NCTC 9001"`)."""
        return self.strain_record.name

    @property
    def strain_synonyms(self) -> tuple[bytes, ...]:
        """All equivalent culture-collection and lab designations from DSMZ StrainInfo v2."""
        return self.strain_record.synonyms

    @property
    def assembly_id(self) -> bytes:
        """INSDC genome assembly or nucleotide accession (e.g. `b"GCA_900706755.1"`)."""
        return self.strain_record.assembly_id

    @property
    def ncbi_taxid(self) -> int:
        """Resolved NCBI Taxonomy ID integer."""
        return self.strain_record.effective_taxid

    @property
    def straininfo_id(self) -> int:
        """DSMZ StrainInfo `SI-ID` integer, or `0` if unknown."""
        return self.strain_record.straininfo_id

    @property
    def bacdive_id(self) -> int:
        """DSMZ BacDive ID integer, or `0` if unknown."""
        return self.strain_record.bacdive_id

    @property
    def pmids(self) -> tuple[int, ...]:
        """Tuple of PubMed IDs associated with this reference strain or locus."""
        return self.strain_record.pmids

    @property
    def archetype(self) -> LocusArchetype:
        """Return the `LocusArchetype` of the reference `locus`, or `LocusArchetype.UNKNOWN`."""
        return self.locus.archetype if self.locus is not None else LocusArchetype.UNKNOWN

    @property
    def category(self) -> ArchetypeCategory:
        """Return the high-level `ArchetypeCategory` of the reference `locus`."""
        return self.archetype.category

    @property
    def antigen_type(self) -> bytes:
        """Surface polysaccharide antigen class (`b"CPS"`, `b"O-antigen"`, `b"EPS"`, `b"WTA"`) derived from `archetype`."""
        return archetype_to_antigen_type(self.archetype)

    @property
    def has_locus(self) -> bool:
        """True if this reference record holds a non-empty `Locus`."""
        return self.locus is not None and len(self.locus.nodes) > 0

    @property
    def compound_key(self) -> bytes:
        """Unique composite reference key `b"<organism>|<antigen_type>|<serotype>|<assembly_id>"`."""
        return b"%s|%s|%s|%s" % (self.organism, self.antigen_type, self.serotype, self.assembly_id)

    def merge(self, other: SeroRefRecord) -> Self:
        """Merge metadata from another `SeroRefRecord` describing the same physical locus."""
        merged_strain = self.strain_record.merge(other.strain_record)
        sero = self.serotype or other.serotype
        if merged_strain.name and sero and merged_strain.name.lower() == sero.lower():
            merged_strain = ReferenceStrain(
                species=merged_strain.species,
                name=b"",
                synonyms=merged_strain.synonyms,
                assembly_id=merged_strain.assembly_id,
                ncbi_taxid=merged_strain.ncbi_taxid,
                straininfo_id=merged_strain.straininfo_id,
                bacdive_id=merged_strain.bacdive_id,
                pmids=merged_strain.pmids,
            )
        base_loc = self.locus if self.locus is not None else other.locus
        loc = base_loc.with_name(sero) if base_loc is not None else None
        return type(self)(
            strain_record=merged_strain,
            serotype=sero,
            locus=loc,
        )

species `property` ¶

species: BacterialSpecies

Canonical BacterialSpecies enum member of this record.

organism `property` ¶

organism: bytes

Canonical bacterial species name (e.g. b"Escherichia coli").

strain `property` ¶

strain: bytes

Primary reference strain designation (e.g. b"NCTC 9001").

strain_synonyms `property` ¶

strain_synonyms: tuple[bytes, ...]

All equivalent culture-collection and lab designations from DSMZ StrainInfo v2.

assembly_id `property` ¶

assembly_id: bytes

INSDC genome assembly or nucleotide accession (e.g. b"GCA_900706755.1").

ncbi_taxid `property` ¶

ncbi_taxid: int

Resolved NCBI Taxonomy ID integer.

straininfo_id `property` ¶

straininfo_id: int

DSMZ StrainInfo SI-ID integer, or 0 if unknown.

bacdive_id `property` ¶

bacdive_id: int

DSMZ BacDive ID integer, or 0 if unknown.

pmids `property` ¶

pmids: tuple[int, ...]

Tuple of PubMed IDs associated with this reference strain or locus.

archetype `property` ¶

archetype: LocusArchetype

Return the LocusArchetype of the reference locus, or LocusArchetype.UNKNOWN.

category `property` ¶

category: ArchetypeCategory

Return the high-level ArchetypeCategory of the reference locus.

antigen_type `property` ¶

antigen_type: bytes

Surface polysaccharide antigen class (b"CPS", b"O-antigen", b"EPS", b"WTA") derived from archetype.

has_locus `property` ¶

has_locus: bool

True if this reference record holds a non-empty Locus.

compound_key `property` ¶

compound_key: bytes

Unique composite reference key b"<organism>|<antigen_type>|<serotype>|<assembly_id>".

from_fields `classmethod` ¶

from_fields(
    organism: str | bytes | int | BacterialSpecies,
    *,
    serotype: str | bytes = b"",
    strain: str | bytes = b"",
    strain_synonyms: Sequence[bytes | str] = (),
    assembly_id: bytes | str = b"",
    locus: Locus | None = None,
    ncbi_taxid: int = 0,
    straininfo_id: int = 0,
    bacdive_id: int = 0,
    pmids: Sequence[int] = (),
) -> Self

Construct a SeroRefRecord from flat organism, strain, serotype, and locus fields.

Source code in src/seroref/db/models.py

@classmethod
def from_fields(
    cls,
    organism: str | bytes | int | BacterialSpecies,
    *,
    serotype: str | bytes = b"",
    strain: str | bytes = b"",
    strain_synonyms: Sequence[bytes | str] = (),
    assembly_id: bytes | str = b"",
    locus: Locus | None = None,
    ncbi_taxid: int = 0,
    straininfo_id: int = 0,
    bacdive_id: int = 0,
    pmids: Sequence[int] = (),
) -> Self:
    """Construct a `SeroRefRecord` from flat organism, strain, serotype, and locus fields."""
    strain_rec = ReferenceStrain.from_raw(
        organism,
        strain,
        raw_serotype=serotype,
        synonyms=strain_synonyms,
        assembly_id=assembly_id,
        ncbi_taxid=ncbi_taxid,
        straininfo_id=straininfo_id,
        bacdive_id=bacdive_id,
        pmids=pmids,
    )
    arch = locus.archetype if locus is not None else None
    parsed = strain_rec.species.parse_serotype(serotype, archetype=arch)
    sero_b = parsed[0][1] if parsed else b""
    if not sero_b and arch is not None and arch in INVARIANT_ARCHETYPE_GLYCAN_NAMES:
        sero_b = INVARIANT_ARCHETYPE_GLYCAN_NAMES[arch]
    loc_named = locus.with_name(sero_b) if locus is not None else None
    return cls(
        strain_record=strain_rec,
        serotype=sero_b,
        locus=loc_named,
    )

merge ¶

merge(other: SeroRefRecord) -> Self

Merge metadata from another SeroRefRecord describing the same physical locus.

Source code in src/seroref/db/models.py

def merge(self, other: SeroRefRecord) -> Self:
    """Merge metadata from another `SeroRefRecord` describing the same physical locus."""
    merged_strain = self.strain_record.merge(other.strain_record)
    sero = self.serotype or other.serotype
    if merged_strain.name and sero and merged_strain.name.lower() == sero.lower():
        merged_strain = ReferenceStrain(
            species=merged_strain.species,
            name=b"",
            synonyms=merged_strain.synonyms,
            assembly_id=merged_strain.assembly_id,
            ncbi_taxid=merged_strain.ncbi_taxid,
            straininfo_id=merged_strain.straininfo_id,
            bacdive_id=merged_strain.bacdive_id,
            pmids=merged_strain.pmids,
        )
    base_loc = self.locus if self.locus is not None else other.locus
    loc = base_loc.with_name(sero) if base_loc is not None else None
    return type(self)(
        strain_record=merged_strain,
        serotype=sero,
        locus=loc,
    )

archetype_to_antigen_type ¶

archetype_to_antigen_type(
    archetype: LocusArchetype,
) -> bytes

Map a LocusArchetype to its canonical surface polysaccharide antigen class (CPS, O-antigen, EPS, WTA).

Source code in src/seroref/db/models.py

def archetype_to_antigen_type(archetype: LocusArchetype) -> bytes:
    """Map a `LocusArchetype` to its canonical surface polysaccharide antigen class (`CPS`, `O-antigen`, `EPS`, `WTA`)."""
    cat = archetype.category
    if cat == ArchetypeCategory.CAPSULE:
        return b"CPS"
    if cat == ArchetypeCategory.EXOPOLYSACCHARIDE:
        return b"EPS"
    if archetype == LocusArchetype.TEICHOIC_ACID:
        return b"WTA"
    return b"O-antigen"

Batched Reference Database Containers (`seroref.db.records`)¶

records ¶

Structure-of-Arrays (SoA) batch containers for ReferenceStrain and SeroRefRecord.

Classes:

Name	Description
`ReferenceStrains`	Ragged Structure-of-Arrays (SoA) container for deduplicated `ReferenceStrain` records.
`SeroRefRecords`	Normalized relational Structure-of-Arrays (SoA) container for `SeroRefRecord` loci.

Functions:

Name	Description
`loci_to_npz_dict`	Serialize a `Loci` SoA batch (and its embedded `Genes`) into a flat `.npz`-ready dictionary.
`loci_from_npz_dict`	Deserialize a `Loci` SoA batch from a flat `.npz` archive dictionary.
`load_seroref_records`	Load curated `SeroRefRecords` from `seroref.npz`.

ReferenceStrains `dataclass` ¶

Bases: RaggedArrayContainer[ReferenceStrain, 'ReferenceStrains']

Ragged Structure-of-Arrays (SoA) container for deduplicated ReferenceStrain records.

Stores per-strain species IDs, primary strain names, INSDC assembly accessions, taxonomy/culture collection IDs, and CSR-packed ragged arrays for culture-collection synonym_values (offsets, lengths) and literature pmid_values (pmid_offsets, pmid_lengths).

Methods:

Name	Description
`empty`	Create an empty 0-length `ReferenceStrains` container.
`from_list`	Compile a sequence of `ReferenceStrain` scalar views into a `ReferenceStrains` SoA container.
`concat`	Concatenate multiple `ReferenceStrains` containers.
`where_species`	Filter strains by canonical `BacterialSpecies`.
`where_assembly_id`	Filter strains by INSDC `assembly_id`.
`equivalence_csr`	Build the `(M, M)` symmetric `CSR` adjacency matrix of `ReferenceStrain` equivalence cliques.
`merge_cliques`	Resolve transitive `ReferenceStrain` equivalence cliques via `CSR.connected_components_from_edges` (`seroverse.core.graph.csr.CSR`).
`to_npz_dict`	Serialize this `ReferenceStrains` container into a flat `.npz`-ready dictionary.
`from_npz_dict`	Deserialize a `ReferenceStrains` container from a flat `.npz` archive dictionary.

Attributes:

Name	Type	Description
`organisms`	`ndarray[tuple[int], dtype[bytes_]]`	Vectorized 1D array of canonical scientific names resolved from `self.species_ids`.
`synonym_offsets`	`ndarray[tuple[int], dtype[int32]]`	Alias for `self.offsets` (starting indices into `self.synonym_values`).
`synonym_lengths`	`ndarray[tuple[int], dtype[int32]]`	Alias for `self.lengths` (number of synonyms per strain record).

Source code in src/seroref/db/records.py

@dataclass(frozen=True, slots=True)
class ReferenceStrains(RaggedArrayContainer[ReferenceStrain, "ReferenceStrains"]):
    """Ragged Structure-of-Arrays (SoA) container for deduplicated [`ReferenceStrain`][seroref.db.models.ReferenceStrain] records.

    Stores per-strain species IDs, primary strain names, INSDC assembly accessions, taxonomy/culture
    collection IDs, and CSR-packed ragged arrays for culture-collection `synonym_values` (`offsets`,
    `lengths`) and literature `pmid_values` (`pmid_offsets`, `pmid_lengths`).
    """

    species_ids: np.ndarray[tuple[int], np.dtype[np.int16]]
    names: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    assembly_ids: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    ncbi_taxids: np.ndarray[tuple[int], np.dtype[np.int32]]
    straininfo_ids: np.ndarray[tuple[int], np.dtype[np.int32]]
    bacdive_ids: np.ndarray[tuple[int], np.dtype[np.int32]]
    synonym_values: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    pmid_values: np.ndarray[tuple[int], np.dtype[np.int64]]
    pmid_offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    pmid_lengths: np.ndarray[tuple[int], np.dtype[np.int32]]
    offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    lengths: np.ndarray[tuple[int], np.dtype[np.int32]]

    def __len__(self) -> int:
        return len(self.species_ids)

    def __iter__(self) -> Iterator[ReferenceStrain]:
        for i in range(len(self)):
            yield self[i]

    @property
    def organisms(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """Vectorized 1D array of canonical scientific names resolved from `self.species_ids`."""
        if len(self) == 0:
            return np.empty(0, dtype="S1")
        specs = BacterialSpecies.specs()
        clipped = np.clip(self.species_ids, 0, len(specs) - 1)
        return specs.names[clipped]

    @property
    def synonym_offsets(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """Alias for `self.offsets` (starting indices into `self.synonym_values`)."""
        return self.offsets

    @property
    def synonym_lengths(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """Alias for `self.lengths` (number of synonyms per strain record)."""
        return self.lengths

    @overload
    def __getitem__(self, item: int | np.integer[Any]) -> ReferenceStrain: ...

    @overload
    def __getitem__(self, item: slice | np.ndarray[tuple[int], Any] | list[Any]) -> Self: ...

    @overload
    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> ReferenceStrain | Self: ...

    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> ReferenceStrain | Self:
        if isinstance(item, (int, np.integer)):
            idx = int(item)
            n = len(self)
            if idx < 0:
                idx += n
            if idx < 0 or idx >= n:
                raise IndexError(f"ReferenceStrains index {item} out of range for batch of length {n}.")
            s_off, s_ln = int(self.offsets[idx]), int(self.lengths[idx])
            syns = tuple(bytes(self.synonym_values[k]) for k in range(s_off, s_off + s_ln))
            p_off, p_ln = int(self.pmid_offsets[idx]), int(self.pmid_lengths[idx])
            pms = tuple(int(self.pmid_values[k]) for k in range(p_off, p_off + p_ln))
            sp_val = int(self.species_ids[idx])
            sp = BacterialSpecies(sp_val) if 0 <= sp_val < len(BacterialSpecies) else BacterialSpecies.UNKNOWN
            return ReferenceStrain(
                species=sp,
                name=bytes(self.names[idx]),
                synonyms=syns,
                assembly_id=bytes(self.assembly_ids[idx]),
                ncbi_taxid=int(self.ncbi_taxids[idx]),
                straininfo_id=int(self.straininfo_ids[idx]),
                bacdive_id=int(self.bacdive_ids[idx]),
                pmids=pms,
            )

        syn_idx_pool = np.arange(len(self.synonym_values), dtype=np.int32)
        indices, new_syn_offsets, new_syn_lengths, (gathered_syn_idx,) = self._slice_ragged(item, syn_idx_pool)
        if len(indices) == 0:
            return self.empty()
        syn_v = self.synonym_values[gathered_syn_idx] if len(gathered_syn_idx) > 0 else np.empty(0, dtype="S1")
        pm_v, pm_o, pm_l = _gather_ragged_secondary(
            self.pmid_values,
            self.pmid_offsets,
            self.pmid_lengths,
            indices,
        )
        return type(self)(
            species_ids=self.species_ids[indices],
            names=self.names[indices],
            assembly_ids=self.assembly_ids[indices],
            ncbi_taxids=self.ncbi_taxids[indices],
            straininfo_ids=self.straininfo_ids[indices],
            bacdive_ids=self.bacdive_ids[indices],
            synonym_values=syn_v,
            offsets=new_syn_offsets,
            lengths=new_syn_lengths,
            pmid_values=pm_v,
            pmid_offsets=pm_o,
            pmid_lengths=pm_l,
        )

    @classmethod
    def empty(cls) -> Self:
        """Create an empty 0-length `ReferenceStrains` container."""
        return cls(
            species_ids=np.empty(0, dtype=np.int16),
            names=np.empty(0, dtype="S1"),
            assembly_ids=np.empty(0, dtype="S1"),
            ncbi_taxids=np.empty(0, dtype=np.int32),
            straininfo_ids=np.empty(0, dtype=np.int32),
            bacdive_ids=np.empty(0, dtype=np.int32),
            synonym_values=np.empty(0, dtype="S1"),
            offsets=np.empty(0, dtype=np.int32),
            lengths=np.empty(0, dtype=np.int32),
            pmid_values=np.empty(0, dtype=np.int64),
            pmid_offsets=np.empty(0, dtype=np.int32),
            pmid_lengths=np.empty(0, dtype=np.int32),
        )

    @classmethod
    def from_list(cls, strains: Sequence[ReferenceStrain]) -> Self:
        """Compile a sequence of `ReferenceStrain` scalar views into a `ReferenceStrains` SoA container."""
        m = len(strains)
        if m == 0:
            return cls.empty()
        syn_flat: list[bytes] = []
        syn_lens = np.empty(m, dtype=np.int32)
        pm_flat: list[int] = []
        pm_lens = np.empty(m, dtype=np.int32)

        for i, s in enumerate(strains):
            syn_lens[i] = len(s.synonyms)
            syn_flat.extend(s.synonyms)
            pm_lens[i] = len(s.pmids)
            pm_flat.extend(s.pmids)

        return cls(
            species_ids=np.array([int(s.species) for s in strains], dtype=np.int16),
            names=np.array([s.name for s in strains], dtype="S"),
            assembly_ids=np.array([s.assembly_id for s in strains], dtype="S"),
            ncbi_taxids=np.array([s.effective_taxid for s in strains], dtype=np.int32),
            straininfo_ids=np.array([s.straininfo_id for s in strains], dtype=np.int32),
            bacdive_ids=np.array([s.bacdive_id for s in strains], dtype=np.int32),
            synonym_values=np.array(syn_flat, dtype="S") if syn_flat else np.empty(0, dtype="S1"),
            offsets=lengths_to_offsets(syn_lens),
            lengths=syn_lens,
            pmid_values=np.array(pm_flat, dtype=np.int64) if pm_flat else np.empty(0, dtype=np.int64),
            pmid_offsets=lengths_to_offsets(pm_lens),
            pmid_lengths=pm_lens,
        )

    @classmethod
    def concat(cls, batches: Iterable[Self]) -> Self:
        """Concatenate multiple `ReferenceStrains` containers."""
        items: list[ReferenceStrain] = []
        for b in batches:
            items.extend(b)
        return cls.from_list(items)

    def where_species(self, species: str | bytes | int | BacterialSpecies) -> Self:
        """Filter strains by canonical `BacterialSpecies`."""
        sp = BacterialSpecies.resolve(species)
        return self[self.species_ids == int(sp)]

    def where_assembly_id(self, assembly_id: bytes | str) -> Self:
        """Filter strains by INSDC `assembly_id`."""
        asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else assembly_id
        return self[self.assembly_ids == asm_b]

    def _equivalence_edges(
        self,
    ) -> tuple[np.ndarray[tuple[int], np.dtype[np.int32]], np.ndarray[tuple[int], np.dtype[np.int32]]]:
        """Extract undirected equivalence edges `(edge_u, edge_v)` across shared assembly, StrainInfo, BacDive, or synonym keys."""
        m = len(self)
        if m <= 1:
            empty = np.empty(0, dtype=np.int32)
            return empty, empty

        buckets: dict[tuple[Any, ...], list[int]] = defaultdict(list)
        for i in range(m):
            sp = int(self.species_ids[i])
            asm = bytes(self.assembly_ids[i])
            if asm:
                buckets[("asm", sp, asm.split(b".")[0].upper())].append(i)
            si = int(self.straininfo_ids[i])
            if si > 0:
                buckets[("si", sp, si)].append(i)
            bd = int(self.bacdive_ids[i])
            if bd > 0:
                buckets[("bd", sp, bd)].append(i)
            name_b = bytes(self.names[i])
            if name_b and not asm:
                norm_n = re.sub(rb"[\s\-_/]+", b"", name_b.upper())
                if norm_n:
                    buckets[("syn", sp, norm_n)].append(i)
            s_off, s_ln = int(self.offsets[i]), int(self.lengths[i])
            for k in range(s_off, s_off + s_ln):
                syn_b = bytes(self.synonym_values[k])
                if syn_b and not asm:
                    norm_s = re.sub(rb"[\s\-_/]+", b"", syn_b.upper())
                    if norm_s:
                        buckets[("syn", sp, norm_s)].append(i)

        edge_u: list[int] = []
        edge_v: list[int] = []
        for members in buckets.values():
            if len(members) > 1:
                root = members[0]
                for other in members[1:]:
                    if other != root:
                        edge_u.append(root)
                        edge_v.append(other)
        if not edge_u:
            empty = np.empty(0, dtype=np.int32)
            return empty, empty
        return np.array(edge_u, dtype=np.int32), np.array(edge_v, dtype=np.int32)

    def equivalence_csr(self) -> CSR:
        """Build the `(M, M)` symmetric `CSR` adjacency matrix of `ReferenceStrain` equivalence cliques."""
        m = len(self)
        if m == 0:
            return CSR.empty((0, 0), dtype=np.int8)
        u, v = self._equivalence_edges()
        if len(u) == 0:
            return CSR.empty((m, m), dtype=np.int8)
        rows = np.concatenate([u, v])
        cols = np.concatenate([v, u])
        return CSR.from_coo(rows, cols, np.ones(len(rows), dtype=np.int8), shape=(m, m))

    def merge_cliques(self) -> tuple[Self, np.ndarray[tuple[int], np.dtype[np.int32]]]:
        """Resolve transitive `ReferenceStrain` equivalence cliques via `CSR.connected_components_from_edges` (`seroverse.core.graph.csr.CSR`).

        Returns:
            tuple[ReferenceStrains, np.ndarray]: `(merged_strains, component_labels)` where `component_labels`
            is a 1D `int32` array of length `M` mapping each input strain index to its merged clique index.
        """
        m = len(self)
        if m == 0:
            return self, np.empty(0, dtype=np.int32)
        u, v = self._equivalence_edges()
        n_comp, labels = CSR.connected_components_from_edges(m, u, v)
        if n_comp == m:
            return self, np.arange(m, dtype=np.int32)
        merged_list: list[ReferenceStrain | None] = [None] * n_comp
        for i in range(m):
            c = int(labels[i])
            sr = self[i]
            cur = merged_list[c]
            merged_list[c] = sr if cur is None else cur.merge(sr)
        return self.from_list([s for s in merged_list if s is not None]), labels

    def to_npz_dict(self, prefix: str = "strain_") -> dict[str, Any]:
        """Serialize this `ReferenceStrains` container into a flat `.npz`-ready dictionary."""
        return {
            f"{prefix}species_ids": self.species_ids,
            f"{prefix}names": self.names,
            f"{prefix}assembly_ids": self.assembly_ids,
            f"{prefix}ncbi_taxids": self.ncbi_taxids,
            f"{prefix}straininfo_ids": self.straininfo_ids,
            f"{prefix}bacdive_ids": self.bacdive_ids,
            f"{prefix}synonym_values": self.synonym_values,
            f"{prefix}synonym_offsets": self.offsets,
            f"{prefix}synonym_lengths": self.lengths,
            f"{prefix}pmid_values": self.pmid_values,
            f"{prefix}pmid_offsets": self.pmid_offsets,
            f"{prefix}pmid_lengths": self.pmid_lengths,
        }

    @classmethod
    def from_npz_dict(cls, data: Any, prefix: str = "strain_") -> Self:
        """Deserialize a `ReferenceStrains` container from a flat `.npz` archive dictionary."""
        return cls(
            species_ids=data[f"{prefix}species_ids"].astype(np.int16, copy=False),
            names=data[f"{prefix}names"],
            assembly_ids=data[f"{prefix}assembly_ids"],
            ncbi_taxids=data[f"{prefix}ncbi_taxids"].astype(np.int32, copy=False),
            straininfo_ids=data[f"{prefix}straininfo_ids"].astype(np.int32, copy=False),
            bacdive_ids=data[f"{prefix}bacdive_ids"].astype(np.int32, copy=False),
            synonym_values=data[f"{prefix}synonym_values"],
            offsets=data[f"{prefix}synonym_offsets"].astype(np.int32, copy=False),
            lengths=data[f"{prefix}synonym_lengths"].astype(np.int32, copy=False),
            pmid_values=data[f"{prefix}pmid_values"].astype(np.int64, copy=False),
            pmid_offsets=data[f"{prefix}pmid_offsets"].astype(np.int32, copy=False),
            pmid_lengths=data[f"{prefix}pmid_lengths"].astype(np.int32, copy=False),
        )

organisms `property` ¶

organisms: ndarray[tuple[int], dtype[bytes_]]

Vectorized 1D array of canonical scientific names resolved from self.species_ids.

synonym_offsets `property` ¶

synonym_offsets: ndarray[tuple[int], dtype[int32]]

Alias for self.offsets (starting indices into self.synonym_values).

synonym_lengths `property` ¶

synonym_lengths: ndarray[tuple[int], dtype[int32]]

Alias for self.lengths (number of synonyms per strain record).

empty `classmethod` ¶

empty() -> Self

Create an empty 0-length ReferenceStrains container.

Source code in src/seroref/db/records.py

@classmethod
def empty(cls) -> Self:
    """Create an empty 0-length `ReferenceStrains` container."""
    return cls(
        species_ids=np.empty(0, dtype=np.int16),
        names=np.empty(0, dtype="S1"),
        assembly_ids=np.empty(0, dtype="S1"),
        ncbi_taxids=np.empty(0, dtype=np.int32),
        straininfo_ids=np.empty(0, dtype=np.int32),
        bacdive_ids=np.empty(0, dtype=np.int32),
        synonym_values=np.empty(0, dtype="S1"),
        offsets=np.empty(0, dtype=np.int32),
        lengths=np.empty(0, dtype=np.int32),
        pmid_values=np.empty(0, dtype=np.int64),
        pmid_offsets=np.empty(0, dtype=np.int32),
        pmid_lengths=np.empty(0, dtype=np.int32),
    )

from_list `classmethod` ¶

from_list(strains: Sequence[ReferenceStrain]) -> Self

Compile a sequence of ReferenceStrain scalar views into a ReferenceStrains SoA container.

Source code in src/seroref/db/records.py

@classmethod
def from_list(cls, strains: Sequence[ReferenceStrain]) -> Self:
    """Compile a sequence of `ReferenceStrain` scalar views into a `ReferenceStrains` SoA container."""
    m = len(strains)
    if m == 0:
        return cls.empty()
    syn_flat: list[bytes] = []
    syn_lens = np.empty(m, dtype=np.int32)
    pm_flat: list[int] = []
    pm_lens = np.empty(m, dtype=np.int32)

    for i, s in enumerate(strains):
        syn_lens[i] = len(s.synonyms)
        syn_flat.extend(s.synonyms)
        pm_lens[i] = len(s.pmids)
        pm_flat.extend(s.pmids)

    return cls(
        species_ids=np.array([int(s.species) for s in strains], dtype=np.int16),
        names=np.array([s.name for s in strains], dtype="S"),
        assembly_ids=np.array([s.assembly_id for s in strains], dtype="S"),
        ncbi_taxids=np.array([s.effective_taxid for s in strains], dtype=np.int32),
        straininfo_ids=np.array([s.straininfo_id for s in strains], dtype=np.int32),
        bacdive_ids=np.array([s.bacdive_id for s in strains], dtype=np.int32),
        synonym_values=np.array(syn_flat, dtype="S") if syn_flat else np.empty(0, dtype="S1"),
        offsets=lengths_to_offsets(syn_lens),
        lengths=syn_lens,
        pmid_values=np.array(pm_flat, dtype=np.int64) if pm_flat else np.empty(0, dtype=np.int64),
        pmid_offsets=lengths_to_offsets(pm_lens),
        pmid_lengths=pm_lens,
    )

concat `classmethod` ¶

concat(batches: Iterable[Self]) -> Self

Concatenate multiple ReferenceStrains containers.

Source code in src/seroref/db/records.py

@classmethod
def concat(cls, batches: Iterable[Self]) -> Self:
    """Concatenate multiple `ReferenceStrains` containers."""
    items: list[ReferenceStrain] = []
    for b in batches:
        items.extend(b)
    return cls.from_list(items)

where_species ¶

where_species(
    species: str | bytes | int | BacterialSpecies,
) -> Self

Filter strains by canonical BacterialSpecies.

Source code in src/seroref/db/records.py

def where_species(self, species: str | bytes | int | BacterialSpecies) -> Self:
    """Filter strains by canonical `BacterialSpecies`."""
    sp = BacterialSpecies.resolve(species)
    return self[self.species_ids == int(sp)]

where_assembly_id ¶

where_assembly_id(assembly_id: bytes | str) -> Self

Filter strains by INSDC assembly_id.

Source code in src/seroref/db/records.py

def where_assembly_id(self, assembly_id: bytes | str) -> Self:
    """Filter strains by INSDC `assembly_id`."""
    asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else assembly_id
    return self[self.assembly_ids == asm_b]

equivalence_csr ¶

equivalence_csr() -> CSR

Build the (M, M) symmetric CSR adjacency matrix of ReferenceStrain equivalence cliques.

Source code in src/seroref/db/records.py

def equivalence_csr(self) -> CSR:
    """Build the `(M, M)` symmetric `CSR` adjacency matrix of `ReferenceStrain` equivalence cliques."""
    m = len(self)
    if m == 0:
        return CSR.empty((0, 0), dtype=np.int8)
    u, v = self._equivalence_edges()
    if len(u) == 0:
        return CSR.empty((m, m), dtype=np.int8)
    rows = np.concatenate([u, v])
    cols = np.concatenate([v, u])
    return CSR.from_coo(rows, cols, np.ones(len(rows), dtype=np.int8), shape=(m, m))

merge_cliques ¶

merge_cliques() -> tuple[
    Self, ndarray[tuple[int], dtype[int32]]
]

Resolve transitive ReferenceStrain equivalence cliques via CSR.connected_components_from_edges (seroverse.core.graph.csr.CSR).

Returns:

Type	Description
`Self`	tuple[ReferenceStrains, np.ndarray]: `(merged_strains, component_labels)` where `component_labels`
`ndarray[tuple[int], dtype[int32]]`	is a 1D `int32` array of length `M` mapping each input strain index to its merged clique index.

Source code in src/seroref/db/records.py

def merge_cliques(self) -> tuple[Self, np.ndarray[tuple[int], np.dtype[np.int32]]]:
    """Resolve transitive `ReferenceStrain` equivalence cliques via `CSR.connected_components_from_edges` (`seroverse.core.graph.csr.CSR`).

    Returns:
        tuple[ReferenceStrains, np.ndarray]: `(merged_strains, component_labels)` where `component_labels`
        is a 1D `int32` array of length `M` mapping each input strain index to its merged clique index.
    """
    m = len(self)
    if m == 0:
        return self, np.empty(0, dtype=np.int32)
    u, v = self._equivalence_edges()
    n_comp, labels = CSR.connected_components_from_edges(m, u, v)
    if n_comp == m:
        return self, np.arange(m, dtype=np.int32)
    merged_list: list[ReferenceStrain | None] = [None] * n_comp
    for i in range(m):
        c = int(labels[i])
        sr = self[i]
        cur = merged_list[c]
        merged_list[c] = sr if cur is None else cur.merge(sr)
    return self.from_list([s for s in merged_list if s is not None]), labels

to_npz_dict ¶

to_npz_dict(prefix: str = 'strain_') -> dict[str, Any]

Serialize this ReferenceStrains container into a flat .npz-ready dictionary.

Source code in src/seroref/db/records.py

def to_npz_dict(self, prefix: str = "strain_") -> dict[str, Any]:
    """Serialize this `ReferenceStrains` container into a flat `.npz`-ready dictionary."""
    return {
        f"{prefix}species_ids": self.species_ids,
        f"{prefix}names": self.names,
        f"{prefix}assembly_ids": self.assembly_ids,
        f"{prefix}ncbi_taxids": self.ncbi_taxids,
        f"{prefix}straininfo_ids": self.straininfo_ids,
        f"{prefix}bacdive_ids": self.bacdive_ids,
        f"{prefix}synonym_values": self.synonym_values,
        f"{prefix}synonym_offsets": self.offsets,
        f"{prefix}synonym_lengths": self.lengths,
        f"{prefix}pmid_values": self.pmid_values,
        f"{prefix}pmid_offsets": self.pmid_offsets,
        f"{prefix}pmid_lengths": self.pmid_lengths,
    }

from_npz_dict `classmethod` ¶

from_npz_dict(data: Any, prefix: str = 'strain_') -> Self

Deserialize a ReferenceStrains container from a flat .npz archive dictionary.

Source code in src/seroref/db/records.py

@classmethod
def from_npz_dict(cls, data: Any, prefix: str = "strain_") -> Self:
    """Deserialize a `ReferenceStrains` container from a flat `.npz` archive dictionary."""
    return cls(
        species_ids=data[f"{prefix}species_ids"].astype(np.int16, copy=False),
        names=data[f"{prefix}names"],
        assembly_ids=data[f"{prefix}assembly_ids"],
        ncbi_taxids=data[f"{prefix}ncbi_taxids"].astype(np.int32, copy=False),
        straininfo_ids=data[f"{prefix}straininfo_ids"].astype(np.int32, copy=False),
        bacdive_ids=data[f"{prefix}bacdive_ids"].astype(np.int32, copy=False),
        synonym_values=data[f"{prefix}synonym_values"],
        offsets=data[f"{prefix}synonym_offsets"].astype(np.int32, copy=False),
        lengths=data[f"{prefix}synonym_lengths"].astype(np.int32, copy=False),
        pmid_values=data[f"{prefix}pmid_values"].astype(np.int64, copy=False),
        pmid_offsets=data[f"{prefix}pmid_offsets"].astype(np.int32, copy=False),
        pmid_lengths=data[f"{prefix}pmid_lengths"].astype(np.int32, copy=False),
    )

SeroRefRecords `dataclass` ¶

Bases: BatchedContainer[SeroRefRecord, 'SeroRefRecords']

Normalized relational Structure-of-Arrays (SoA) container for SeroRefRecord loci.

Deduplicates reference isolates and genome assemblies into self.strains (ReferenceStrains, length M) and links each of the N predicted surface polysaccharide loci (self.loci, self.serotypes) to its parent strain via self.strain_indices (int32[N], analogous to Genes.contig_indices).

Methods:

Name	Description
`empty`	Create an empty 0-length `SeroRefRecords` container.
`from_list`	Compile a sequence of `SeroRefRecord` scalar views into a normalized `SeroRefRecords` container.
`concat`	Concatenate and normalize multiple `SeroRefRecords` batches.
`where_species`	Filter records by canonical `BacterialSpecies`.
`where_organism`	Filter records by organism name or `BacterialSpecies`.
`where_serotype`	Filter records by canonical `serotype` byte string.
`where_antigen_type`	Filter records by derived surface polysaccharide `antigen_type` (`CPS`, `O-antigen`, `EPS`, `WTA`).
`where_assembly_id`	Filter records by INSDC genome or locus `assembly_id`.
`where_archetype`	Filter records by `LocusArchetype`.
`save`	Atomically serialize this `SeroRefRecords` container to a compressed `.npz` archive.
`load`	Load a `SeroRefRecords` container from `.npz` (supporting both relational and legacy flat schemas).

Attributes:

Name	Type	Description
`species_ids`	`ndarray[tuple[int], dtype[int16]]`	1D `int16` array (`(N,)`) of `BacterialSpecies` IDs for each locus record.
`organisms`	`ndarray[tuple[int], dtype[bytes_]]`	1D byte-string array (`(N,)`) of canonical species names for each locus record.
`strain_names`	`ndarray[tuple[int], dtype[bytes_]]`	1D byte-string array (`(N,)`) of primary reference strain designations for each locus record.
`assembly_ids`	`ndarray[tuple[int], dtype[bytes_]]`	1D byte-string array (`(N,)`) of INSDC assembly/nucleotide accessions for each locus record.
`ncbi_taxids`	`ndarray[tuple[int], dtype[int32]]`	1D `int32` array (`(N,)`) of NCBI Taxonomy IDs for each locus record.
`straininfo_ids`	`ndarray[tuple[int], dtype[int32]]`	1D `int32` array (`(N,)`) of DSMZ StrainInfo IDs for each locus record.
`bacdive_ids`	`ndarray[tuple[int], dtype[int32]]`	1D `int32` array (`(N,)`) of DSMZ BacDive IDs for each locus record.
`archetypes`	`ndarray[tuple[int], dtype[int8]]`	1D `int8` array (`(N,)`) of `LocusArchetype` integer IDs across all records in the batch.
`antigen_types`	`ndarray[tuple[int], dtype[bytes_]]`	1D byte-string array (`(N,)`) of surface antigen classes (`b"CPS"`, `b"O-antigen"`, `b"EPS"`, `b"WTA"`).
`strain_locus_csr`	`CSR`	Sparse `(M, N)` `CSR` (`seroverse.core.graph.csr.CSR`) incidence matrix mapping `ReferenceStrains` to `Loci`.
`species_locus_csr`	`CSR`	Sparse `(S, N)` `CSR` (`seroverse.core.graph.csr.CSR`) incidence matrix mapping `BacterialSpecies` to `Loci`.
`species_strain_csr`	`CSR`	Sparse `(S, M)` `CSR` (`seroverse.core.graph.csr.CSR`) incidence matrix mapping `BacterialSpecies` to `ReferenceStrains`.

Source code in src/seroref/db/records.py

@dataclass(frozen=True, slots=True)
class SeroRefRecords(BatchedContainer[SeroRefRecord, "SeroRefRecords"]):
    """Normalized relational Structure-of-Arrays (SoA) container for [`SeroRefRecord`][seroref.db.models.SeroRefRecord] loci.

    Deduplicates reference isolates and genome assemblies into `self.strains` (`ReferenceStrains`, length `M`)
    and links each of the `N` predicted surface polysaccharide loci (`self.loci`, `self.serotypes`) to its parent
    strain via `self.strain_indices` (`int32[N]`, analogous to `Genes.contig_indices`).
    """

    strains: ReferenceStrains
    strain_indices: np.ndarray[tuple[int], np.dtype[np.int32]]
    serotypes: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    loci: Loci

    def __len__(self) -> int:
        return len(self.serotypes)

    def __iter__(self) -> Iterator[SeroRefRecord]:
        for i in range(len(self)):
            yield self[i]

    @property
    def species_ids(self) -> np.ndarray[tuple[int], np.dtype[np.int16]]:
        """1D `int16` array (`(N,)`) of `BacterialSpecies` IDs for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype=np.int16)
        return self.strains.species_ids[self.strain_indices]

    @property
    def organisms(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """1D byte-string array (`(N,)`) of canonical species names for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype="S1")
        return self.strains.organisms[self.strain_indices]

    @property
    def strain_names(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """1D byte-string array (`(N,)`) of primary reference strain designations for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype="S1")
        return self.strains.names[self.strain_indices]

    @property
    def assembly_ids(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """1D byte-string array (`(N,)`) of INSDC assembly/nucleotide accessions for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype="S1")
        return self.strains.assembly_ids[self.strain_indices]

    @property
    def ncbi_taxids(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """1D `int32` array (`(N,)`) of NCBI Taxonomy IDs for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype=np.int32)
        return self.strains.ncbi_taxids[self.strain_indices]

    @property
    def straininfo_ids(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """1D `int32` array (`(N,)`) of DSMZ StrainInfo IDs for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype=np.int32)
        return self.strains.straininfo_ids[self.strain_indices]

    @property
    def bacdive_ids(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """1D `int32` array (`(N,)`) of DSMZ BacDive IDs for each locus record."""
        if len(self) == 0:
            return np.empty(0, dtype=np.int32)
        return self.strains.bacdive_ids[self.strain_indices]

    @property
    def archetypes(self) -> np.ndarray[tuple[int], np.dtype[np.int8]]:
        """1D `int8` array (`(N,)`) of `LocusArchetype` integer IDs across all records in the batch."""
        return self.loci.archetypes

    @property
    def antigen_types(self) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """1D byte-string array (`(N,)`) of surface antigen classes (`b"CPS"`, `b"O-antigen"`, `b"EPS"`, `b"WTA"`)."""
        if len(self) == 0:
            return np.empty(0, dtype="S1")
        arch_ids = np.clip(self.loci.archetypes, 0, len(_ARCHETYPE_TO_ANTIGEN_BYTES) - 1)
        return _ARCHETYPE_TO_ANTIGEN_BYTES[arch_ids]

    @property
    def strain_locus_csr(self) -> CSR:
        """Sparse `(M, N)` `CSR` (`seroverse.core.graph.csr.CSR`) incidence matrix mapping `ReferenceStrains` to `Loci`."""
        m = len(self.strains)
        n = len(self)
        if m == 0 or n == 0:
            return CSR.empty((m, n), dtype=np.int8)
        locus_idx = np.arange(n, dtype=np.int32)
        return CSR.from_coo(self.strain_indices, locus_idx, np.ones(n, dtype=np.int8), shape=(m, n))

    @property
    def species_locus_csr(self) -> CSR:
        """Sparse `(S, N)` `CSR` (`seroverse.core.graph.csr.CSR`) incidence matrix mapping `BacterialSpecies` to `Loci`."""
        s_count = len(BacterialSpecies)
        n = len(self)
        if n == 0:
            return CSR.empty((s_count, 0), dtype=np.int8)
        locus_idx = np.arange(n, dtype=np.int32)
        return CSR.from_coo(
            self.species_ids.astype(np.int32, copy=False),
            locus_idx,
            np.ones(n, dtype=np.int8),
            shape=(s_count, n),
        )

    @property
    def species_strain_csr(self) -> CSR:
        """Sparse `(S, M)` `CSR` (`seroverse.core.graph.csr.CSR`) incidence matrix mapping `BacterialSpecies` to `ReferenceStrains`."""
        s_count = len(BacterialSpecies)
        m = len(self.strains)
        if m == 0:
            return CSR.empty((s_count, 0), dtype=np.int8)
        strain_idx = np.arange(m, dtype=np.int32)
        return CSR.from_coo(
            self.strains.species_ids.astype(np.int32, copy=False),
            strain_idx,
            np.ones(m, dtype=np.int8),
            shape=(s_count, m),
        )

    @overload
    def __getitem__(self, item: int | np.integer[Any]) -> SeroRefRecord: ...

    @overload
    def __getitem__(self, item: slice | np.ndarray[tuple[int], Any] | list[Any]) -> Self: ...

    @overload
    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> SeroRefRecord | Self: ...

    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> SeroRefRecord | Self:
        if isinstance(item, (int, np.integer)):
            idx = int(item)
            n = len(self)
            if idx < 0:
                idx += n
            if idx < 0 or idx >= n:
                raise IndexError(f"SeroRefRecords index {item} out of range for batch of length {n}.")
            s_idx = int(self.strain_indices[idx])
            strain_rec = self.strains[s_idx]
            loc_obj = self.loci[idx]
            has_loc = len(loc_obj.nodes) > 0 or loc_obj.archetype != LocusArchetype.UNKNOWN
            return SeroRefRecord(
                strain_record=strain_rec,
                serotype=bytes(self.serotypes[idx]),
                locus=loc_obj if has_loc else None,
            )

        indices = self._normalize_indices(item)
        if len(indices) == 0:
            return self.empty()
        sub_strain_idx = self.strain_indices[indices]
        unique_sidx, inverse_sidx = np.unique(sub_strain_idx, return_inverse=True)
        return type(self)(
            strains=self.strains[unique_sidx.astype(np.int32)],
            strain_indices=inverse_sidx.astype(np.int32),
            serotypes=self.serotypes[indices],
            loci=self.loci[indices],
        )

    @classmethod
    def empty(cls) -> Self:
        """Create an empty 0-length `SeroRefRecords` container."""
        return cls(
            strains=ReferenceStrains.empty(),
            strain_indices=np.empty(0, dtype=np.int32),
            serotypes=np.empty(0, dtype="S1"),
            loci=Loci.empty(),
        )

    @classmethod
    def from_list(cls, records: Sequence[SeroRefRecord], *, deduplicate: bool = True) -> Self:
        """Compile a sequence of `SeroRefRecord` scalar views into a normalized `SeroRefRecords` container.

        Deduplicates `ReferenceStrain` entries into `self.strains`, merges untyped/typed duplicate
        physical loci when `deduplicate=True`, and synchronizes `Loci.names` with `self.serotypes`.
        """
        if len(records) == 0:
            return cls.empty()

        empty_locus = Locus(nodes=(), edges=())
        normalized_records: list[SeroRefRecord] = []
        for r in records:
            loc = r.locus if r.locus is not None else empty_locus
            sero_b = r.serotype
            if sero_b:
                parsed = r.species.parse_serotype(sero_b, archetype=loc.archetype)
                sero_b = parsed[0][1] if parsed else b""
            if not sero_b and loc.archetype in INVARIANT_ARCHETYPE_GLYCAN_NAMES:
                sero_b = INVARIANT_ARCHETYPE_GLYCAN_NAMES[loc.archetype]
            if sero_b:
                if loc.name != sero_b:
                    loc = loc.with_name(sero_b)
            elif loc.name:
                loc = loc.with_name(b"")
            has_loc = len(loc.nodes) > 0 or loc.archetype != LocusArchetype.UNKNOWN
            normalized_records.append(
                SeroRefRecord(
                    strain_record=r.strain_record,
                    serotype=sero_b,
                    locus=loc if has_loc else None,
                )
            )

        if deduplicate:
            by_key: dict[tuple[Any, ...], SeroRefRecord] = {}
            for r in normalized_records:
                loc = r.locus
                arch = int(loc.archetype) if loc is not None else int(LocusArchetype.UNKNOWN)
                start = int(loc.start) if loc is not None else 0
                end = int(loc.end) if loc is not None else 0
                if r.assembly_id and loc is not None and len(loc.nodes) > 0:
                    key: tuple[Any, ...] = ("asm_loc", int(r.species), r.assembly_id, arch, start, end, r.serotype)
                else:
                    key = ("meta", int(r.species), r.assembly_id, r.strain, arch, r.serotype, start, end)
                existing = by_key.get(key)
                if existing is None:
                    by_key[key] = r
                else:
                    by_key[key] = existing.merge(r)

            # Fold untyped (serotype == b"") records into typed records on the exact same physical locus
            typed_phys: dict[tuple[int, bytes, int, int, int], tuple[Any, ...]] = {}
            for k, r in by_key.items():
                if r.serotype and r.assembly_id and r.locus is not None:
                    phys = (int(r.species), r.assembly_id, int(r.locus.archetype), int(r.locus.start), int(r.locus.end))
                    typed_phys[phys] = k

            deduped_list: list[SeroRefRecord] = []
            for k, r in list(by_key.items()):
                if not r.serotype and r.assembly_id and r.locus is not None:
                    phys = (int(r.species), r.assembly_id, int(r.locus.archetype), int(r.locus.start), int(r.locus.end))
                    target_k = typed_phys.get(phys)
                    if target_k is not None and target_k in by_key:
                        by_key[target_k] = by_key[target_k].merge(r)
                        del by_key[k]
            deduped_list = list(by_key.values())
        else:
            deduped_list = normalized_records

        # Build deduplicated ReferenceStrains table via CSR connected components (`merge_cliques`)
        raw_strains = ReferenceStrains.from_list([r.strain_record for r in deduped_list])
        merged_strains, strain_labels = raw_strains.merge_cliques()

        n = len(deduped_list)
        loci_list = [r.locus if r.locus is not None else empty_locus for r in deduped_list]
        serotypes_arr = np.array([r.serotype for r in deduped_list], dtype="S") if n > 0 else np.empty(0, dtype="S1")

        return cls(
            strains=merged_strains,
            strain_indices=strain_labels,
            serotypes=serotypes_arr,
            loci=Loci.from_list(loci_list),
        )

    @classmethod
    def concat(cls, batches: Iterable[Self]) -> Self:
        """Concatenate and normalize multiple `SeroRefRecords` batches."""
        items: list[SeroRefRecord] = []
        for b in batches:
            items.extend(b)
        return cls.from_list(items)

    def where_species(self, species: str | bytes | int | BacterialSpecies) -> Self:
        """Filter records by canonical `BacterialSpecies`."""
        sp = BacterialSpecies.resolve(species)
        return self[self.species_ids == int(sp)]

    def where_organism(self, organism: bytes | str | BacterialSpecies) -> Self:
        """Filter records by organism name or `BacterialSpecies`."""
        if isinstance(organism, BacterialSpecies):
            return self[self.species_ids == int(organism)]
        sp = BacterialSpecies.resolve(organism)
        if sp != BacterialSpecies.UNKNOWN:
            return self[self.species_ids == int(sp)]
        org_b = organism.encode("utf-8") if isinstance(organism, str) else organism
        return self[self.organisms == org_b]

    def where_serotype(self, serotype: bytes | str) -> Self:
        """Filter records by canonical `serotype` byte string."""
        sero_b = serotype.encode("utf-8") if isinstance(serotype, str) else serotype
        return self[self.serotypes == sero_b]

    def where_antigen_type(self, antigen_type: bytes | str) -> Self:
        """Filter records by derived surface polysaccharide `antigen_type` (`CPS`, `O-antigen`, `EPS`, `WTA`)."""
        at_b = antigen_type.encode("utf-8") if isinstance(antigen_type, str) else antigen_type
        return self[self.antigen_types == at_b]

    def where_assembly_id(self, assembly_id: bytes | str) -> Self:
        """Filter records by INSDC genome or locus `assembly_id`."""
        asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else assembly_id
        return self[self.assembly_ids == asm_b]

    def where_archetype(self, archetype: LocusArchetype | int | str | bytes) -> Self:
        """Filter records by `LocusArchetype`."""
        arch = LocusArchetype.from_name(archetype)
        return self[self.loci.archetypes == int(arch)]

    def save(self, path: Path = DEFAULT_SEROREF_DB_PATH) -> Path:
        """Atomically serialize this `SeroRefRecords` container to a compressed `.npz` archive."""
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_name(f"{path.stem}.{os.getpid()}.tmp.npz")
        np.savez_compressed(
            tmp,
            strain_indices=self.strain_indices,
            serotypes=self.serotypes,
            **self.strains.to_npz_dict(prefix="strain_"),
            **loci_to_npz_dict(self.loci, prefix="locus_"),
        )
        os.replace(tmp, path)
        return path

    @classmethod
    def load(cls, path: Path = DEFAULT_SEROREF_DB_PATH) -> Self:
        """Load a `SeroRefRecords` container from `.npz` (supporting both relational and legacy flat schemas)."""
        if not path.exists():
            return cls.empty()
        with np.load(path, allow_pickle=False) as data:
            files = set(data.files)
            loci = loci_from_npz_dict(data, prefix="locus_")
            if "strain_indices" in files and "strain_species_ids" in files:
                return cls(
                    strains=ReferenceStrains.from_npz_dict(data, prefix="strain_"),
                    strain_indices=data["strain_indices"].astype(np.int32, copy=False),
                    serotypes=data["serotypes"],
                    loci=loci,
                )

            # Transparent migration path from legacy flat schema
            organisms = data["organisms"]
            serotypes = data["serotypes"]
            strains = data["strains"]
            assembly_ids = data["assembly_ids"]
            ncbi_taxids = data["ncbi_taxids"]
            straininfo_ids = data["straininfo_ids"]
            bacdive_ids = data["bacdive_ids"]
            syn_v = data["synonym_values"]
            syn_o = data["synonym_offsets"]
            syn_l = data["synonym_lengths"]
            pm_v = data["pmid_values"]
            pm_o = data["pmid_offsets"]
            pm_l = data["pmid_lengths"]

            migrated: list[SeroRefRecord] = []
            for i in range(len(organisms)):
                s_off, s_ln = int(syn_o[i]), int(syn_l[i])
                syns = tuple(bytes(syn_v[k]) for k in range(s_off, s_off + s_ln))
                p_off, p_ln = int(pm_o[i]), int(pm_l[i])
                pms = tuple(int(pm_v[k]) for k in range(p_off, p_off + p_ln))
                loc_obj = loci[i]
                has_loc = len(loc_obj.nodes) > 0 or loc_obj.archetype != LocusArchetype.UNKNOWN
                migrated.append(
                    SeroRefRecord.from_fields(
                        bytes(organisms[i]),
                        serotype=bytes(serotypes[i]),
                        strain=bytes(strains[i]),
                        strain_synonyms=syns,
                        assembly_id=bytes(assembly_ids[i]),
                        locus=loc_obj if has_loc else None,
                        ncbi_taxid=int(ncbi_taxids[i]),
                        straininfo_id=int(straininfo_ids[i]),
                        bacdive_id=int(bacdive_ids[i]),
                        pmids=pms,
                    )
                )
            return cls.from_list(migrated, deduplicate=True)

species_ids `property` ¶

species_ids: ndarray[tuple[int], dtype[int16]]

1D int16 array ((N,)) of BacterialSpecies IDs for each locus record.

organisms `property` ¶

organisms: ndarray[tuple[int], dtype[bytes_]]

1D byte-string array ((N,)) of canonical species names for each locus record.

strain_names `property` ¶

strain_names: ndarray[tuple[int], dtype[bytes_]]

1D byte-string array ((N,)) of primary reference strain designations for each locus record.

assembly_ids `property` ¶

assembly_ids: ndarray[tuple[int], dtype[bytes_]]

1D byte-string array ((N,)) of INSDC assembly/nucleotide accessions for each locus record.

ncbi_taxids `property` ¶

ncbi_taxids: ndarray[tuple[int], dtype[int32]]

1D int32 array ((N,)) of NCBI Taxonomy IDs for each locus record.

straininfo_ids `property` ¶

straininfo_ids: ndarray[tuple[int], dtype[int32]]

1D int32 array ((N,)) of DSMZ StrainInfo IDs for each locus record.

bacdive_ids `property` ¶

bacdive_ids: ndarray[tuple[int], dtype[int32]]

1D int32 array ((N,)) of DSMZ BacDive IDs for each locus record.

archetypes `property` ¶

archetypes: ndarray[tuple[int], dtype[int8]]

1D int8 array ((N,)) of LocusArchetype integer IDs across all records in the batch.

antigen_types `property` ¶

antigen_types: ndarray[tuple[int], dtype[bytes_]]

1D byte-string array ((N,)) of surface antigen classes (b"CPS", b"O-antigen", b"EPS", b"WTA").

strain_locus_csr `property` ¶

strain_locus_csr: CSR

Sparse (M, N) CSR (seroverse.core.graph.csr.CSR) incidence matrix mapping ReferenceStrains to Loci.

species_locus_csr `property` ¶

species_locus_csr: CSR

Sparse (S, N) CSR (seroverse.core.graph.csr.CSR) incidence matrix mapping BacterialSpecies to Loci.

species_strain_csr `property` ¶

species_strain_csr: CSR

Sparse (S, M) CSR (seroverse.core.graph.csr.CSR) incidence matrix mapping BacterialSpecies to ReferenceStrains.

empty `classmethod` ¶

empty() -> Self

Create an empty 0-length SeroRefRecords container.

Source code in src/seroref/db/records.py

@classmethod
def empty(cls) -> Self:
    """Create an empty 0-length `SeroRefRecords` container."""
    return cls(
        strains=ReferenceStrains.empty(),
        strain_indices=np.empty(0, dtype=np.int32),
        serotypes=np.empty(0, dtype="S1"),
        loci=Loci.empty(),
    )

from_list `classmethod` ¶

from_list(
    records: Sequence[SeroRefRecord],
    *,
    deduplicate: bool = True,
) -> Self

Compile a sequence of SeroRefRecord scalar views into a normalized SeroRefRecords container.

Deduplicates ReferenceStrain entries into self.strains, merges untyped/typed duplicate physical loci when deduplicate=True, and synchronizes Loci.names with self.serotypes.

Source code in src/seroref/db/records.py

@classmethod
def from_list(cls, records: Sequence[SeroRefRecord], *, deduplicate: bool = True) -> Self:
    """Compile a sequence of `SeroRefRecord` scalar views into a normalized `SeroRefRecords` container.

    Deduplicates `ReferenceStrain` entries into `self.strains`, merges untyped/typed duplicate
    physical loci when `deduplicate=True`, and synchronizes `Loci.names` with `self.serotypes`.
    """
    if len(records) == 0:
        return cls.empty()

    empty_locus = Locus(nodes=(), edges=())
    normalized_records: list[SeroRefRecord] = []
    for r in records:
        loc = r.locus if r.locus is not None else empty_locus
        sero_b = r.serotype
        if sero_b:
            parsed = r.species.parse_serotype(sero_b, archetype=loc.archetype)
            sero_b = parsed[0][1] if parsed else b""
        if not sero_b and loc.archetype in INVARIANT_ARCHETYPE_GLYCAN_NAMES:
            sero_b = INVARIANT_ARCHETYPE_GLYCAN_NAMES[loc.archetype]
        if sero_b:
            if loc.name != sero_b:
                loc = loc.with_name(sero_b)
        elif loc.name:
            loc = loc.with_name(b"")
        has_loc = len(loc.nodes) > 0 or loc.archetype != LocusArchetype.UNKNOWN
        normalized_records.append(
            SeroRefRecord(
                strain_record=r.strain_record,
                serotype=sero_b,
                locus=loc if has_loc else None,
            )
        )

    if deduplicate:
        by_key: dict[tuple[Any, ...], SeroRefRecord] = {}
        for r in normalized_records:
            loc = r.locus
            arch = int(loc.archetype) if loc is not None else int(LocusArchetype.UNKNOWN)
            start = int(loc.start) if loc is not None else 0
            end = int(loc.end) if loc is not None else 0
            if r.assembly_id and loc is not None and len(loc.nodes) > 0:
                key: tuple[Any, ...] = ("asm_loc", int(r.species), r.assembly_id, arch, start, end, r.serotype)
            else:
                key = ("meta", int(r.species), r.assembly_id, r.strain, arch, r.serotype, start, end)
            existing = by_key.get(key)
            if existing is None:
                by_key[key] = r
            else:
                by_key[key] = existing.merge(r)

        # Fold untyped (serotype == b"") records into typed records on the exact same physical locus
        typed_phys: dict[tuple[int, bytes, int, int, int], tuple[Any, ...]] = {}
        for k, r in by_key.items():
            if r.serotype and r.assembly_id and r.locus is not None:
                phys = (int(r.species), r.assembly_id, int(r.locus.archetype), int(r.locus.start), int(r.locus.end))
                typed_phys[phys] = k

        deduped_list: list[SeroRefRecord] = []
        for k, r in list(by_key.items()):
            if not r.serotype and r.assembly_id and r.locus is not None:
                phys = (int(r.species), r.assembly_id, int(r.locus.archetype), int(r.locus.start), int(r.locus.end))
                target_k = typed_phys.get(phys)
                if target_k is not None and target_k in by_key:
                    by_key[target_k] = by_key[target_k].merge(r)
                    del by_key[k]
        deduped_list = list(by_key.values())
    else:
        deduped_list = normalized_records

    # Build deduplicated ReferenceStrains table via CSR connected components (`merge_cliques`)
    raw_strains = ReferenceStrains.from_list([r.strain_record for r in deduped_list])
    merged_strains, strain_labels = raw_strains.merge_cliques()

    n = len(deduped_list)
    loci_list = [r.locus if r.locus is not None else empty_locus for r in deduped_list]
    serotypes_arr = np.array([r.serotype for r in deduped_list], dtype="S") if n > 0 else np.empty(0, dtype="S1")

    return cls(
        strains=merged_strains,
        strain_indices=strain_labels,
        serotypes=serotypes_arr,
        loci=Loci.from_list(loci_list),
    )

concat `classmethod` ¶

concat(batches: Iterable[Self]) -> Self

Concatenate and normalize multiple SeroRefRecords batches.

Source code in src/seroref/db/records.py

@classmethod
def concat(cls, batches: Iterable[Self]) -> Self:
    """Concatenate and normalize multiple `SeroRefRecords` batches."""
    items: list[SeroRefRecord] = []
    for b in batches:
        items.extend(b)
    return cls.from_list(items)

where_species ¶

where_species(
    species: str | bytes | int | BacterialSpecies,
) -> Self

Filter records by canonical BacterialSpecies.

Source code in src/seroref/db/records.py

def where_species(self, species: str | bytes | int | BacterialSpecies) -> Self:
    """Filter records by canonical `BacterialSpecies`."""
    sp = BacterialSpecies.resolve(species)
    return self[self.species_ids == int(sp)]

where_organism ¶

where_organism(
    organism: bytes | str | BacterialSpecies,
) -> Self

Filter records by organism name or BacterialSpecies.

Source code in src/seroref/db/records.py

def where_organism(self, organism: bytes | str | BacterialSpecies) -> Self:
    """Filter records by organism name or `BacterialSpecies`."""
    if isinstance(organism, BacterialSpecies):
        return self[self.species_ids == int(organism)]
    sp = BacterialSpecies.resolve(organism)
    if sp != BacterialSpecies.UNKNOWN:
        return self[self.species_ids == int(sp)]
    org_b = organism.encode("utf-8") if isinstance(organism, str) else organism
    return self[self.organisms == org_b]

where_serotype ¶

where_serotype(serotype: bytes | str) -> Self

Filter records by canonical serotype byte string.

Source code in src/seroref/db/records.py

def where_serotype(self, serotype: bytes | str) -> Self:
    """Filter records by canonical `serotype` byte string."""
    sero_b = serotype.encode("utf-8") if isinstance(serotype, str) else serotype
    return self[self.serotypes == sero_b]

where_antigen_type ¶

where_antigen_type(antigen_type: bytes | str) -> Self

Filter records by derived surface polysaccharide antigen_type (CPS, O-antigen, EPS, WTA).

Source code in src/seroref/db/records.py

def where_antigen_type(self, antigen_type: bytes | str) -> Self:
    """Filter records by derived surface polysaccharide `antigen_type` (`CPS`, `O-antigen`, `EPS`, `WTA`)."""
    at_b = antigen_type.encode("utf-8") if isinstance(antigen_type, str) else antigen_type
    return self[self.antigen_types == at_b]

where_assembly_id ¶

where_assembly_id(assembly_id: bytes | str) -> Self

Filter records by INSDC genome or locus assembly_id.

Source code in src/seroref/db/records.py

def where_assembly_id(self, assembly_id: bytes | str) -> Self:
    """Filter records by INSDC genome or locus `assembly_id`."""
    asm_b = assembly_id.encode("utf-8") if isinstance(assembly_id, str) else assembly_id
    return self[self.assembly_ids == asm_b]

where_archetype ¶

where_archetype(
    archetype: LocusArchetype | int | str | bytes,
) -> Self

Filter records by LocusArchetype.

Source code in src/seroref/db/records.py

def where_archetype(self, archetype: LocusArchetype | int | str | bytes) -> Self:
    """Filter records by `LocusArchetype`."""
    arch = LocusArchetype.from_name(archetype)
    return self[self.loci.archetypes == int(arch)]

save ¶

save(path: Path = DEFAULT_SEROREF_DB_PATH) -> Path

Atomically serialize this SeroRefRecords container to a compressed .npz archive.

Source code in src/seroref/db/records.py

def save(self, path: Path = DEFAULT_SEROREF_DB_PATH) -> Path:
    """Atomically serialize this `SeroRefRecords` container to a compressed `.npz` archive."""
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f"{path.stem}.{os.getpid()}.tmp.npz")
    np.savez_compressed(
        tmp,
        strain_indices=self.strain_indices,
        serotypes=self.serotypes,
        **self.strains.to_npz_dict(prefix="strain_"),
        **loci_to_npz_dict(self.loci, prefix="locus_"),
    )
    os.replace(tmp, path)
    return path

load `classmethod` ¶

load(path: Path = DEFAULT_SEROREF_DB_PATH) -> Self

Load a SeroRefRecords container from .npz (supporting both relational and legacy flat schemas).

Source code in src/seroref/db/records.py

@classmethod
def load(cls, path: Path = DEFAULT_SEROREF_DB_PATH) -> Self:
    """Load a `SeroRefRecords` container from `.npz` (supporting both relational and legacy flat schemas)."""
    if not path.exists():
        return cls.empty()
    with np.load(path, allow_pickle=False) as data:
        files = set(data.files)
        loci = loci_from_npz_dict(data, prefix="locus_")
        if "strain_indices" in files and "strain_species_ids" in files:
            return cls(
                strains=ReferenceStrains.from_npz_dict(data, prefix="strain_"),
                strain_indices=data["strain_indices"].astype(np.int32, copy=False),
                serotypes=data["serotypes"],
                loci=loci,
            )

        # Transparent migration path from legacy flat schema
        organisms = data["organisms"]
        serotypes = data["serotypes"]
        strains = data["strains"]
        assembly_ids = data["assembly_ids"]
        ncbi_taxids = data["ncbi_taxids"]
        straininfo_ids = data["straininfo_ids"]
        bacdive_ids = data["bacdive_ids"]
        syn_v = data["synonym_values"]
        syn_o = data["synonym_offsets"]
        syn_l = data["synonym_lengths"]
        pm_v = data["pmid_values"]
        pm_o = data["pmid_offsets"]
        pm_l = data["pmid_lengths"]

        migrated: list[SeroRefRecord] = []
        for i in range(len(organisms)):
            s_off, s_ln = int(syn_o[i]), int(syn_l[i])
            syns = tuple(bytes(syn_v[k]) for k in range(s_off, s_off + s_ln))
            p_off, p_ln = int(pm_o[i]), int(pm_l[i])
            pms = tuple(int(pm_v[k]) for k in range(p_off, p_off + p_ln))
            loc_obj = loci[i]
            has_loc = len(loc_obj.nodes) > 0 or loc_obj.archetype != LocusArchetype.UNKNOWN
            migrated.append(
                SeroRefRecord.from_fields(
                    bytes(organisms[i]),
                    serotype=bytes(serotypes[i]),
                    strain=bytes(strains[i]),
                    strain_synonyms=syns,
                    assembly_id=bytes(assembly_ids[i]),
                    locus=loc_obj if has_loc else None,
                    ncbi_taxid=int(ncbi_taxids[i]),
                    straininfo_id=int(straininfo_ids[i]),
                    bacdive_id=int(bacdive_ids[i]),
                    pmids=pms,
                )
            )
        return cls.from_list(migrated, deduplicate=True)

loci_to_npz_dict ¶

loci_to_npz_dict(
    loci: Loci, prefix: str = "locus_"
) -> dict[str, Any]

Serialize a Loci SoA batch (and its embedded Genes) into a flat .npz-ready dictionary.

Source code in src/seroref/db/records.py

def loci_to_npz_dict(loci: Loci, prefix: str = "locus_") -> dict[str, Any]:
    """Serialize a `Loci` SoA batch (and its embedded `Genes`) into a flat `.npz`-ready dictionary."""
    g = loci.genes
    return {
        f"{prefix}node_types": loci.node_types,
        f"{prefix}node_offsets": loci.node_offsets,
        f"{prefix}edge_sources": loci.edge_sources,
        f"{prefix}edge_targets": loci.edge_targets,
        f"{prefix}edge_types": loci.edge_types,
        f"{prefix}edge_offsets": loci.edge_offsets,
        f"{prefix}starts": loci.intervals.starts,
        f"{prefix}ends": loci.intervals.ends,
        f"{prefix}strands": loci.intervals.strands,
        f"{prefix}contig_indices": loci.contig_indices,
        f"{prefix}archetypes": loci.archetypes,
        f"{prefix}completeness": loci.completeness,
        f"{prefix}scores": loci.scores,
        f"{prefix}is_tandem_split": loci.is_tandem_split,
        f"{prefix}names": loci.names if len(loci.names) > 0 else np.empty(0, dtype="S1"),
        f"{prefix}gene_starts": g.intervals.starts,
        f"{prefix}gene_ends": g.intervals.ends,
        f"{prefix}gene_strands": g.intervals.strands,
        f"{prefix}gene_contig_indices": g.contig_indices,
        f"{prefix}gene_is_partial": g.is_partial,
        f"{prefix}gene_genetic_code_id": np.array([g.genetic_code.table_id], dtype=np.int32),
        f"{prefix}gene_seq_ids": g.seqs.ids,
        f"{prefix}gene_seq_bytes": g.seqs.seqs,
        f"{prefix}gene_seq_offsets": g.seqs.offsets,
        f"{prefix}gene_seq_lengths": g.seqs.lengths,
        f"{prefix}gene_indel_positions": g.indels.positions,
        f"{prefix}gene_indel_kinds": g.indels.kinds,
        f"{prefix}gene_indel_offsets": g.indels.offsets,
        f"{prefix}gene_indel_lengths": g.indels.lengths,
    }

loci_from_npz_dict ¶

loci_from_npz_dict(
    data: Any, prefix: str = "locus_"
) -> Loci

Deserialize a Loci SoA batch from a flat .npz archive dictionary.

Source code in src/seroref/db/records.py

def loci_from_npz_dict(data: Any, prefix: str = "locus_") -> Loci:
    """Deserialize a `Loci` SoA batch from a flat `.npz` archive dictionary."""
    files = set(data.files) if hasattr(data, "files") else set(data.keys())
    g_starts = data[f"{prefix}gene_starts"]
    n_genes = len(g_starts)
    gene_intervals = Intervals(
        starts=g_starts,
        ends=data[f"{prefix}gene_ends"],
        strands=data[f"{prefix}gene_strands"],
    )
    seq_lengths = data[f"{prefix}gene_seq_lengths"]
    gene_seqs = (
        Sequences(
            ids=data[f"{prefix}gene_seq_ids"],
            seqs=data[f"{prefix}gene_seq_bytes"],
            offsets=data[f"{prefix}gene_seq_offsets"],
            lengths=seq_lengths,
        )
        if len(seq_lengths) > 0
        else Sequences.empty()
    )
    gene_indels = Indels(
        positions=data[f"{prefix}gene_indel_positions"],
        kinds=data[f"{prefix}gene_indel_kinds"],
        offsets=data[f"{prefix}gene_indel_offsets"],
        lengths=data[f"{prefix}gene_indel_lengths"],
    )
    is_partial = (
        data[f"{prefix}gene_is_partial"].astype(np.bool_)
        if f"{prefix}gene_is_partial" in files
        else np.zeros(n_genes, dtype=np.bool_)
    )
    gc_id = int(data[f"{prefix}gene_genetic_code_id"][0]) if f"{prefix}gene_genetic_code_id" in files else 11
    genes = Genes(
        intervals=gene_intervals,
        seqs=gene_seqs,
        contig_indices=data[f"{prefix}gene_contig_indices"],
        indels=gene_indels,
        is_partial=is_partial,
        genetic_code=GeneticCode.from_table(gc_id),
    )
    node_offsets = data[f"{prefix}node_offsets"]
    n_loci = max(0, len(node_offsets) - 1)
    names_raw = data[f"{prefix}names"] if f"{prefix}names" in files else np.full(n_loci, b"", dtype="S1")
    names_arr = np.char.encode(names_raw, "utf-8") if names_raw.dtype.kind == "U" else names_raw
    return Loci(
        node_types=data[f"{prefix}node_types"],
        node_offsets=node_offsets,
        edge_sources=data[f"{prefix}edge_sources"],
        edge_targets=data[f"{prefix}edge_targets"],
        edge_types=data[f"{prefix}edge_types"],
        edge_offsets=data[f"{prefix}edge_offsets"],
        genes=genes,
        intervals=Intervals(
            starts=data[f"{prefix}starts"],
            ends=data[f"{prefix}ends"],
            strands=data[f"{prefix}strands"],
        ),
        contig_indices=data[f"{prefix}contig_indices"],
        archetypes=data[f"{prefix}archetypes"],
        completeness=data[f"{prefix}completeness"],
        scores=data[f"{prefix}scores"],
        is_tandem_split=data[f"{prefix}is_tandem_split"],
        names=names_arr,
    )

load_seroref_records ¶

load_seroref_records(
    path: Path = DEFAULT_SEROREF_DB_PATH,
) -> SeroRefRecords

Load curated SeroRefRecords from seroref.npz.

Source code in src/seroref/db/records.py

def load_seroref_records(path: Path = DEFAULT_SEROREF_DB_PATH) -> SeroRefRecords:
    """Load curated `SeroRefRecords` from `seroref.npz`."""
    return SeroRefRecords.load(path)

seroref.db¶

Species Specifications & Serotype Rules (seroref.db.species)¶

species ¶

GramStain ¶

label property ¶

label_str property ¶

from_name classmethod ¶

resolve classmethod ¶

SerotypeRule dataclass ¶

SpeciesSpec dataclass ¶

scientific_name property ¶

synonyms property ¶

name_str property ¶

scientific_name_str property ¶

genus property ¶

archetypes property ¶

supports_archetype ¶

is_strain_or_mutant_token ¶

canonicalize_token ¶

parse_serotype ¶

normalize_strain ¶

SpeciesSpecs dataclass ¶

scientific_names property ¶

alias_offsets property ¶

alias_lengths property ¶

empty classmethod ¶

from_list classmethod ¶

concat classmethod ¶

BacterialSpecies ¶

spec property ¶

name_bytes property ¶

scientific_name property ¶

name_str property ¶

scientific_name_str property ¶

ncbi_taxid property ¶

gram_stain property ¶

default_category property ¶

archetype_mask property ¶

aliases property ¶

synonyms property ¶

landmark_accessions property ¶

parse_serotype ¶

normalize_strain ¶

from_name classmethod ¶

resolve classmethod ¶

from_taxid classmethod ¶

specs classmethod ¶

archetypes_to_mask ¶

mask_to_archetypes ¶

Scalar Views (seroref.db.models)¶

models ¶

ReferenceStrain dataclass ¶

organism property ¶

effective_taxid property ¶

compound_key property ¶

from_raw classmethod ¶

with_passport ¶

merge ¶

bind_loci ¶

SeroRefRecord dataclass ¶

species property ¶

organism property ¶

strain property ¶

strain_synonyms property ¶

assembly_id property ¶

ncbi_taxid property ¶

straininfo_id property ¶

bacdive_id property ¶

pmids property ¶

archetype property ¶

category property ¶

antigen_type property ¶

has_locus property ¶

compound_key property ¶

from_fields classmethod ¶

merge ¶

archetype_to_antigen_type ¶

Batched Reference Database Containers (seroref.db.records)¶

records ¶

ReferenceStrains dataclass ¶

`seroref.db`¶

Species Specifications & Serotype Rules (`seroref.db.species`)¶

label `property` ¶

label_str `property` ¶

from_name `classmethod` ¶

resolve `classmethod` ¶

SerotypeRule `dataclass` ¶

SpeciesSpec `dataclass` ¶

scientific_name `property` ¶

synonyms `property` ¶

name_str `property` ¶

scientific_name_str `property` ¶

genus `property` ¶

archetypes `property` ¶

SpeciesSpecs `dataclass` ¶

scientific_names `property` ¶

alias_offsets `property` ¶

alias_lengths `property` ¶

empty `classmethod` ¶

from_list `classmethod` ¶

concat `classmethod` ¶

spec `property` ¶

name_bytes `property` ¶

scientific_name `property` ¶

name_str `property` ¶

scientific_name_str `property` ¶

ncbi_taxid `property` ¶

gram_stain `property` ¶

default_category `property` ¶

archetype_mask `property` ¶

aliases `property` ¶

synonyms `property` ¶

landmark_accessions `property` ¶

from_name `classmethod` ¶

resolve `classmethod` ¶

from_taxid `classmethod` ¶

specs `classmethod` ¶

Scalar Views (`seroref.db.models`)¶

ReferenceStrain `dataclass` ¶

organism `property` ¶

effective_taxid `property` ¶

compound_key `property` ¶

from_raw `classmethod` ¶

SeroRefRecord `dataclass` ¶

species `property` ¶

organism `property` ¶

strain `property` ¶

strain_synonyms `property` ¶

assembly_id `property` ¶

ncbi_taxid `property` ¶

straininfo_id `property` ¶

bacdive_id `property` ¶

pmids `property` ¶

archetype `property` ¶

category `property` ¶

antigen_type `property` ¶

has_locus `property` ¶

compound_key `property` ¶

from_fields `classmethod` ¶

Batched Reference Database Containers (`seroref.db.records`)¶

ReferenceStrains `dataclass` ¶

organisms `property` ¶

synonym_offsets `property` ¶

synonym_lengths `property` ¶

empty `classmethod` ¶

from_list `classmethod` ¶

concat `classmethod` ¶

from_npz_dict `classmethod` ¶

SeroRefRecords `dataclass` ¶

species_ids `property` ¶

organisms `property` ¶

strain_names `property` ¶

assembly_ids `property` ¶

ncbi_taxids `property` ¶

straininfo_ids `property` ¶

bacdive_ids `property` ¶

archetypes `property` ¶

antigen_types `property` ¶

strain_locus_csr `property` ¶

species_locus_csr `property` ¶