`seroref.genes`¶

Single-pass bacterial and prophage Open Reading Frame (ORF) prediction, functional gene role ontology, and multi-channel 1D-CNN protein classification.

ORF Prediction (`seroref.genes.orf.predictor`)¶

predictor ¶

OrfPredictor high-level execution engine for sequences and assemblies.

Classes:

Name	Description
`OrfPredictor`	High-performance Open Reading Frame (ORF) predictor.

OrfPredictor ¶

High-performance Open Reading Frame (ORF) predictor.

Executes Numba JIT-compiled FragGeneScan HMM decoding over single sequences, seroverse.core.seq.seq.Sequences, or seroverse.core.genome.GenomeAssembly containers using a unified flat window table and Numba prange parallelism, returning batched Genes.

Attributes:

Name	Type	Description
`model`	`HMM`	Baked HMM model parameter container.
`genetic_code`	`GeneticCode`	Translation table defining start/stop codons and amino acid mappings.
`min_gene_len`	`int`	Minimum gene length in nucleotides.
`local_gc_window`	`int`	Sliding window size in bp for local GC bin adaptation.

Methods:

Name	Description
`predict`	Predict Open Reading Frames across target sequences.

Source code in src/seroref/genes/orf/predictor.py

class OrfPredictor:
    r"""High-performance Open Reading Frame (ORF) predictor.

    Executes Numba JIT-compiled FragGeneScan HMM decoding over single sequences,
    `seroverse.core.seq.seq.Sequences`, or `seroverse.core.genome.GenomeAssembly`
    containers using a unified flat window table and Numba `prange` parallelism, returning
    batched [`Genes`][seroref.genes.models.Genes].

    Attributes:
        model (HMM): Baked HMM model parameter container.
        genetic_code (GeneticCode): Translation table defining start/stop codons and amino acid mappings.
        min_gene_len (int): Minimum gene length in nucleotides.
        local_gc_window (int): Sliding window size in bp for local GC bin adaptation.
    """

    def __init__(
        self,
        model: HMM | None = None,
        window_size: int = 50_000,
        overlap: int = 5_000,
        genetic_code: GeneticCode | int | bytes | str = GeneticCode.bacterial,
        min_gene_len: int = 120,
        local_gc_window: int = 4_000,
    ) -> None:
        r"""Initialize an [`OrfPredictor`][seroref.genes.orf.predictor.OrfPredictor] instance.

        Args:
            model (HMM | None): Optional custom HMM parameters. If None, uses default baked model.
            window_size (int): Size in base pairs for local GC-adaptive sliding windows on long contigs.
                Set to 0 to disable windowing. Defaults to 50_000.
            overlap (int): Overlap size in base pairs between consecutive windows to ensure seamless
                gene boundary detection. Defaults to 5_000.
            genetic_code (GeneticCode | int | bytes | str): Translation table (instance, NCBI table ID, or name).
                Defaults to `GeneticCode.bacterial` (Table 11).
            min_gene_len (int): Minimum ORF length threshold in base pairs. Defaults to 120 bp.
            local_gc_window (int): Sliding window size in base pairs for local GC bin adaptation.
                Defaults to 4_000.
        """
        self.model = model if model is not None else get_hmm_model()
        self.window_size = window_size
        self.overlap = overlap
        gc_key = genetic_code.encode("utf-8") if isinstance(genetic_code, str) else genetic_code
        self.genetic_code = gc_key if isinstance(gc_key, GeneticCode) else GeneticCode.from_table(gc_key)
        self.min_gene_len = int(min_gene_len)
        self.local_gc_window = int(local_gc_window)

    def _predict_flat_batch(
        self,
        flat_seqs: np.ndarray,  # type: ignore[type-arg]
        contig_offsets: np.ndarray,  # type: ignore[type-arg]
        contig_lengths: np.ndarray,  # type: ignore[type-arg]
    ) -> Genes:
        r"""Decode all windows across all contigs in parallel via Numba `prange` and build SoA `Genes`."""
        n_contigs = len(contig_lengths)
        if n_contigs == 0 or len(flat_seqs) == 0:
            return Genes.empty()

        min_contig_len = min(self.min_gene_len, 60)
        win_table = _build_window_table(
            contig_offsets,
            contig_lengths,
            self.window_size,
            self.overlap,
            min_contig_len,
        )

        n_windows = len(win_table[0])
        if n_windows == 0:
            return Genes.empty()

        m = self.model
        gc = self.genetic_code
        encode_lut = Alphabet.DNA.kmer_lut
        lut_rc = Alphabet.DNA.complement_lut
        viterbi_params = (m.tr, m.tr_mi, m.tr_ii, m.pi)
        score_params = (
            m.e_match,
            m.tr_rr,
            m.tr_start,
            m.tr_stop,
            m.dist_start,
            m.dist_stop,
            gc.start_mask_64,
            gc.stop_mask_64,
            gc.rev_start_mask_64,
            gc.rev_stop_mask_64,
            gc.start_log_priors_64,
            gc.rev_start_log_priors_64,
            gc.stop_log_priors_64,
            gc.rev_stop_log_priors_64,
        )

        (
            encoded_flat,
            vpath_flat,
            cg_bins_flat,
            win_gene_counts,
            win_cds_bytes,
            win_indel_counts,
        ) = _decode_and_count_windows(
            flat_seqs,
            encode_lut,
            lut_rc,
            win_table,
            self.min_gene_len,
            self.local_gc_window,
            viterbi_params,
            score_params,
        )

        total_genes = int(np.sum(win_gene_counts))
        if total_genes == 0:
            return Genes.empty()

        total_cds = int(np.sum(win_cds_bytes))
        total_indels = int(np.sum(win_indel_counts))

        win_gene_offsets = np.zeros(n_windows, dtype=np.int32)
        win_cds_offsets = np.zeros(n_windows, dtype=np.int32)
        win_indel_offsets = np.zeros(n_windows, dtype=np.int32)
        if n_windows > 1:
            np.cumsum(win_gene_counts[:-1], out=win_gene_offsets[1:])
            np.cumsum(win_cds_bytes[:-1], out=win_cds_offsets[1:])
            np.cumsum(win_indel_counts[:-1], out=win_indel_offsets[1:])

        out_starts = np.empty(total_genes, dtype=np.int32)
        out_ends = np.empty(total_genes, dtype=np.int32)
        out_strands = np.empty(total_genes, dtype=np.int8)
        out_partials = np.empty(total_genes, dtype=np.bool_)
        out_contig_indices = np.empty(total_genes, dtype=np.uint32)

        out_cds_seqs = np.empty(total_cds, dtype=np.uint8)
        out_cds_offsets = np.empty(total_genes, dtype=np.int32)
        out_cds_lengths = np.empty(total_genes, dtype=np.int32)

        out_indel_pos = np.empty(total_indels, dtype=np.int32)
        out_indel_kinds = np.empty(total_indels, dtype=np.int8)
        out_indel_offsets = np.empty(total_genes, dtype=np.int32)
        out_indel_lengths = np.empty(total_genes, dtype=np.int32)

        win_offsets = (win_gene_offsets, win_gene_counts, win_cds_offsets, win_indel_offsets)
        out_bufs = (
            out_starts,
            out_ends,
            out_strands,
            out_partials,
            out_contig_indices,
            out_cds_seqs,
            out_cds_offsets,
            out_cds_lengths,
            out_indel_pos,
            out_indel_kinds,
            out_indel_offsets,
            out_indel_lengths,
        )

        _write_windows_parallel(
            flat_seqs,
            encoded_flat,
            vpath_flat,
            cg_bins_flat,
            lut_rc,
            win_table,
            win_offsets,
            self.min_gene_len,
            score_params,
            out_bufs,
        )

        ids_arr = np.arange(total_genes).astype("S")
        return Genes(
            intervals=Intervals(starts=out_starts, ends=out_ends, strands=out_strands),
            seqs=Sequences(ids_arr, out_cds_seqs, out_cds_offsets, out_cds_lengths),
            contig_indices=out_contig_indices,
            indels=Indels(
                positions=out_indel_pos,
                kinds=out_indel_kinds,
                offsets=out_indel_offsets,
                lengths=out_indel_lengths,
            ),
            is_partial=out_partials,
            genetic_code=gc,
        )

    def predict(
        self,
        target: bytes | str | Sequence | Sequences | GenomeAssembly,
    ) -> Genes:
        r"""Predict Open Reading Frames across target sequences.

        Args:
            target (bytes | str | Sequence | Sequences | GenomeAssembly): Input genomic sequence(s).

        Returns:
            Genes: Batched [`Genes`][seroref.genes.models.Genes] container containing interval coordinates,
                in-frame coding sequences (5' -> 3'), contig assignments, frameshift indels, and `is_partial` mask.

        Raises:
            TypeError: If target is not a supported sequence type.
        """
        if isinstance(target, (str, bytes)):
            raw_arr = np.frombuffer(target.encode("ascii") if isinstance(target, str) else target, dtype=np.uint8)
            return self._predict_flat_batch(
                raw_arr,
                np.zeros(1, dtype=np.int32),
                np.array([len(raw_arr)], dtype=np.int32),
            )

        if isinstance(target, Sequence):
            raw_arr = np.frombuffer(target.seq, dtype=np.uint8)
            return self._predict_flat_batch(
                raw_arr,
                np.zeros(1, dtype=np.int32),
                np.array([len(raw_arr)], dtype=np.int32),
            )

        if isinstance(target, (Sequences, GenomeAssembly)):
            if len(target) == 0:
                return Genes.empty()
            return self._predict_flat_batch(target.seqs, target.offsets, target.lengths)

        raise TypeError(f"Unsupported target type for ORF prediction: {type(target)}")

predict ¶

predict(
    target: bytes
    | str
    | Sequence
    | Sequences
    | GenomeAssembly,
) -> Genes

Predict Open Reading Frames across target sequences.

Parameters:

Name	Type	Description	Default
`target`	`bytes \| str \| Sequence \| Sequences \| GenomeAssembly`	Input genomic sequence(s).	required

Returns:

Name	Type	Description
`Genes`	`Genes`	Batched `Genes` container containing interval coordinates, in-frame coding sequences (5' -> 3'), contig assignments, frameshift indels, and `is_partial` mask.

Raises:

Type	Description
`TypeError`	If target is not a supported sequence type.

Source code in src/seroref/genes/orf/predictor.py

def predict(
    self,
    target: bytes | str | Sequence | Sequences | GenomeAssembly,
) -> Genes:
    r"""Predict Open Reading Frames across target sequences.

    Args:
        target (bytes | str | Sequence | Sequences | GenomeAssembly): Input genomic sequence(s).

    Returns:
        Genes: Batched [`Genes`][seroref.genes.models.Genes] container containing interval coordinates,
            in-frame coding sequences (5' -> 3'), contig assignments, frameshift indels, and `is_partial` mask.

    Raises:
        TypeError: If target is not a supported sequence type.
    """
    if isinstance(target, (str, bytes)):
        raw_arr = np.frombuffer(target.encode("ascii") if isinstance(target, str) else target, dtype=np.uint8)
        return self._predict_flat_batch(
            raw_arr,
            np.zeros(1, dtype=np.int32),
            np.array([len(raw_arr)], dtype=np.int32),
        )

    if isinstance(target, Sequence):
        raw_arr = np.frombuffer(target.seq, dtype=np.uint8)
        return self._predict_flat_batch(
            raw_arr,
            np.zeros(1, dtype=np.int32),
            np.array([len(raw_arr)], dtype=np.int32),
        )

    if isinstance(target, (Sequences, GenomeAssembly)):
        if len(target) == 0:
            return Genes.empty()
        return self._predict_flat_batch(target.seqs, target.offsets, target.lengths)

    raise TypeError(f"Unsupported target type for ORF prediction: {type(target)}")

Gene Containers & HMM Parameters (`seroref.genes.models`)¶

models ¶

Structure-of-Arrays data containers for predicted genes, indels, and HMM parameters.

Classes:

Name	Description
`IndelKind`	Indel mutation type for frameshift tracking.
`IndelRecord`	Scalar view of frameshift indel events for a single gene.
`Indels`	Ragged Structure-of-Arrays container for frameshift indels across a Genes batch.
`Gene`	Scalar view of a single predicted gene.
`Genes`	Structure-of-Arrays (SoA) container for predicted genes.
`HMM`	Container holding baked FragGeneScan HMM parameters.

Functions:

Name	Description
`get_hmm_model`	Return cached singleton instance of the default bundled HMM model.

IndelKind ¶

Bases: IntEnum

Indel mutation type for frameshift tracking.

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

class IndelKind(IntEnum):
    r"""Indel mutation type for frameshift tracking."""

    INSERTION = 1
    DELETION = -1

IndelRecord `dataclass` ¶

Scalar view of frameshift indel events for a single gene.

Attributes:

Name	Type	Description
`positions`	`ndarray`	0-based contig coordinates where indels occurred.
`kinds`	`ndarray`	Indel mutation types (+1 for Insertion, -1 for Deletion).

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

@dataclass(slots=True, frozen=True, eq=False)
class IndelRecord:
    r"""Scalar view of frameshift indel events for a single gene.

    Attributes:
        positions (np.ndarray): 0-based contig coordinates where indels occurred.
        kinds (np.ndarray): Indel mutation types (+1 for Insertion, -1 for Deletion).
    """

    positions: np.ndarray[tuple[int], np.dtype[np.int32]]
    kinds: np.ndarray[tuple[int], np.dtype[np.int8]]

    def __len__(self) -> int:
        r"""Return the number of indel events in this record."""
        return len(self.positions)

    def __eq__(self, other: object) -> bool:
        r"""Compare two IndelRecord views for exact coordinate and kind equality."""
        if not isinstance(other, IndelRecord):
            return NotImplemented
        return bool(np.array_equal(self.positions, other.positions) and np.array_equal(self.kinds, other.kinds))

Indels `dataclass` ¶

Bases: RaggedArrayContainer[IndelRecord, 'Indels']

Ragged Structure-of-Arrays container for frameshift indels across a Genes batch.

Stores variable-length indel event coordinates and mutation kinds in flat contiguous arrays, indexed via offsets and lengths per gene.

Attributes:

Name	Type	Description
`positions`	`ndarray`	1D array of 0-based contig coordinates across all indels.
`kinds`	`ndarray`	1D array of indel kinds (+1=ins, -1=del).
`offsets`	`ndarray`	Starting index into `positions` and `kinds` for each gene.
`lengths`	`ndarray`	Number of indels for each gene.

Methods:

Name	Description
`empty`	Create an empty Indels container.
`concat`	Concatenate multiple Indels containers into a single batch.

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

@dataclass(slots=True, frozen=True)
class Indels(RaggedArrayContainer[IndelRecord, "Indels"]):
    r"""Ragged Structure-of-Arrays container for frameshift indels across a Genes batch.

    Stores variable-length indel event coordinates and mutation kinds in flat contiguous arrays,
    indexed via `offsets` and `lengths` per gene.

    Attributes:
        positions (np.ndarray): 1D array of 0-based contig coordinates across all indels.
        kinds (np.ndarray): 1D array of indel kinds (+1=ins, -1=del).
        offsets (np.ndarray): Starting index into `positions` and `kinds` for each gene.
        lengths (np.ndarray): Number of indels for each gene.
    """

    positions: np.ndarray[tuple[int], np.dtype[np.int32]]
    kinds: np.ndarray[tuple[int], np.dtype[np.int8]]
    offsets: np.ndarray[tuple[int], np.dtype[np.int32]]
    lengths: np.ndarray[tuple[int], np.dtype[np.int32]]

    def __len__(self) -> int:
        r"""Return the number of genes in the batch."""
        return len(self.offsets)

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

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

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

    def __getitem__(
        self, item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any]
    ) -> IndelRecord | Indels:
        r"""Access indel records by index, slice, or boolean mask.

        Args:
            item (int | slice | np.ndarray | list): Integer index, slice, or mask/indices array.

        Returns:
            IndelRecord | Indels: A single gene's indel record or a sliced [`Indels`][serogen.bgc.genes.Indels] batch.

        Raises:
            IndexError: If an integer index is out of range.
        """
        if isinstance(item, (int, np.integer)):
            idx = int(item)
            if idx < 0:
                idx += len(self)
            if idx < 0 or idx >= len(self):
                raise IndexError("Batch index out of range")
            off_val = int(self.offsets[idx])
            len_val = int(self.lengths[idx])
            return IndelRecord(
                positions=self.positions[off_val : off_val + len_val],
                kinds=self.kinds[off_val : off_val + len_val],
            )

        indices, new_offsets, new_lengths, (new_pos, new_kinds) = self._slice_ragged(item, self.positions, self.kinds)
        if len(indices) == 0:
            return self.empty()
        return Indels(positions=new_pos, kinds=new_kinds, offsets=new_offsets, lengths=new_lengths)

    @classmethod
    def empty(cls, length: int = 0) -> Indels:
        r"""Create an empty Indels container."""
        if length == 0:
            return cls(
                positions=np.empty(0, dtype=np.int32),
                kinds=np.empty(0, dtype=np.int8),
                offsets=np.empty(0, dtype=np.int32),
                lengths=np.empty(0, dtype=np.int32),
            )
        return cls(
            positions=np.empty(0, dtype=np.int32),
            kinds=np.empty(0, dtype=np.int8),
            offsets=np.zeros(length, dtype=np.int32),
            lengths=np.zeros(length, dtype=np.int32),
        )

    @classmethod
    def concat(cls, batches: Iterable[Indels]) -> Indels:
        r"""Concatenate multiple Indels containers into a single batch."""
        batches_list = list(batches)
        if not batches_list:
            return cls.empty()
        lengths = np.concatenate([b.lengths for b in batches_list])
        offsets = cls.lengths_to_offsets(lengths)
        total_items = sum(len(b.positions) for b in batches_list)
        if total_items == 0:
            return cls(
                positions=np.empty(0, dtype=np.int32),
                kinds=np.empty(0, dtype=np.int8),
                offsets=offsets,
                lengths=lengths,
            )
        positions = np.concatenate([b.positions for b in batches_list if len(b.positions) > 0])
        kinds = np.concatenate([b.kinds for b in batches_list if len(b.kinds) > 0])
        return cls(positions=positions, kinds=kinds, offsets=offsets, lengths=lengths)

empty `classmethod` ¶

empty(length: int = 0) -> Indels

Create an empty Indels container.

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

@classmethod
def empty(cls, length: int = 0) -> Indels:
    r"""Create an empty Indels container."""
    if length == 0:
        return cls(
            positions=np.empty(0, dtype=np.int32),
            kinds=np.empty(0, dtype=np.int8),
            offsets=np.empty(0, dtype=np.int32),
            lengths=np.empty(0, dtype=np.int32),
        )
    return cls(
        positions=np.empty(0, dtype=np.int32),
        kinds=np.empty(0, dtype=np.int8),
        offsets=np.zeros(length, dtype=np.int32),
        lengths=np.zeros(length, dtype=np.int32),
    )

concat `classmethod` ¶

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

Concatenate multiple Indels containers into a single batch.

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

@classmethod
def concat(cls, batches: Iterable[Indels]) -> Indels:
    r"""Concatenate multiple Indels containers into a single batch."""
    batches_list = list(batches)
    if not batches_list:
        return cls.empty()
    lengths = np.concatenate([b.lengths for b in batches_list])
    offsets = cls.lengths_to_offsets(lengths)
    total_items = sum(len(b.positions) for b in batches_list)
    if total_items == 0:
        return cls(
            positions=np.empty(0, dtype=np.int32),
            kinds=np.empty(0, dtype=np.int8),
            offsets=offsets,
            lengths=lengths,
        )
    positions = np.concatenate([b.positions for b in batches_list if len(b.positions) > 0])
    kinds = np.concatenate([b.kinds for b in batches_list if len(b.kinds) > 0])
    return cls(positions=positions, kinds=kinds, offsets=offsets, lengths=lengths)

Gene `dataclass` ¶

Scalar view of a single predicted gene.

Attributes:

Name	Type	Description
`interval`	`Interval`	Genomic interval coordinate (`seroverse.core.interval.Interval`).
`seq`	`Sequence \| None`	Coding nucleotide or protein sequence (`seroverse.core.seq.seq.Sequence`).
`contig_index`	`int`	0-based contig index offset corresponding to parent contig name.
`indels`	`IndelRecord`	Scalar frameshift indel record (`IndelRecord`).
`is_partial`	`bool`	True if the ORF is truncated at a contig boundary without a start or stop codon.

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

@dataclass(slots=True, frozen=True)
class Gene:
    r"""Scalar view of a single predicted gene.

    Attributes:
        interval (Interval): Genomic interval coordinate (`seroverse.core.interval.Interval`).
        seq (Sequence | None): Coding nucleotide or protein sequence (`seroverse.core.seq.seq.Sequence`).
        contig_index (int): 0-based contig index offset corresponding to parent contig name.
        indels (IndelRecord): Scalar frameshift indel record ([`IndelRecord`][seroref.genes.models.IndelRecord]).
        is_partial (bool): True if the ORF is truncated at a contig boundary without a start or stop codon.
    """

    interval: Interval
    seq: Sequence | None
    contig_index: int
    indels: IndelRecord
    is_partial: bool = False

Genes `dataclass` ¶

Bases: BatchedContainer[Gene, 'Genes']

Structure-of-Arrays (SoA) container for predicted genes.

Stores interval coordinates, in-frame coding nucleotide sequences, and contig index mappings for a collection of predicted Open Reading Frames (ORFs). Inherits vectorised slicing and concatenation interface from seroverse.core.collections.BatchedContainer.

Attributes:

Name	Type	Description
`intervals`	`Intervals`	Genomic interval coordinates for predicted ORFs (`seroverse.core.interval.Intervals`).
`seqs`	`Sequences`	In-frame coding nucleotide sequences oriented 5' to 3' (`seroverse.core.seq.seq.Sequences`).
`contig_indices`	`ndarray[tuple[int], dtype[uint32]]`	1D array of 0-based contig index offsets corresponding to parent contig names.
`indels`	`Indels`	Ragged frameshift indel records.
`is_partial`	`ndarray[tuple[int], dtype[bool_]]`	Boolean mask indicating contig-edge partial ORFs.

Methods:

Name	Description
`translate`	Translate coding nucleotide sequences into protein sequences.
`empty`	Create an empty `Genes` container.
`concat`	Concatenate multiple `Genes` batches into a single container.

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

@dataclass(slots=True, frozen=True, init=False)
class Genes(BatchedContainer[Gene, "Genes"]):
    r"""Structure-of-Arrays (SoA) container for predicted genes.

    Stores interval coordinates, in-frame coding nucleotide sequences, and contig index
    mappings for a collection of predicted Open Reading Frames (ORFs). Inherits
    vectorised slicing and concatenation interface from
    `seroverse.core.collections.BatchedContainer`.

    Attributes:
        intervals (Intervals): Genomic interval coordinates for predicted ORFs
            (`seroverse.core.interval.Intervals`).
        seqs (Sequences): In-frame coding nucleotide sequences oriented 5' to 3'
            (`seroverse.core.seq.seq.Sequences`).
        contig_indices (np.ndarray[tuple[int], np.dtype[np.uint32]]): 1D array of 0-based contig index
            offsets corresponding to parent contig names.
        indels (Indels): Ragged frameshift indel records.
        is_partial (np.ndarray[tuple[int], np.dtype[np.bool_]]): Boolean mask indicating contig-edge partial ORFs.
    """

    intervals: Intervals
    seqs: Sequences
    contig_indices: np.ndarray[tuple[int], np.dtype[np.uint32]]
    indels: Indels
    is_partial: np.ndarray[tuple[int], np.dtype[np.bool_]]
    genetic_code: GeneticCode
    _translations: Sequences | None

    def __init__(
        self,
        intervals: Intervals,
        seqs: Sequences | None = None,
        contig_indices: np.ndarray[tuple[int], np.dtype[np.uint32]] | None = None,
        indels: Indels | None = None,
        translations: Sequences | None = None,
        _translations: Sequences | None = None,
        is_partial: np.ndarray[tuple[int], np.dtype[np.bool_]] | None = None,
        genetic_code: GeneticCode | int | bytes | str = GeneticCode.bacterial,
    ) -> None:
        r"""Initialize a [`Genes`][seroref.genes.models.Genes] container."""
        n = len(intervals)
        object.__setattr__(self, "intervals", intervals)
        if seqs is None:
            if translations is not None:
                object.__setattr__(self, "seqs", Sequences.empty())
                object.__setattr__(self, "_translations", translations)
            elif _translations is not None:
                object.__setattr__(self, "seqs", Sequences.empty())
                object.__setattr__(self, "_translations", _translations)
            else:
                object.__setattr__(self, "seqs", Sequences.empty())
                object.__setattr__(self, "_translations", None)
        else:
            object.__setattr__(self, "seqs", seqs)
            object.__setattr__(self, "_translations", translations if translations is not None else _translations)

        object.__setattr__(
            self,
            "contig_indices",
            contig_indices if contig_indices is not None else np.zeros(n, dtype=np.uint32),
        )
        object.__setattr__(
            self,
            "indels",
            indels if indels is not None else Indels.empty(n),
        )
        object.__setattr__(
            self,
            "is_partial",
            np.asarray(is_partial, dtype=np.bool_) if is_partial is not None else np.zeros(n, dtype=np.bool_),
        )
        gc_key = genetic_code.encode("utf-8") if isinstance(genetic_code, str) else genetic_code
        object.__setattr__(
            self,
            "genetic_code",
            gc_key if isinstance(gc_key, GeneticCode) else GeneticCode.from_table(gc_key),
        )

    @property
    def has_frameshift(self) -> np.ndarray[tuple[int], np.dtype[np.bool_]]:
        r"""Boolean mask indicating which genes contain at least one frameshift indel."""
        return self.indels.lengths > 0

    @property
    def frameshift_count(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        r"""Array of frameshift indel counts per gene."""
        return self.indels.lengths

    @property
    def translations(self) -> Sequences:
        r"""Translated protein sequences.

        Returns pre-computed translations if provided, or computes them lazily using
        `self.genetic_code` (`seroverse.core.seq.alphabet.GeneticCode`).

        Returns:
            Sequences: Protein translation sequences.
        """
        if self._translations is not None:
            return self._translations
        if len(self.seqs) == 0:
            return Sequences.empty()
        translated = self.seqs.translate(code=self.genetic_code, to_stop=True)
        object.__setattr__(self, "_translations", translated)
        return translated

    def translate(self, code: GeneticCode | int | bytes | str | None = None, to_stop: bool = True) -> Sequences:
        r"""Translate coding nucleotide sequences into protein sequences.

        Args:
            code (GeneticCode | int | bytes | str | None): Genetic code translation table to use.
                If None, uses `self.genetic_code`.
            to_stop (bool): If True, truncates translation at the first stop codon (*). Defaults to True.

        Returns:
            Sequences: Translated protein sequences.
        """
        if len(self.seqs) == 0 and self._translations is not None:
            return self._translations
        c_key = code.encode("utf-8") if isinstance(code, str) else code
        resolved = (
            self.genetic_code
            if c_key is None
            else (c_key if isinstance(c_key, GeneticCode) else GeneticCode.from_table(c_key))
        )
        return self.seqs.translate(code=resolved, to_stop=to_stop)

    def __len__(self) -> int:
        r"""Return total number of predicted genes in container.

        Returns:
            int: Number of genes.
        """
        return len(self.intervals)

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

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

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

    def __getitem__(
        self, item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any]
    ) -> Gene | Genes:
        r"""Retrieve gene entry by index or slice container by range or boolean mask."""
        if isinstance(item, (int, np.integer)):
            idx = int(item)
            seq_val = (
                self.seqs[idx]
                if len(self.seqs) > 0
                else (self._translations[idx] if self._translations is not None else None)
            )
            return Gene(
                interval=self.intervals[idx],
                seq=seq_val,
                contig_index=int(self.contig_indices[idx]) if len(self.contig_indices) > 0 else 0,
                indels=self.indels[idx],
                is_partial=bool(self.is_partial[idx]) if len(self.is_partial) > 0 else False,
            )
        return Genes(
            intervals=self.intervals[item],
            seqs=self.seqs[item] if len(self.seqs) > 0 else None,
            contig_indices=self.contig_indices[item] if len(self.contig_indices) > 0 else None,
            indels=self.indels[item],
            _translations=self._translations[item] if self._translations is not None else None,
            is_partial=self.is_partial[item] if len(self.is_partial) > 0 else None,
            genetic_code=self.genetic_code,
        )

    @classmethod
    def empty(cls) -> Self:
        r"""Create an empty [`Genes`][seroref.genes.models.Genes] container."""
        return cls(
            intervals=Intervals.empty(),
            seqs=Sequences.empty(),
            contig_indices=np.empty(0, dtype=np.uint32),
            indels=Indels.empty(),
            is_partial=np.empty(0, dtype=np.bool_),
        )

    @classmethod
    def concat(cls, batches: Iterable[Self]) -> Self:
        r"""Concatenate multiple [`Genes`][seroref.genes.models.Genes] batches into a single container."""
        batches_list = [b for b in batches if len(b) > 0]
        if not batches_list:
            return cls.empty()

        has_seqs = any(len(b.seqs) > 0 for b in batches_list)
        has_trans = any(b._translations is not None for b in batches_list)

        return cls(
            intervals=Intervals.concat([b.intervals for b in batches_list]),
            seqs=Sequences.concat([b.seqs for b in batches_list]) if has_seqs else Sequences.empty(),
            contig_indices=np.concatenate([b.contig_indices for b in batches_list]),
            indels=Indels.concat([b.indels for b in batches_list]),
            _translations=(
                Sequences.concat([b.translations for b in batches_list]) if has_trans and not has_seqs else None
            ),
            is_partial=np.concatenate([b.is_partial for b in batches_list]),
            genetic_code=batches_list[0].genetic_code,
        )

has_frameshift `property` ¶

has_frameshift: ndarray[tuple[int], dtype[bool_]]

Boolean mask indicating which genes contain at least one frameshift indel.

frameshift_count `property` ¶

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

Array of frameshift indel counts per gene.

translations `property` ¶

translations: Sequences

Translated protein sequences.

Returns pre-computed translations if provided, or computes them lazily using self.genetic_code (seroverse.core.seq.alphabet.GeneticCode).

Returns:

Name	Type	Description
`Sequences`	`Sequences`	Protein translation sequences.

translate ¶

translate(
    code: GeneticCode | int | bytes | str | None = None,
    to_stop: bool = True,
) -> Sequences

Translate coding nucleotide sequences into protein sequences.

Parameters:

Name	Type	Description	Default
`code`	`GeneticCode \| int \| bytes \| str \| None`	Genetic code translation table to use. If None, uses `self.genetic_code`.	`None`
`to_stop`	`bool`	If True, truncates translation at the first stop codon (*). Defaults to True.	`True`

Returns:

Name	Type	Description
`Sequences`	`Sequences`	Translated protein sequences.

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

def translate(self, code: GeneticCode | int | bytes | str | None = None, to_stop: bool = True) -> Sequences:
    r"""Translate coding nucleotide sequences into protein sequences.

    Args:
        code (GeneticCode | int | bytes | str | None): Genetic code translation table to use.
            If None, uses `self.genetic_code`.
        to_stop (bool): If True, truncates translation at the first stop codon (*). Defaults to True.

    Returns:
        Sequences: Translated protein sequences.
    """
    if len(self.seqs) == 0 and self._translations is not None:
        return self._translations
    c_key = code.encode("utf-8") if isinstance(code, str) else code
    resolved = (
        self.genetic_code
        if c_key is None
        else (c_key if isinstance(c_key, GeneticCode) else GeneticCode.from_table(c_key))
    )
    return self.seqs.translate(code=resolved, to_stop=to_stop)

empty `classmethod` ¶

empty() -> Self

Create an empty Genes container.

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

@classmethod
def empty(cls) -> Self:
    r"""Create an empty [`Genes`][seroref.genes.models.Genes] container."""
    return cls(
        intervals=Intervals.empty(),
        seqs=Sequences.empty(),
        contig_indices=np.empty(0, dtype=np.uint32),
        indels=Indels.empty(),
        is_partial=np.empty(0, dtype=np.bool_),
    )

concat `classmethod` ¶

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

Concatenate multiple Genes batches into a single container.

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

@classmethod
def concat(cls, batches: Iterable[Self]) -> Self:
    r"""Concatenate multiple [`Genes`][seroref.genes.models.Genes] batches into a single container."""
    batches_list = [b for b in batches if len(b) > 0]
    if not batches_list:
        return cls.empty()

    has_seqs = any(len(b.seqs) > 0 for b in batches_list)
    has_trans = any(b._translations is not None for b in batches_list)

    return cls(
        intervals=Intervals.concat([b.intervals for b in batches_list]),
        seqs=Sequences.concat([b.seqs for b in batches_list]) if has_seqs else Sequences.empty(),
        contig_indices=np.concatenate([b.contig_indices for b in batches_list]),
        indels=Indels.concat([b.indels for b in batches_list]),
        _translations=(
            Sequences.concat([b.translations for b in batches_list]) if has_trans and not has_seqs else None
        ),
        is_partial=np.concatenate([b.is_partial for b in batches_list]),
        genetic_code=batches_list[0].genetic_code,
    )

HMM `dataclass` ¶

Container holding baked FragGeneScan HMM parameters.

Attributes:

Name	Type	Description
`tr`	`ndarray`	Global transition penalties (14,).
`tr_mi`	`ndarray`	Match-to-insertion transitions (4, 4).
`tr_ii`	`ndarray`	Insertion-to-insertion transitions (4, 4).
`pi`	`ndarray`	Initial state log-probabilities (29,).
`e_m`	`ndarray`	Forward match state emission log-probabilities (44, 6, 16, 4).
`e_m1`	`ndarray`	Reverse match state emission log-probabilities (44, 6, 16, 4).
`tr_rr`	`ndarray`	Non-coding state transitions (44, 4, 4).
`tr_s`	`ndarray`	PWM window transitions around forward start (44, 61, 64).
`tr_e`	`ndarray`	PWM window transitions around forward stop (44, 61, 64).
`tr_s1`	`ndarray`	PWM window transitions around reverse start (44, 61, 64).
`tr_e1`	`ndarray`	PWM window transitions around reverse stop (44, 61, 64).
`dist_s`	`ndarray`	Gaussian parameters for forward start (44, 6).
`dist_e`	`ndarray`	Gaussian parameters for forward stop (44, 6).
`dist_s1`	`ndarray`	Gaussian parameters for reverse start (44, 6).
`dist_e1`	`ndarray`	Gaussian parameters for reverse stop (44, 6).

Methods:

Name	Description
`load`	Load HMM parameters from compressed NumPy archive (.npz).

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

@dataclass(frozen=True, slots=True)
class HMM:
    r"""Container holding baked FragGeneScan HMM parameters.

    Attributes:
        tr (np.ndarray): Global transition penalties (14,).
        tr_mi (np.ndarray): Match-to-insertion transitions (4, 4).
        tr_ii (np.ndarray): Insertion-to-insertion transitions (4, 4).
        pi (np.ndarray): Initial state log-probabilities (29,).
        e_m (np.ndarray): Forward match state emission log-probabilities (44, 6, 16, 4).
        e_m1 (np.ndarray): Reverse match state emission log-probabilities (44, 6, 16, 4).
        tr_rr (np.ndarray): Non-coding state transitions (44, 4, 4).
        tr_s (np.ndarray): PWM window transitions around forward start (44, 61, 64).
        tr_e (np.ndarray): PWM window transitions around forward stop (44, 61, 64).
        tr_s1 (np.ndarray): PWM window transitions around reverse start (44, 61, 64).
        tr_e1 (np.ndarray): PWM window transitions around reverse stop (44, 61, 64).
        dist_s (np.ndarray): Gaussian parameters for forward start (44, 6).
        dist_e (np.ndarray): Gaussian parameters for forward stop (44, 6).
        dist_s1 (np.ndarray): Gaussian parameters for reverse start (44, 6).
        dist_e1 (np.ndarray): Gaussian parameters for reverse stop (44, 6).
    """

    tr: np.ndarray[tuple[int], np.dtype[np.float64]]
    tr_mi: np.ndarray[tuple[int, int], np.dtype[np.float64]]
    tr_ii: np.ndarray[tuple[int, int], np.dtype[np.float64]]
    pi: np.ndarray[tuple[int], np.dtype[np.float64]]
    tr_rr: np.ndarray[tuple[int, int, int], np.dtype[np.float64]]
    e_match: np.ndarray[tuple[int, int, int, int, int], np.dtype[np.float64]]
    tr_start: np.ndarray[tuple[int, int, int, int], np.dtype[np.float64]]
    tr_stop: np.ndarray[tuple[int, int, int, int], np.dtype[np.float64]]
    dist_start: np.ndarray[tuple[int, int, int], np.dtype[np.float64]]
    dist_stop: np.ndarray[tuple[int, int, int], np.dtype[np.float64]]

    @classmethod
    def load(cls, path: str | Path | None = None) -> HMM:
        r"""Load HMM parameters from compressed NumPy archive (.npz).

        Args:
            path (str | Path | None): Optional path to .npz archive. If None, loads the
                bundled default model located at `serogen/bgc/data/fgs_model.npz`.

        Returns:
            HMM: Instantiated HMM parameter container.
        """
        if path is None:
            path = Path(__file__).resolve().parent / "data" / "fgs_model.npz"
        data = np.load(path)
        return cls(
            tr=data["tr"],
            tr_mi=data["tr_mi"],
            tr_ii=data["tr_ii"],
            pi=data["pi"],
            tr_rr=data["tr_rr"],
            e_match=np.stack([data["e_m"], data["e_m1"]]),
            tr_start=np.stack([data["tr_s"], data["tr_s1"]]),
            tr_stop=np.stack([data["tr_e"], data["tr_e1"]]),
            dist_start=np.stack([data["dist_s"], data["dist_s1"]]),
            dist_stop=np.stack([data["dist_e"], data["dist_e1"]]),
        )

load `classmethod` ¶

load(path: str | Path | None = None) -> HMM

Load HMM parameters from compressed NumPy archive (.npz).

Parameters:

Name	Type	Description	Default
`path`	`str \| Path \| None`	Optional path to .npz archive. If None, loads the bundled default model located at `serogen/bgc/data/fgs_model.npz`.	`None`

Returns:

Name	Type	Description
`HMM`	`HMM`	Instantiated HMM parameter container.

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

@classmethod
def load(cls, path: str | Path | None = None) -> HMM:
    r"""Load HMM parameters from compressed NumPy archive (.npz).

    Args:
        path (str | Path | None): Optional path to .npz archive. If None, loads the
            bundled default model located at `serogen/bgc/data/fgs_model.npz`.

    Returns:
        HMM: Instantiated HMM parameter container.
    """
    if path is None:
        path = Path(__file__).resolve().parent / "data" / "fgs_model.npz"
    data = np.load(path)
    return cls(
        tr=data["tr"],
        tr_mi=data["tr_mi"],
        tr_ii=data["tr_ii"],
        pi=data["pi"],
        tr_rr=data["tr_rr"],
        e_match=np.stack([data["e_m"], data["e_m1"]]),
        tr_start=np.stack([data["tr_s"], data["tr_s1"]]),
        tr_stop=np.stack([data["tr_e"], data["tr_e1"]]),
        dist_start=np.stack([data["dist_s"], data["dist_s1"]]),
        dist_stop=np.stack([data["dist_e"], data["dist_e1"]]),
    )

get_hmm_model `cached` ¶

get_hmm_model() -> HMM

Return cached singleton instance of the default bundled HMM model.

Returns:

Name	Type	Description
`HMM`	`HMM`	Default baked model.

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

@lru_cache(maxsize=1)
def get_hmm_model() -> HMM:
    r"""Return cached singleton instance of the default bundled HMM model.

    Returns:
        HMM: Default baked model.
    """
    return HMM.load()

Functional Gene Role Ontology (`seroref.genes.roles`)¶

roles ¶

Canonical biosynthetic gene roles and Structure-of-Arrays (SoA) metadata specifications.

Classes:

Name	Description
`RoleCategory`	Functional category of a biosynthetic gene role.
`RoleSpec`	Immutable scalar view of a canonical `GeneRole` metadata specification.
`RoleSpecs`	Pure CSR-packed Ragged Structure-of-Arrays (SoA) batch container over `RoleSpec` definitions.
`GeneRole`	Canonical biosynthetic gene roles (0..29) for surface polysaccharide loci.

RoleCategory ¶

Bases: IntEnum

Functional category of a biosynthetic gene role.

Methods:

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

Attributes:

Name	Type	Description
`label`	`bytes`	Canonical ASCII byte-string label for this category.

Source code in src/seroref/genes/roles.py

class RoleCategory(IntEnum):
    """Functional category of a biosynthetic gene role."""

    TRANSPORT = 0
    POLYMER_ASSEMBLY = 1
    MODIFIER = 2
    PRECURSOR_SYNTHESIS = 3
    MOBILE_ELEMENT = 4
    BACKGROUND = 5

    @property
    def label(self) -> bytes:
        """Canonical ASCII byte-string label for this category."""
        return _ROLE_CATEGORY_LABELS[int(self)]

    @classmethod
    def from_name(cls, key: str | bytes | int | RoleCategory) -> RoleCategory:
        """Resolve a `RoleCategory` by integer 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)
        b_clean = b.strip().lower().replace(b"-", b"").replace(b"_", b"").replace(b" ", b"")
        for cat in cls:
            if b_clean in (
                cat.name.encode("ascii").lower().replace(b"_", b""),
                cat.label.lower().replace(b"_", b""),
            ):
                return cat
        raise KeyError(f"Unknown RoleCategory identifier: {key!r}")

    @classmethod
    def resolve(cls, key: str | bytes | int | RoleCategory, default: RoleCategory | None = None) -> RoleCategory:
        """Resolve a `RoleCategory` identifier, falling back to `default` (or `RoleCategory.BACKGROUND`)."""
        fallback = cls.BACKGROUND 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 category.

from_name `classmethod` ¶

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

Resolve a RoleCategory by integer ID, enum name, or label.

Source code in src/seroref/genes/roles.py

@classmethod
def from_name(cls, key: str | bytes | int | RoleCategory) -> RoleCategory:
    """Resolve a `RoleCategory` by integer 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)
    b_clean = b.strip().lower().replace(b"-", b"").replace(b"_", b"").replace(b" ", b"")
    for cat in cls:
        if b_clean in (
            cat.name.encode("ascii").lower().replace(b"_", b""),
            cat.label.lower().replace(b"_", b""),
        ):
            return cat
    raise KeyError(f"Unknown RoleCategory identifier: {key!r}")

resolve `classmethod` ¶

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

Resolve a RoleCategory identifier, falling back to default (or RoleCategory.BACKGROUND).

Source code in src/seroref/genes/roles.py

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

RoleSpec `dataclass` ¶

Bases: ScalarView

Immutable scalar view of a canonical GeneRole metadata specification.

Attributes:

Name	Type	Description
`is_mobile_element`	`bool`	True if this role is a mobile genetic element (`TRANSPOSASE`, `INTEGRASE`).
`is_biosynthetic`	`bool`	True if this role represents active locus biosynthesis, modification, or transport.

Source code in src/seroref/genes/roles.py

@dataclass(frozen=True, slots=True)
class RoleSpec(ScalarView):
    """Immutable scalar view of a canonical `GeneRole` metadata specification."""

    name: bytes
    abbreviation: bytes
    category: RoleCategory
    description: bytes
    aliases: tuple[bytes, ...] = ()
    is_machinery: bool = False
    null_prior: float = 0.005

    @property
    def is_mobile_element(self) -> bool:
        """True if this role is a mobile genetic element (`TRANSPOSASE`, `INTEGRASE`)."""
        return self.category == RoleCategory.MOBILE_ELEMENT

    @property
    def is_biosynthetic(self) -> bool:
        """True if this role represents active locus biosynthesis, modification, or transport."""
        return self.category not in (RoleCategory.MOBILE_ELEMENT, RoleCategory.BACKGROUND)

is_mobile_element `property` ¶

is_mobile_element: bool

True if this role is a mobile genetic element (TRANSPOSASE, INTEGRASE).

is_biosynthetic `property` ¶

is_biosynthetic: bool

True if this role represents active locus biosynthesis, modification, or transport.

RoleSpecs `dataclass` ¶

Bases: RaggedArrayContainer[RoleSpec, 'RoleSpecs']

Pure CSR-packed Ragged Structure-of-Arrays (SoA) batch container over RoleSpec definitions.

Methods:

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

Attributes:

Name	Type	Description
`biosynthetic_mask`	`ndarray[tuple[int], dtype[bool_]]`	1D boolean mask (`(N,)`) marking active biosynthetic/transport/modifier gene roles.
`machinery_roles`	`ndarray[tuple[int], dtype[int32]]`	1D `int32` array of role indices marked as core export/polymerization machinery.

Source code in src/seroref/genes/roles.py

@dataclass(frozen=True, slots=True)
class RoleSpecs(RaggedArrayContainer[RoleSpec, "RoleSpecs"]):
    """Pure CSR-packed Ragged Structure-of-Arrays (SoA) batch container over `RoleSpec` definitions."""

    names: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    abbreviations: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    categories: np.ndarray[tuple[int], np.dtype[np.int8]]
    descriptions: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    is_mobile_element: np.ndarray[tuple[int], np.dtype[np.bool_]]
    is_biosynthetic: np.ndarray[tuple[int], np.dtype[np.bool_]]
    is_machinery: np.ndarray[tuple[int], np.dtype[np.bool_]]
    null_priors: np.ndarray[tuple[int], np.dtype[np.float64]]
    feature_matrix: np.ndarray[tuple[int, int], np.dtype[np.float32]]
    alias_values: np.ndarray[tuple[int], np.dtype[np.bytes_]]
    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[RoleSpec]:
        for i in range(len(self)):
            yield self[i]

    @property
    def biosynthetic_mask(self) -> np.ndarray[tuple[int], np.dtype[np.bool_]]:
        """1D boolean mask (`(N,)`) marking active biosynthetic/transport/modifier gene roles."""
        return self.is_biosynthetic

    @property
    def machinery_roles(self) -> np.ndarray[tuple[int], np.dtype[np.int32]]:
        """1D `int32` array of role indices marked as core export/polymerization machinery."""
        return np.flatnonzero(self.is_machinery).astype(np.int32)

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

    @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],
    ) -> RoleSpec | Self: ...

    def __getitem__(
        self,
        item: int | np.integer[Any] | slice | np.ndarray[tuple[int], Any] | list[Any],
    ) -> RoleSpec | 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"RoleSpecs index {item} out of range for length {n}.")
            off, ln = int(self.offsets[idx]), int(self.lengths[idx])
            aliases = tuple(bytes(self.alias_values[k]) for k in range(off, off + ln))
            return RoleSpec(
                name=bytes(self.names[idx]),
                abbreviation=bytes(self.abbreviations[idx]),
                category=RoleCategory(int(self.categories[idx])),
                description=bytes(self.descriptions[idx]),
                aliases=aliases,
                is_machinery=bool(self.is_machinery[idx]),
                null_prior=float(self.null_priors[idx]),
            )

        idx_pool = np.arange(len(self.alias_values), dtype=np.int32)
        indices, new_offsets, new_lengths, (gathered_idx,) = self._slice_ragged(item, idx_pool)
        if len(indices) == 0:
            return self.empty()
        alias_v = self.alias_values[gathered_idx] if len(gathered_idx) > 0 else np.empty(0, dtype="S1")
        return type(self)(
            names=self.names[indices],
            abbreviations=self.abbreviations[indices],
            categories=self.categories[indices],
            descriptions=self.descriptions[indices],
            is_mobile_element=self.is_mobile_element[indices],
            is_biosynthetic=self.is_biosynthetic[indices],
            is_machinery=self.is_machinery[indices],
            null_priors=self.null_priors[indices],
            feature_matrix=self.feature_matrix[indices],
            alias_values=alias_v,
            offsets=new_offsets,
            lengths=new_lengths,
        )

    @classmethod
    def empty(cls) -> Self:
        """Create an empty 0-length `RoleSpecs` container."""
        n_feat = len(RoleCategory) + 1
        return cls(
            names=np.empty(0, dtype="S1"),
            abbreviations=np.empty(0, dtype="S1"),
            categories=np.empty(0, dtype=np.int8),
            descriptions=np.empty(0, dtype="S1"),
            is_mobile_element=np.empty(0, dtype=np.bool_),
            is_biosynthetic=np.empty(0, dtype=np.bool_),
            is_machinery=np.empty(0, dtype=np.bool_),
            null_priors=np.empty(0, dtype=np.float64),
            feature_matrix=np.empty((0, n_feat), dtype=np.float32),
            alias_values=np.empty(0, dtype="S1"),
            offsets=np.empty(0, dtype=np.int32),
            lengths=np.empty(0, dtype=np.int32),
        )

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

        names = np.array([s.name for s in specs], dtype="S24")
        abbreviations = np.array([s.abbreviation for s in specs], dtype="S8")
        categories = np.array([int(s.category) for s in specs], dtype=np.int8)
        descriptions = np.array([s.description for s in specs], dtype="S")
        is_mobile = np.array([s.is_mobile_element for s in specs], dtype=np.bool_)
        is_biosyn = np.array([s.is_biosynthetic for s in specs], dtype=np.bool_)
        is_mach = np.array([s.is_machinery for s in specs], dtype=np.bool_)

        null_priors = np.array([s.null_prior for s in specs], dtype=np.float64)
        prior_sum = float(null_priors.sum())
        if prior_sum > 0.0:
            null_priors /= prior_sum

        n_cats = len(RoleCategory)
        feature_matrix = np.zeros((n, n_cats + 1), dtype=np.float32)
        feature_matrix[np.arange(n), np.clip(categories, 0, n_cats - 1)] = 1.0
        for i, s in enumerate(specs):
            if s.name in (b"POLYMERASE", b"TRANSMEMBRANE_SYNTHASE", b"GLYCOSYLTRANSFERASE"):
                feature_matrix[i, -1] = 1.0

        alias_flat: list[bytes] = []
        alias_lens = np.empty(n, dtype=np.int32)
        for i, s in enumerate(specs):
            alias_lens[i] = len(s.aliases)
            alias_flat.extend(s.aliases)
        alias_values = np.array(alias_flat, dtype="S") if alias_flat else np.empty(0, dtype="S1")
        offsets = lengths_to_offsets(alias_lens)

        for arr in (
            names,
            abbreviations,
            categories,
            descriptions,
            is_mobile,
            is_biosyn,
            is_mach,
            null_priors,
            feature_matrix,
            alias_values,
            offsets,
            alias_lens,
        ):
            arr.setflags(write=False)

        return cls(
            names=names,
            abbreviations=abbreviations,
            categories=categories,
            descriptions=descriptions,
            is_mobile_element=is_mobile,
            is_biosynthetic=is_biosyn,
            is_machinery=is_mach,
            null_priors=null_priors,
            feature_matrix=feature_matrix,
            alias_values=alias_values,
            offsets=offsets,
            lengths=alias_lens,
        )

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

biosynthetic_mask `property` ¶

biosynthetic_mask: ndarray[tuple[int], dtype[bool_]]

1D boolean mask ((N,)) marking active biosynthetic/transport/modifier gene roles.

machinery_roles `property` ¶

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

1D int32 array of role indices marked as core export/polymerization machinery.

empty `classmethod` ¶

empty() -> Self

Create an empty 0-length RoleSpecs container.

Source code in src/seroref/genes/roles.py

@classmethod
def empty(cls) -> Self:
    """Create an empty 0-length `RoleSpecs` container."""
    n_feat = len(RoleCategory) + 1
    return cls(
        names=np.empty(0, dtype="S1"),
        abbreviations=np.empty(0, dtype="S1"),
        categories=np.empty(0, dtype=np.int8),
        descriptions=np.empty(0, dtype="S1"),
        is_mobile_element=np.empty(0, dtype=np.bool_),
        is_biosynthetic=np.empty(0, dtype=np.bool_),
        is_machinery=np.empty(0, dtype=np.bool_),
        null_priors=np.empty(0, dtype=np.float64),
        feature_matrix=np.empty((0, n_feat), dtype=np.float32),
        alias_values=np.empty(0, dtype="S1"),
        offsets=np.empty(0, dtype=np.int32),
        lengths=np.empty(0, dtype=np.int32),
    )

from_list `classmethod` ¶

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

Compile a sequence of RoleSpec scalar views into a frozen RoleSpecs SoA container.

Source code in src/seroref/genes/roles.py

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

    names = np.array([s.name for s in specs], dtype="S24")
    abbreviations = np.array([s.abbreviation for s in specs], dtype="S8")
    categories = np.array([int(s.category) for s in specs], dtype=np.int8)
    descriptions = np.array([s.description for s in specs], dtype="S")
    is_mobile = np.array([s.is_mobile_element for s in specs], dtype=np.bool_)
    is_biosyn = np.array([s.is_biosynthetic for s in specs], dtype=np.bool_)
    is_mach = np.array([s.is_machinery for s in specs], dtype=np.bool_)

    null_priors = np.array([s.null_prior for s in specs], dtype=np.float64)
    prior_sum = float(null_priors.sum())
    if prior_sum > 0.0:
        null_priors /= prior_sum

    n_cats = len(RoleCategory)
    feature_matrix = np.zeros((n, n_cats + 1), dtype=np.float32)
    feature_matrix[np.arange(n), np.clip(categories, 0, n_cats - 1)] = 1.0
    for i, s in enumerate(specs):
        if s.name in (b"POLYMERASE", b"TRANSMEMBRANE_SYNTHASE", b"GLYCOSYLTRANSFERASE"):
            feature_matrix[i, -1] = 1.0

    alias_flat: list[bytes] = []
    alias_lens = np.empty(n, dtype=np.int32)
    for i, s in enumerate(specs):
        alias_lens[i] = len(s.aliases)
        alias_flat.extend(s.aliases)
    alias_values = np.array(alias_flat, dtype="S") if alias_flat else np.empty(0, dtype="S1")
    offsets = lengths_to_offsets(alias_lens)

    for arr in (
        names,
        abbreviations,
        categories,
        descriptions,
        is_mobile,
        is_biosyn,
        is_mach,
        null_priors,
        feature_matrix,
        alias_values,
        offsets,
        alias_lens,
    ):
        arr.setflags(write=False)

    return cls(
        names=names,
        abbreviations=abbreviations,
        categories=categories,
        descriptions=descriptions,
        is_mobile_element=is_mobile,
        is_biosynthetic=is_biosyn,
        is_machinery=is_mach,
        null_priors=null_priors,
        feature_matrix=feature_matrix,
        alias_values=alias_values,
        offsets=offsets,
        lengths=alias_lens,
    )

concat `classmethod` ¶

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

Concatenate multiple RoleSpecs containers.

Source code in src/seroref/genes/roles.py

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

GeneRole ¶

Bases: IntEnum

Canonical biosynthetic gene roles (0..29) for surface polysaccharide loci.

Subclasses IntEnum for zero-overhead integer semantics in bitfields (IntNode), 32-bit presence masks (1 << role), Numba kernels, and CSR matrices.

Methods:

Name	Description
`specs`	Return the cached singleton `RoleSpecs` SoA container over all `GeneRole` members.
`from_name`	Resolve a role from an integer index, canonical name, abbreviation, or alias.
`resolve`	Resolve a role identifier, falling back to `default` (or `GeneRole.UNKNOWN`) if not found.
`names_array`	Contiguous 1D NumPy byte-string array of canonical role names.
`abbreviations_array`	Contiguous 1D NumPy byte-string array of role abbreviations.
`categories_array`	Contiguous 1D `int8` NumPy array of `RoleCategory` IDs.
`feature_matrix`	Return precomputed `(30, D)` float32 one-hot category feature matrix for embeddings.

Attributes:

Name	Type	Description
`spec`	`RoleSpec`	Return the immutable `RoleSpec` scalar view for this role.
`abbreviation`	`bytes`	Short display abbreviation (e.g. `b"Wzx"`, `b"GT"`).
`category`	`RoleCategory`	Functional `RoleCategory` of this gene role.
`description`	`bytes`	Human-readable biochemical description.
`aliases`	`tuple[bytes, ...]`	Known gene symbol aliases for this role.
`bitmask`	`int`	32-bit integer bitmask (`1 << int(self)`) for archetype presence masks.
`is_mobile_element`	`bool`	True if this role is a mobile genetic element (`TRANSPOSASE`, `INTEGRASE`).
`is_biosynthetic`	`bool`	True if this role represents active locus biosynthesis, modification, or transport.
`is_machinery`	`bool`	True if this role is core export, translocation, or polymerization machinery.

Source code in src/seroref/genes/roles.py

class GeneRole(IntEnum):
    """Canonical biosynthetic gene roles (0..29) for surface polysaccharide loci.

    Subclasses `IntEnum` for zero-overhead integer semantics in bitfields (`IntNode`),
    32-bit presence masks (`1 << role`), Numba kernels, and `CSR` matrices.
    """

    FLIPPASE = 0
    POLYMERASE = 1
    COPOLYMERASE = 2
    KINASE = 3
    PHOSPHATASE = 4
    ABC_PERMEASE = 5
    ABC_ATPASE = 6
    LIGASE = 7
    TRANSLOCON = 8
    CAPSULE_ANCHOR = 9
    PYRUVYLTRANSFERASE = 10
    METHYLTRANSFERASE = 11
    ACETYLTRANSFERASE = 12
    PHOSPHOTRANSFERASE = 13
    FORMYLTRANSFERASE = 14
    ALANINE_TRANSFERASE = 15
    AMINOTRANSFERASE = 16
    ALDOLASE = 17
    DEACETYLASE = 18
    EPIMERASE = 19
    OXIDOREDUCTASE = 20
    TRANSMEMBRANE_SYNTHASE = 21
    NUCLEOTIDYLTRANSFERASE = 22
    NUCLEOTIDASE = 23
    INITIATOR = 24
    GLYCOSYLHYDROLASE = 25
    GLYCOSYLTRANSFERASE = 26
    TRANSPOSASE = 27
    INTEGRASE = 28
    UNKNOWN = 29

    @property
    def spec(self) -> RoleSpec:
        """Return the immutable `RoleSpec` scalar view for this role."""
        return self.specs()[int(self)]

    @property
    def abbreviation(self) -> bytes:
        """Short display abbreviation (e.g. `b"Wzx"`, `b"GT"`)."""
        return bytes(self.specs().abbreviations[int(self)])

    @property
    def category(self) -> RoleCategory:
        """Functional `RoleCategory` of this gene role."""
        return RoleCategory(int(self.specs().categories[int(self)]))

    @property
    def description(self) -> bytes:
        """Human-readable biochemical description."""
        return bytes(self.specs().descriptions[int(self)])

    @property
    def aliases(self) -> tuple[bytes, ...]:
        """Known gene symbol aliases for this role."""
        return self.spec.aliases

    @property
    def bitmask(self) -> int:
        """32-bit integer bitmask (`1 << int(self)`) for archetype presence masks."""
        return 1 << int(self)

    @property
    def is_mobile_element(self) -> bool:
        """True if this role is a mobile genetic element (`TRANSPOSASE`, `INTEGRASE`)."""
        return bool(self.specs().is_mobile_element[int(self)])

    @property
    def is_biosynthetic(self) -> bool:
        """True if this role represents active locus biosynthesis, modification, or transport."""
        return bool(self.specs().is_biosynthetic[int(self)])

    @property
    def is_machinery(self) -> bool:
        """True if this role is core export, translocation, or polymerization machinery."""
        return bool(self.specs().is_machinery[int(self)])

    @classmethod
    def specs(cls) -> RoleSpecs:
        """Return the cached singleton `RoleSpecs` SoA container over all `GeneRole` members."""
        return _get_role_specs()

    @classmethod
    def from_name(cls, key: str | bytes | int | GeneRole) -> GeneRole:
        """Resolve a role from an integer index, canonical name, abbreviation, or 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.strip()
        lookup = _get_role_lookup()
        if b_clean in lookup:
            return lookup[b_clean]
        b_lower = b_clean.lower()
        if b_lower in lookup:
            return lookup[b_lower]
        raise KeyError(f"Unknown GeneRole identifier: {key!r}")

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

    @classmethod
    def names_array(cls) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """Contiguous 1D NumPy byte-string array of canonical role names."""
        return cls.specs().names

    @classmethod
    def abbreviations_array(cls) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
        """Contiguous 1D NumPy byte-string array of role abbreviations."""
        return cls.specs().abbreviations

    @classmethod
    def categories_array(cls) -> np.ndarray[tuple[int], np.dtype[np.int8]]:
        """Contiguous 1D `int8` NumPy array of `RoleCategory` IDs."""
        return cls.specs().categories

    @classmethod
    def feature_matrix(cls) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
        """Return precomputed `(30, D)` float32 one-hot category feature matrix for embeddings."""
        return cls.specs().feature_matrix

spec `property` ¶

spec: RoleSpec

Return the immutable RoleSpec scalar view for this role.

abbreviation `property` ¶

abbreviation: bytes

Short display abbreviation (e.g. b"Wzx", b"GT").

category `property` ¶

category: RoleCategory

Functional RoleCategory of this gene role.

description `property` ¶

description: bytes

Human-readable biochemical description.

aliases `property` ¶

aliases: tuple[bytes, ...]

Known gene symbol aliases for this role.

bitmask `property` ¶

bitmask: int

32-bit integer bitmask (1 << int(self)) for archetype presence masks.

is_mobile_element `property` ¶

is_mobile_element: bool

True if this role is a mobile genetic element (TRANSPOSASE, INTEGRASE).

is_biosynthetic `property` ¶

is_biosynthetic: bool

True if this role represents active locus biosynthesis, modification, or transport.

is_machinery `property` ¶

is_machinery: bool

True if this role is core export, translocation, or polymerization machinery.

specs `classmethod` ¶

specs() -> RoleSpecs

Return the cached singleton RoleSpecs SoA container over all GeneRole members.

Source code in src/seroref/genes/roles.py

@classmethod
def specs(cls) -> RoleSpecs:
    """Return the cached singleton `RoleSpecs` SoA container over all `GeneRole` members."""
    return _get_role_specs()

from_name `classmethod` ¶

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

Resolve a role from an integer index, canonical name, abbreviation, or alias.

Source code in src/seroref/genes/roles.py

@classmethod
def from_name(cls, key: str | bytes | int | GeneRole) -> GeneRole:
    """Resolve a role from an integer index, canonical name, abbreviation, or 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.strip()
    lookup = _get_role_lookup()
    if b_clean in lookup:
        return lookup[b_clean]
    b_lower = b_clean.lower()
    if b_lower in lookup:
        return lookup[b_lower]
    raise KeyError(f"Unknown GeneRole identifier: {key!r}")

resolve `classmethod` ¶

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

Resolve a role identifier, falling back to default (or GeneRole.UNKNOWN) if not found.

Source code in src/seroref/genes/roles.py

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

names_array `classmethod` ¶

names_array() -> ndarray[tuple[int], dtype[bytes_]]

Contiguous 1D NumPy byte-string array of canonical role names.

Source code in src/seroref/genes/roles.py

@classmethod
def names_array(cls) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
    """Contiguous 1D NumPy byte-string array of canonical role names."""
    return cls.specs().names

abbreviations_array `classmethod` ¶

abbreviations_array() -> ndarray[tuple[int], dtype[bytes_]]

Contiguous 1D NumPy byte-string array of role abbreviations.

Source code in src/seroref/genes/roles.py

@classmethod
def abbreviations_array(cls) -> np.ndarray[tuple[int], np.dtype[np.bytes_]]:
    """Contiguous 1D NumPy byte-string array of role abbreviations."""
    return cls.specs().abbreviations

categories_array `classmethod` ¶

categories_array() -> ndarray[tuple[int], dtype[int8]]

Contiguous 1D int8 NumPy array of RoleCategory IDs.

Source code in src/seroref/genes/roles.py

@classmethod
def categories_array(cls) -> np.ndarray[tuple[int], np.dtype[np.int8]]:
    """Contiguous 1D `int8` NumPy array of `RoleCategory` IDs."""
    return cls.specs().categories

feature_matrix `classmethod` ¶

feature_matrix() -> ndarray[
    tuple[int, int], dtype[float32]
]

Return precomputed (30, D) float32 one-hot category feature matrix for embeddings.

Source code in src/seroref/genes/roles.py

@classmethod
def feature_matrix(cls) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
    """Return precomputed `(30, D)` float32 one-hot category feature matrix for embeddings."""
    return cls.specs().feature_matrix

1D CNN Role Classifier (`seroref.genes.cnn`)¶

cnn ¶

Classifying gene roles from amino acid sequences using a Multi-Scale 1D-CNN trained via compiled Numba AdamW backpropagation with dual (max + average) pooling.

Classes:

Name	Description
`GeneRoleClassifier`	Classifies `GeneRole`s from amino acid sequences using a Multi-Scale 1D-CNN with Dual Pooling.

GeneRoleClassifier ¶

Classifies GeneRoles from amino acid sequences using a Multi-Scale 1D-CNN with Dual Pooling.

Learns representations from variable-length sequence token arrays using an embedding layer, multi-scale 1D convolutions (e.g. receptive fields 5, 11, 17), ReLU activation, emergent dual pooling ([conv_max, conv_avg]), a dense role classification head, and a joint 1-bit genomic source readout head (0 = bacterial, 1 = phage/prophage). Optimization is performed via a parallel compiled Numba AdamW training kernel with exact gradient routing.

Attributes:

Name	Type	Description
`vocab_size`		Number of discrete tokens in sequence vocabulary (default 23: 22 AMINO tokens + 1 unknown/pad).
`embed_dim`		Dimensionality of token embedding vectors.
`num_filters`		Number of 1D convolutional feature channels across all kernel sizes.
`kernel_size`		Width of 1D convolution receptive field window (int), or tuple of multi-scale kernel sizes (e.g. `(5, 9, 13)`).
`learning_rate`		AdamW step size hyperparameter.
`epochs`		Number of complete training epochs over dataset.
`batch_size`		Mini-batch size for parallel AdamW gradient accumulation and updates.
`weight_decay`		Decoupled L2 weight decay coefficient applied to model parameters.
`random_state`		Optional seed for reproducible parameter initialization and sample shuffling.
`class_weight`		Optional `"balanced"` inverse-frequency class weighting or `None`.

Methods:

Name	Description
`fit`	Fit the 1D-CNN using parallel mini-batch AdamW optimization over ragged token sequences.
`transform`	Extract sequence representations from the specified network layer.
`decision_function`	Compute raw pre-softmax class logits `(n_samples, n_classes)` for each sequence in `X`.
`predict_proba`	Predict class probabilities `(n_samples, n_classes)` for each sequence in `X`.
`predict_source_proba`	Predict probability `P(phage \| x)` in `[0.0, 1.0]` for each sequence in `X`.
`predict_source`	Predict binary genomic source (`0` = bacterial, `1` = phage/prophage) for each sequence in `X`.
`predict`	Predict most likely class label (e.g. `GeneRole` integer index) for each sequence in `X`.
`score`	Return classification accuracy on `(X, y)`.
`top_k_accuracy`	Compute Top-`k` classification accuracy on `(X, y)`.
`save`	Save model weights, classes, and hyperparameters to a compressed `.npz` archive.
`load`	Load model weights, classes, and hyperparameters from a compressed `.npz` archive.
`pretrained`	Load the packaged pre-trained `GeneRoleClassifier` from `serogen.data/model.npz`.

Source code in src/seroref/genes/cnn.py

class GeneRoleClassifier:
    r"""Classifies `GeneRole`s from amino acid sequences using a Multi-Scale 1D-CNN with Dual Pooling.

    Learns representations from variable-length sequence token arrays using an embedding layer,
    multi-scale 1D convolutions (e.g. receptive fields 5, 11, 17), ReLU activation, emergent dual
    pooling (`[conv_max, conv_avg]`), a dense role classification head, and a joint 1-bit genomic
    `source` readout head (`0 = bacterial`, `1 = phage/prophage`). Optimization is performed via a
    parallel compiled Numba AdamW training kernel with exact gradient routing.

    Attributes:
        vocab_size: Number of discrete tokens in sequence vocabulary (default 23: 22 AMINO tokens + 1 unknown/pad).
        embed_dim: Dimensionality of token embedding vectors.
        num_filters: Number of 1D convolutional feature channels across all kernel sizes.
        kernel_size: Width of 1D convolution receptive field window (int), or tuple of multi-scale
            kernel sizes (e.g. `(5, 9, 13)`).
        learning_rate: AdamW step size hyperparameter.
        epochs: Number of complete training epochs over dataset.
        batch_size: Mini-batch size for parallel AdamW gradient accumulation and updates.
        weight_decay: Decoupled L2 weight decay coefficient applied to model parameters.
        random_state: Optional seed for reproducible parameter initialization and sample shuffling.
        class_weight: Optional `"balanced"` inverse-frequency class weighting or `None`.
    """

    classes_: np.ndarray[tuple[int], np.dtype[Any]]
    embedding_: np.ndarray[tuple[int, int], np.dtype[np.float32]]
    conv_w_: np.ndarray[tuple[int, int, int], np.dtype[np.float32]]
    conv_b_: np.ndarray[tuple[int], np.dtype[np.float32]]
    dense_w_: np.ndarray[tuple[int, int], np.dtype[np.float32]]
    dense_b_: np.ndarray[tuple[int], np.dtype[np.float32]]
    source_w_: np.ndarray[tuple[int], np.dtype[np.float32]]
    source_b_: np.ndarray[tuple[int], np.dtype[np.float32]]
    loss_history_: list[float]
    lut_: np.ndarray[tuple[int, int, int], np.dtype[np.float32]]
    k_sizes_: np.ndarray[tuple[int], np.dtype[np.int32]]
    unique_ks_: np.ndarray[tuple[int], np.dtype[np.int32]]

    def __init__(
        self,
        vocab_size: int = 23,
        embed_dim: int = 16,
        num_filters: int = 64,
        kernel_size: int | tuple[int, ...] = (5, 9, 13),
        learning_rate: float = 0.01,
        epochs: int = 20,
        batch_size: int = 32,
        weight_decay: float = 1e-4,
        random_state: int | None = None,
        class_weight: str | None = "balanced",
    ) -> None:
        """Initialize the `GeneRoleClassifier` with architecture and optimizer hyperparameters."""
        self.vocab_size = vocab_size
        self.embed_dim = embed_dim
        self.num_filters = num_filters
        self.kernel_size = kernel_size
        self.learning_rate = learning_rate
        self.epochs = epochs
        self.batch_size = batch_size
        self.weight_decay = weight_decay
        self.random_state = random_state
        self.class_weight = class_weight

    @property
    def is_fitted(self) -> bool:
        """Return `True` if all model parameters have been fitted or loaded."""
        return all(
            getattr(self, attr, None) is not None
            for attr in ("classes_", "embedding_", "conv_w_", "conv_b_", "dense_w_", "dense_b_", "k_sizes_")
        )

    @property
    def num_classes(self) -> int:
        """Number of target classes learned by the fitted model."""
        return len(self.classes_) if getattr(self, "classes_", None) is not None else 0

    @property
    def feature_dim(self) -> int:
        """Dimensionality of the penultimate 1D-CNN dual-pooling (`[conv_max, conv_avg]`) feature vector."""
        if getattr(self, "dense_w_", None) is not None:
            return int(self.dense_w_.shape[0])
        return 2 * int(self.num_filters)

    def _pack_ragged(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    ) -> tuple[np.ndarray[tuple[int], np.dtype[np.uint8]], np.ndarray[tuple[int], np.dtype[np.int64]]]:
        """Pack sequence inputs into a flat `uint8` token array (`0..vocab_size-1`) and `int64` CSR `offsets`."""
        unk_tok = np.uint8(max(0, self.vocab_size - 1))

        if isinstance(X, Genes):
            X = X.translations

        if isinstance(X, Sequences):
            raw = X.seqs
            if len(raw) == 0:
                return np.empty(0, dtype=np.uint8), X.indptr.astype(np.int64, copy=False)
            if int(np.max(raw)) >= self.vocab_size:
                encoded = Alphabet.AMINO.encode(raw)
                flat_u8 = np.minimum(encoded, unk_tok)
            else:
                flat_u8 = raw.astype(np.uint8, copy=False)
            return flat_u8, X.indptr.astype(np.int64, copy=False)

        if isinstance(X, tuple) and len(X) == 2 and isinstance(X[0], np.ndarray) and isinstance(X[1], np.ndarray):
            raw_flat = X[0]
            if len(raw_flat) > 0 and raw_flat.dtype == np.uint8 and int(np.max(raw_flat)) >= self.vocab_size:
                flat_u8 = np.minimum(Alphabet.AMINO.encode(raw_flat), unk_tok)
            else:
                flat_u8 = np.clip(raw_flat, 0, int(unk_tok)).astype(np.uint8, copy=False)
            offsets = X[1].astype(np.int64, copy=False)
            if len(offsets) > 0 and offsets[-1] != len(flat_u8):
                offsets = np.append(offsets, len(flat_u8))
            return flat_u8, offsets

        seq_list = list(X)
        n_seqs = len(seq_list)
        if n_seqs == 0:
            return np.empty(0, dtype=np.uint8), np.zeros(1, dtype=np.int64)

        first = seq_list[0]
        if isinstance(first, (bytes, bytearray, str, SeqRecord)):
            bytes_list = [
                s.seq if isinstance(s, SeqRecord) else (s.encode("ascii") if isinstance(s, str) else bytes(s))
                for s in seq_list
            ]
            seqs_obj = Sequences.from_bytes(bytes_list)
            encoded = Alphabet.AMINO.encode(seqs_obj.seqs)
            return np.minimum(encoded, unk_tok), seqs_obj.indptr.astype(np.int64, copy=False)

        offsets = np.zeros(n_seqs + 1, dtype=np.int64)
        total_len = sum(len(s) for s in seq_list)
        flat = np.empty(total_len, dtype=np.uint8)
        curr = 0
        for i, s in enumerate(seq_list):
            n = len(s)
            if n > 0:
                arr = np.asarray(s)
                if arr.dtype == np.uint8 and int(np.max(arr)) >= self.vocab_size:
                    flat[curr : curr + n] = np.minimum(Alphabet.AMINO.encode(arr), unk_tok)
                else:
                    flat[curr : curr + n] = np.clip(arr, 0, int(unk_tok)).astype(np.uint8, copy=False)
            curr += n
            offsets[i + 1] = curr
        return flat, offsets

    def fit(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
        y: Sequence[int | GeneRole] | np.ndarray,
        sample_weight: Sequence[float] | np.ndarray | None = None,
        *,
        y_source: Sequence[int] | np.ndarray | None = None,
        source_loss_weight: float = 0.25,
        init_from: GeneRoleClassifier | None = None,
        backbone_lr_scale: float = 1.0,
    ) -> Self:
        """Fit the 1D-CNN using parallel mini-batch AdamW optimization over ragged token sequences."""
        y_raw = np.asarray(y)
        if len(y_raw) == 0:
            raise ValueError("Training dataset cannot be empty.")

        y_i64 = y_raw.astype(np.int64, copy=False)
        if self.__class__ is GeneRoleClassifier:
            pos_mask = (y_i64 >= 0) & (y_i64 != int(GeneRole.UNKNOWN))
        else:
            pos_mask = y_i64 >= 0

        if not np.any(pos_mask):
            raise ValueError("Training dataset must contain at least one positive class label.")

        y_pos = y_i64[pos_mask]
        if int(np.min(y_pos)) >= -128 and int(np.max(y_pos)) <= 127:
            y_arr = y_i64.astype(np.int8)
            self.classes_ = np.unique(y_pos).astype(np.int8)
        else:
            y_arr = y_i64
            self.classes_ = np.unique(y_pos).astype(np.int64)

        n_classes = len(self.classes_)
        if n_classes < 2:
            raise ValueError(f"{self.__class__.__name__} requires at least 2 distinct classes, found {n_classes}.")

        y_indexed = np.full(len(y_arr), -1, dtype=np.int32)
        y_indexed[pos_mask] = np.searchsorted(self.classes_, y_arr[pos_mask]).astype(np.int32)

        if sample_weight is not None:
            sw_arr = np.asarray(sample_weight, dtype=np.float32)
            if len(sw_arr) != len(y_arr):
                raise ValueError(f"Length of sample_weight ({len(sw_arr)}) does not match y ({len(y_arr)}).")
            sw_arr = np.maximum(sw_arr, np.float32(0.0))
            mean_sw = float(np.mean(sw_arr))
            if mean_sw > 0.0:
                sw_arr = (sw_arr / np.float32(mean_sw)).astype(np.float32)
        else:
            sw_arr = np.ones(len(y_arr), dtype=np.float32)

        if self.class_weight == "balanced":
            counts = np.bincount(y_indexed[pos_mask], weights=sw_arr[pos_mask], minlength=n_classes).astype(np.float32)
            counts_safe = np.maximum(counts, 1.0)
            weights = float(np.sum(sw_arr[pos_mask])) / (float(n_classes) * counts_safe)
            class_weights = (weights / np.mean(weights)).astype(np.float32)
        else:
            class_weights = np.ones(n_classes, dtype=np.float32)

        if y_source is not None:
            y_src_raw = np.asarray(y_source, dtype=np.int8)
            if len(y_src_raw) != len(y_arr):
                raise ValueError(f"Length of y_source ({len(y_src_raw)}) does not match y ({len(y_arr)}).")
            y_src_arr = np.where((y_src_raw == 0) | (y_src_raw == 1), y_src_raw, np.int8(-1))
            valid_src = y_src_arr >= 0
            if np.any(valid_src):
                src_counts = np.bincount(
                    y_src_arr[valid_src].astype(np.int32), weights=sw_arr[valid_src], minlength=2
                ).astype(np.float32)
                src_safe = np.maximum(src_counts, 1.0)
                src_w = float(np.sum(sw_arr[valid_src])) / (2.0 * src_safe)
                src_class_weights = (src_w / np.mean(src_w)).astype(np.float32)
            else:
                src_class_weights = np.ones(2, dtype=np.float32)
        else:
            y_src_arr = np.full(len(y_arr), -1, dtype=np.int8)
            src_class_weights = np.ones(2, dtype=np.float32)

        k_tuple = (self.kernel_size,) if isinstance(self.kernel_size, int) else tuple(int(k) for k in self.kernel_size)
        n_k = len(k_tuple)
        self.k_sizes_ = np.empty(self.num_filters, dtype=np.int32)
        base_count = self.num_filters // n_k
        remainder = self.num_filters % n_k
        idx = 0
        for i, k in enumerate(k_tuple):
            count = base_count + (1 if i < remainder else 0)
            self.k_sizes_[idx : idx + count] = k
            idx += count

        self.unique_ks_ = np.unique(self.k_sizes_).astype(np.int32)
        max_k = int(np.max(self.unique_ks_))

        rng = np.random.default_rng(self.random_state)
        if (
            init_from is not None
            and init_from.is_fitted
            and init_from.vocab_size == self.vocab_size
            and init_from.embed_dim == self.embed_dim
            and init_from.num_filters == self.num_filters
            and np.array_equal(init_from.k_sizes_, self.k_sizes_)
        ):
            self.embedding_ = init_from.embedding_.astype(np.float32, copy=True)
            self.conv_w_ = init_from.conv_w_.astype(np.float32, copy=True)
            self.conv_b_ = init_from.conv_b_.astype(np.float32, copy=True)
        else:
            self.embedding_ = rng.normal(0, 0.1, (self.vocab_size, self.embed_dim)).astype(np.float32)
            self.conv_w_ = np.zeros((self.num_filters, self.embed_dim, max_k), dtype=np.float32)
            for oc in range(self.num_filters):
                k_width = int(self.k_sizes_[oc])
                std = float(np.sqrt(2.0 / (self.embed_dim * k_width)))
                self.conv_w_[oc, :, :k_width] = rng.normal(0.0, std, (self.embed_dim, k_width)).astype(np.float32)
            self.conv_b_ = np.zeros(self.num_filters, dtype=np.float32)

        feat_dim = 2 * self.num_filters
        self.dense_w_ = rng.normal(
            0.0,
            np.sqrt(2.0 / feat_dim),
            (feat_dim, n_classes),
        ).astype(np.float32)
        self.dense_b_ = np.zeros(n_classes, dtype=np.float32)

        if (
            init_from is not None
            and getattr(init_from, "source_w_", None) is not None
            and len(init_from.source_w_) == feat_dim
        ):
            self.source_w_ = init_from.source_w_.astype(np.float32, copy=True)
            self.source_b_ = init_from.source_b_.astype(np.float32, copy=True)
        else:
            self.source_w_ = (
                rng.normal(0.0, np.sqrt(2.0 / feat_dim), feat_dim).astype(np.float32)
                if np.any(y_src_arr >= 0)
                else np.zeros(feat_dim, dtype=np.float32)
            )
            self.source_b_ = np.zeros(1, dtype=np.float32)

        x_flat, offsets = self._pack_ragged(X)
        if len(offsets) - 1 != len(y_arr):
            raise ValueError(f"Number of samples in X ({len(offsets) - 1}) does not match y ({len(y_arr)}).")

        seed = -1 if self.random_state is None else int(self.random_state)

        (
            self.embedding_,
            self.conv_w_,
            self.conv_b_,
            self.dense_w_,
            self.dense_b_,
            self.source_w_,
            self.source_b_,
            loss_hist,
        ) = _train_adam(
            x_flat,
            offsets,
            y_indexed,
            class_weights,
            sw_arr,
            y_src_arr,
            src_class_weights,
            float(source_loss_weight),
            self.embedding_,
            self.conv_w_,
            self.conv_b_,
            self.dense_w_,
            self.dense_b_,
            self.source_w_,
            self.source_b_,
            self.k_sizes_,
            int(self.epochs),
            int(self.batch_size),
            float(self.learning_rate),
            float(self.weight_decay),
            seed,
            float(backbone_lr_scale),
        )
        self.loss_history_ = [float(x) for x in loss_hist]
        self.lut_ = _compute_lut(self.embedding_, self.conv_w_, self.k_sizes_)
        if self.class_weight == "balanced" and n_classes > 1 and len(x_flat) > 0:
            n_bg = min(len(offsets) - 1, 256)
            bg_step = max(1, (len(offsets) - 1) // n_bg)
            bg_offsets = np.zeros(n_bg + 1, dtype=np.int64)
            for idx_b in range(n_bg):
                s_i = idx_b * bg_step
                bg_offsets[idx_b + 1] = bg_offsets[idx_b] + (offsets[s_i + 1] - offsets[s_i])
            bg_flat = np.empty(int(bg_offsets[-1]), dtype=x_flat.dtype)
            for idx_b in range(n_bg):
                s_i = idx_b * bg_step
                seg = x_flat[offsets[s_i] : offsets[s_i + 1]].copy()
                rng.shuffle(seg)
                bg_flat[bg_offsets[idx_b] : bg_offsets[idx_b + 1]] = seg
            bg_logits = _logits_batch(
                bg_flat,
                bg_offsets,
                self.lut_,
                self.conv_b_,
                self.dense_w_,
                self.dense_b_,
                self.k_sizes_,
            )
            mean_bg = np.mean(bg_logits, axis=0)
            self.dense_b_ -= np.float32(0.55) * (mean_bg - np.mean(mean_bg))
        return self

    def transform(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
        *,
        layer: str = "penultimate",
    ) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
        """Extract sequence representations from the specified network layer."""
        self._check_is_fitted()
        if layer == "logits":
            return self.decision_function(X)
        if layer == "proba":
            return self.predict_proba(X)

        x_flat, offsets = self._pack_ragged(X)
        if layer == "embed_mean":
            return _embed_mean_batch(x_flat, offsets, self.embedding_)

        h = _extract_features_batch(
            x_flat,
            offsets,
            self.lut_,
            self.conv_b_,
            self.k_sizes_,
            self.feature_dim,
        )
        nf = self.num_filters
        if layer in ("penultimate", "full"):
            return h
        if layer == "conv_max":
            return np.ascontiguousarray(h[:, :nf])
        if layer == "conv_avg":
            return np.ascontiguousarray(h[:, nf : 2 * nf])
        if layer == "task_weighted":
            w_norms = np.maximum(
                np.linalg.norm(self.dense_w_[: self.feature_dim], axis=1).astype(np.float32), np.float32(1e-6)
            )
            return np.ascontiguousarray(h * w_norms[None, :])
        if layer == "task_weighted_conv_max":
            w_norms = np.maximum(np.linalg.norm(self.dense_w_[:nf], axis=1).astype(np.float32), np.float32(1e-6))
            return np.ascontiguousarray(h[:, :nf] * w_norms[None, :])
        if layer.startswith("conv_k"):
            try:
                k_val = int(layer[6:])
            except ValueError as err:
                raise ValueError(f"Invalid kernel width in layer={layer!r}.") from err
            k_mask = np.where(self.k_sizes_ == k_val)[0]
            if len(k_mask) == 0:
                raise ValueError(
                    f"No filters with kernel size {k_val} in model (available: {self.unique_ks_.tolist()})."
                )
            idx = np.concatenate([k_mask, nf + k_mask])
            return np.ascontiguousarray(h[:, idx])
        raise ValueError(
            f"Unsupported layer {layer!r}. Expected 'penultimate', 'conv_max', 'conv_avg', "
            "'task_weighted', 'task_weighted_conv_max', 'embed_mean', 'logits', 'proba', or 'conv_k<N>'."
        )

    def decision_function(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    ) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
        """Compute raw pre-softmax class logits `(n_samples, n_classes)` for each sequence in `X`."""
        self._check_is_fitted()
        x_flat, offsets = self._pack_ragged(X)
        return _logits_batch(
            x_flat,
            offsets,
            self.lut_,
            self.conv_b_,
            self.dense_w_,
            self.dense_b_,
            self.k_sizes_,
        )

    def predict_proba(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    ) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
        """Predict class probabilities `(n_samples, n_classes)` for each sequence in `X`."""
        self._check_is_fitted()
        x_flat, offsets = self._pack_ragged(X)
        return _forward_batch(
            x_flat,
            offsets,
            self.lut_,
            self.conv_b_,
            self.dense_w_,
            self.dense_b_,
            self.k_sizes_,
        )

    def predict_source_proba(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    ) -> np.ndarray[tuple[int], np.dtype[np.float32]]:
        """Predict probability `P(phage | x)` in `[0.0, 1.0]` for each sequence in `X`."""
        self._check_is_fitted()
        x_flat, offsets = self._pack_ragged(X)
        if len(offsets) <= 1:
            return np.empty(0, dtype=np.float32)
        h = _extract_features_batch(
            x_flat,
            offsets,
            self.lut_,
            self.conv_b_,
            self.k_sizes_,
            self.feature_dim,
        )
        sw = getattr(self, "source_w_", None)
        sb = getattr(self, "source_b_", None)
        if sw is None or sb is None or len(sw) != h.shape[1] or not np.any(sw != 0.0):
            return np.full(h.shape[0], 0.5, dtype=np.float32)
        logits = np.clip(h @ sw + sb[0], np.float32(-30.0), np.float32(30.0))
        return (np.float32(1.0) / (np.float32(1.0) + np.exp(-logits))).astype(np.float32)

    def predict_source(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
        threshold: float = 0.5,
    ) -> np.ndarray[tuple[int], np.dtype[np.uint8]]:
        """Predict binary genomic source (`0` = bacterial, `1` = phage/prophage) for each sequence in `X`."""
        probs = self.predict_source_proba(X)
        return (probs >= np.float32(threshold)).astype(np.uint8)

    def predict(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
        min_confidence: float = 0.0,
        unknown_label: int | None = None,
    ) -> np.ndarray[tuple[int], np.dtype[Any]]:
        """Predict most likely class label (e.g. `GeneRole` integer index) for each sequence in `X`."""
        probs = self.predict_proba(X)
        out_dtype = self.classes_.dtype
        if len(probs) == 0:
            return np.empty(0, dtype=out_dtype)
        top_indices = np.argmax(probs, axis=1)
        top_classes = self.classes_[top_indices].astype(out_dtype)
        if min_confidence > 0.0:
            unk = (
                (int(GeneRole.UNKNOWN) if out_dtype == np.int8 else -1) if unknown_label is None else int(unknown_label)
            )
            max_probs = np.max(probs, axis=1)
            return np.where(max_probs >= min_confidence, top_classes, unk).astype(out_dtype)
        return top_classes

    def score(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
        y: Sequence[int | GeneRole] | np.ndarray,
    ) -> float:
        """Return classification accuracy on `(X, y)`."""
        preds = self.predict(X)
        y_arr = np.asarray(y, dtype=self.classes_.dtype)
        return float(np.mean(preds == y_arr))

    def top_k_accuracy(
        self,
        X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
        y: Sequence[int | GeneRole] | np.ndarray,
        k: int = 3,
    ) -> float:
        """Compute Top-`k` classification accuracy on `(X, y)`."""
        probs = self.predict_proba(X)
        y_arr = np.asarray(y, dtype=self.classes_.dtype)
        if len(y_arr) == 0 or len(probs) == 0:
            return 0.0
        k_eff = min(max(1, int(k)), self.num_classes)
        top_k_idx = np.argsort(-probs, axis=1)[:, :k_eff]
        top_k_classes = self.classes_[top_k_idx]
        hits = np.any(top_k_classes == y_arr[:, None], axis=1)
        return float(np.mean(hits))

    def _check_is_fitted(self) -> None:
        """Verify that all required model parameters are fitted."""
        if not self.is_fitted:
            raise RuntimeError(
                f"This {self.__class__.__name__} instance is not fitted yet. "
                "Call 'fit' or load pre-trained weights first."
            )
        if getattr(self, "lut_", None) is None:
            self.lut_ = _compute_lut(self.embedding_, self.conv_w_, self.k_sizes_)

    def save(self, path: Path | str) -> None:
        """Save model weights, classes, and hyperparameters to a compressed `.npz` archive."""
        self._check_is_fitted()
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)

        cw_flag = 1 if self.class_weight == "balanced" else 0
        first_k = int(self.k_sizes_[0])
        meta = np.array(
            [
                self.vocab_size,
                self.embed_dim,
                self.num_filters,
                first_k,
                self.batch_size,
                self.epochs,
                cw_flag,
            ],
            dtype=np.int64,
        )
        float_meta = np.array(
            [
                self.learning_rate,
                self.weight_decay,
                -1.0 if self.random_state is None else float(self.random_state),
            ],
            dtype=np.float64,
        )
        sw = getattr(self, "source_w_", np.zeros(self.feature_dim, dtype=np.float32))
        sb = getattr(self, "source_b_", np.zeros(1, dtype=np.float32))
        np.savez_compressed(
            p,
            meta=meta,
            float_meta=float_meta,
            classes=self.classes_,
            embedding=self.embedding_,
            conv_w=self.conv_w_,
            conv_b=self.conv_b_,
            dense_w=self.dense_w_,
            dense_b=self.dense_b_,
            source_w=sw,
            source_b=sb,
            k_sizes=self.k_sizes_,
            loss_history=np.array(getattr(self, "loss_history_", []), dtype=np.float32),
        )

    @classmethod
    def load(cls, path: Path | str) -> Self:
        """Load model weights, classes, and hyperparameters from a compressed `.npz` archive."""
        p = Path(path)
        if not p.suffix:
            p = p.with_suffix(".npz")

        with np.load(p) as data:
            meta = data["meta"]
            float_meta = data["float_meta"]
            rs = None if float_meta[2] < 0 else int(float_meta[2])
            cw = "balanced" if (len(meta) > 6 and meta[6] == 1) else None

            num_filters = int(meta[2])
            if "k_sizes" in data:
                k_sizes = data["k_sizes"].astype(np.int32)
                ordered_ks = tuple(dict.fromkeys(int(x) for x in k_sizes))
                ks_param: int | tuple[int, ...] = ordered_ks[0] if len(ordered_ks) == 1 else ordered_ks
            else:
                ks_param = int(meta[3])
                k_sizes = np.full(num_filters, ks_param, dtype=np.int32)

            inst = cls(
                vocab_size=int(meta[0]),
                embed_dim=int(meta[1]),
                num_filters=num_filters,
                kernel_size=ks_param,
                batch_size=int(meta[4]),
                epochs=int(meta[5]),
                learning_rate=float(float_meta[0]),
                weight_decay=float(float_meta[1]),
                random_state=rs,
                class_weight=cw,
            )
            raw_classes = data["classes"]
            if (
                len(raw_classes) > 0
                and int(np.min(raw_classes)) >= -128
                and int(np.max(raw_classes)) <= 127
                and raw_classes.dtype == np.int8
            ):
                inst.classes_ = raw_classes.astype(np.int8)
            else:
                inst.classes_ = raw_classes.astype(np.int64)
            inst.embedding_ = data["embedding"].astype(np.float32)
            inst.conv_w_ = data["conv_w"].astype(np.float32)
            inst.conv_b_ = data["conv_b"].astype(np.float32)
            inst.dense_w_ = data["dense_w"].astype(np.float32)
            inst.dense_b_ = data["dense_b"].astype(np.float32)
            f_dim = int(inst.dense_w_.shape[0])
            inst.source_w_ = (
                data["source_w"].astype(np.float32) if "source_w" in data else np.zeros(f_dim, dtype=np.float32)
            )
            inst.source_b_ = (
                data["source_b"].astype(np.float32) if "source_b" in data else np.zeros(1, dtype=np.float32)
            )
            inst.k_sizes_ = k_sizes
            inst.unique_ks_ = np.unique(k_sizes).astype(np.int32)
            inst.loss_history_ = [float(x) for x in data["loss_history"].tolist()]
            inst.lut_ = _compute_lut(inst.embedding_, inst.conv_w_, inst.k_sizes_)
            return inst

    @classmethod
    @cache
    def pretrained(cls) -> Self:
        """Load the packaged pre-trained `GeneRoleClassifier` from `serogen.data/model.npz`."""
        resource = files("seroref.data").joinpath("model.npz")
        with as_file(resource) as model_path:
            return cls.load(model_path)

is_fitted `property` ¶

is_fitted: bool

Return True if all model parameters have been fitted or loaded.

num_classes `property` ¶

num_classes: int

Number of target classes learned by the fitted model.

feature_dim `property` ¶

feature_dim: int

Dimensionality of the penultimate 1D-CNN dual-pooling ([conv_max, conv_avg]) feature vector.

fit ¶

fit(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
    y: Sequence[int | GeneRole] | ndarray,
    sample_weight: Sequence[float] | ndarray | None = None,
    *,
    y_source: Sequence[int] | ndarray | None = None,
    source_loss_weight: float = 0.25,
    init_from: GeneRoleClassifier | None = None,
    backbone_lr_scale: float = 1.0,
) -> Self

Fit the 1D-CNN using parallel mini-batch AdamW optimization over ragged token sequences.

Source code in src/seroref/genes/cnn.py

def fit(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    y: Sequence[int | GeneRole] | np.ndarray,
    sample_weight: Sequence[float] | np.ndarray | None = None,
    *,
    y_source: Sequence[int] | np.ndarray | None = None,
    source_loss_weight: float = 0.25,
    init_from: GeneRoleClassifier | None = None,
    backbone_lr_scale: float = 1.0,
) -> Self:
    """Fit the 1D-CNN using parallel mini-batch AdamW optimization over ragged token sequences."""
    y_raw = np.asarray(y)
    if len(y_raw) == 0:
        raise ValueError("Training dataset cannot be empty.")

    y_i64 = y_raw.astype(np.int64, copy=False)
    if self.__class__ is GeneRoleClassifier:
        pos_mask = (y_i64 >= 0) & (y_i64 != int(GeneRole.UNKNOWN))
    else:
        pos_mask = y_i64 >= 0

    if not np.any(pos_mask):
        raise ValueError("Training dataset must contain at least one positive class label.")

    y_pos = y_i64[pos_mask]
    if int(np.min(y_pos)) >= -128 and int(np.max(y_pos)) <= 127:
        y_arr = y_i64.astype(np.int8)
        self.classes_ = np.unique(y_pos).astype(np.int8)
    else:
        y_arr = y_i64
        self.classes_ = np.unique(y_pos).astype(np.int64)

    n_classes = len(self.classes_)
    if n_classes < 2:
        raise ValueError(f"{self.__class__.__name__} requires at least 2 distinct classes, found {n_classes}.")

    y_indexed = np.full(len(y_arr), -1, dtype=np.int32)
    y_indexed[pos_mask] = np.searchsorted(self.classes_, y_arr[pos_mask]).astype(np.int32)

    if sample_weight is not None:
        sw_arr = np.asarray(sample_weight, dtype=np.float32)
        if len(sw_arr) != len(y_arr):
            raise ValueError(f"Length of sample_weight ({len(sw_arr)}) does not match y ({len(y_arr)}).")
        sw_arr = np.maximum(sw_arr, np.float32(0.0))
        mean_sw = float(np.mean(sw_arr))
        if mean_sw > 0.0:
            sw_arr = (sw_arr / np.float32(mean_sw)).astype(np.float32)
    else:
        sw_arr = np.ones(len(y_arr), dtype=np.float32)

    if self.class_weight == "balanced":
        counts = np.bincount(y_indexed[pos_mask], weights=sw_arr[pos_mask], minlength=n_classes).astype(np.float32)
        counts_safe = np.maximum(counts, 1.0)
        weights = float(np.sum(sw_arr[pos_mask])) / (float(n_classes) * counts_safe)
        class_weights = (weights / np.mean(weights)).astype(np.float32)
    else:
        class_weights = np.ones(n_classes, dtype=np.float32)

    if y_source is not None:
        y_src_raw = np.asarray(y_source, dtype=np.int8)
        if len(y_src_raw) != len(y_arr):
            raise ValueError(f"Length of y_source ({len(y_src_raw)}) does not match y ({len(y_arr)}).")
        y_src_arr = np.where((y_src_raw == 0) | (y_src_raw == 1), y_src_raw, np.int8(-1))
        valid_src = y_src_arr >= 0
        if np.any(valid_src):
            src_counts = np.bincount(
                y_src_arr[valid_src].astype(np.int32), weights=sw_arr[valid_src], minlength=2
            ).astype(np.float32)
            src_safe = np.maximum(src_counts, 1.0)
            src_w = float(np.sum(sw_arr[valid_src])) / (2.0 * src_safe)
            src_class_weights = (src_w / np.mean(src_w)).astype(np.float32)
        else:
            src_class_weights = np.ones(2, dtype=np.float32)
    else:
        y_src_arr = np.full(len(y_arr), -1, dtype=np.int8)
        src_class_weights = np.ones(2, dtype=np.float32)

    k_tuple = (self.kernel_size,) if isinstance(self.kernel_size, int) else tuple(int(k) for k in self.kernel_size)
    n_k = len(k_tuple)
    self.k_sizes_ = np.empty(self.num_filters, dtype=np.int32)
    base_count = self.num_filters // n_k
    remainder = self.num_filters % n_k
    idx = 0
    for i, k in enumerate(k_tuple):
        count = base_count + (1 if i < remainder else 0)
        self.k_sizes_[idx : idx + count] = k
        idx += count

    self.unique_ks_ = np.unique(self.k_sizes_).astype(np.int32)
    max_k = int(np.max(self.unique_ks_))

    rng = np.random.default_rng(self.random_state)
    if (
        init_from is not None
        and init_from.is_fitted
        and init_from.vocab_size == self.vocab_size
        and init_from.embed_dim == self.embed_dim
        and init_from.num_filters == self.num_filters
        and np.array_equal(init_from.k_sizes_, self.k_sizes_)
    ):
        self.embedding_ = init_from.embedding_.astype(np.float32, copy=True)
        self.conv_w_ = init_from.conv_w_.astype(np.float32, copy=True)
        self.conv_b_ = init_from.conv_b_.astype(np.float32, copy=True)
    else:
        self.embedding_ = rng.normal(0, 0.1, (self.vocab_size, self.embed_dim)).astype(np.float32)
        self.conv_w_ = np.zeros((self.num_filters, self.embed_dim, max_k), dtype=np.float32)
        for oc in range(self.num_filters):
            k_width = int(self.k_sizes_[oc])
            std = float(np.sqrt(2.0 / (self.embed_dim * k_width)))
            self.conv_w_[oc, :, :k_width] = rng.normal(0.0, std, (self.embed_dim, k_width)).astype(np.float32)
        self.conv_b_ = np.zeros(self.num_filters, dtype=np.float32)

    feat_dim = 2 * self.num_filters
    self.dense_w_ = rng.normal(
        0.0,
        np.sqrt(2.0 / feat_dim),
        (feat_dim, n_classes),
    ).astype(np.float32)
    self.dense_b_ = np.zeros(n_classes, dtype=np.float32)

    if (
        init_from is not None
        and getattr(init_from, "source_w_", None) is not None
        and len(init_from.source_w_) == feat_dim
    ):
        self.source_w_ = init_from.source_w_.astype(np.float32, copy=True)
        self.source_b_ = init_from.source_b_.astype(np.float32, copy=True)
    else:
        self.source_w_ = (
            rng.normal(0.0, np.sqrt(2.0 / feat_dim), feat_dim).astype(np.float32)
            if np.any(y_src_arr >= 0)
            else np.zeros(feat_dim, dtype=np.float32)
        )
        self.source_b_ = np.zeros(1, dtype=np.float32)

    x_flat, offsets = self._pack_ragged(X)
    if len(offsets) - 1 != len(y_arr):
        raise ValueError(f"Number of samples in X ({len(offsets) - 1}) does not match y ({len(y_arr)}).")

    seed = -1 if self.random_state is None else int(self.random_state)

    (
        self.embedding_,
        self.conv_w_,
        self.conv_b_,
        self.dense_w_,
        self.dense_b_,
        self.source_w_,
        self.source_b_,
        loss_hist,
    ) = _train_adam(
        x_flat,
        offsets,
        y_indexed,
        class_weights,
        sw_arr,
        y_src_arr,
        src_class_weights,
        float(source_loss_weight),
        self.embedding_,
        self.conv_w_,
        self.conv_b_,
        self.dense_w_,
        self.dense_b_,
        self.source_w_,
        self.source_b_,
        self.k_sizes_,
        int(self.epochs),
        int(self.batch_size),
        float(self.learning_rate),
        float(self.weight_decay),
        seed,
        float(backbone_lr_scale),
    )
    self.loss_history_ = [float(x) for x in loss_hist]
    self.lut_ = _compute_lut(self.embedding_, self.conv_w_, self.k_sizes_)
    if self.class_weight == "balanced" and n_classes > 1 and len(x_flat) > 0:
        n_bg = min(len(offsets) - 1, 256)
        bg_step = max(1, (len(offsets) - 1) // n_bg)
        bg_offsets = np.zeros(n_bg + 1, dtype=np.int64)
        for idx_b in range(n_bg):
            s_i = idx_b * bg_step
            bg_offsets[idx_b + 1] = bg_offsets[idx_b] + (offsets[s_i + 1] - offsets[s_i])
        bg_flat = np.empty(int(bg_offsets[-1]), dtype=x_flat.dtype)
        for idx_b in range(n_bg):
            s_i = idx_b * bg_step
            seg = x_flat[offsets[s_i] : offsets[s_i + 1]].copy()
            rng.shuffle(seg)
            bg_flat[bg_offsets[idx_b] : bg_offsets[idx_b + 1]] = seg
        bg_logits = _logits_batch(
            bg_flat,
            bg_offsets,
            self.lut_,
            self.conv_b_,
            self.dense_w_,
            self.dense_b_,
            self.k_sizes_,
        )
        mean_bg = np.mean(bg_logits, axis=0)
        self.dense_b_ -= np.float32(0.55) * (mean_bg - np.mean(mean_bg))
    return self

transform ¶

transform(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
    *,
    layer: str = "penultimate",
) -> ndarray[tuple[int, int], dtype[float32]]

Extract sequence representations from the specified network layer.

Source code in src/seroref/genes/cnn.py

def transform(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    *,
    layer: str = "penultimate",
) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
    """Extract sequence representations from the specified network layer."""
    self._check_is_fitted()
    if layer == "logits":
        return self.decision_function(X)
    if layer == "proba":
        return self.predict_proba(X)

    x_flat, offsets = self._pack_ragged(X)
    if layer == "embed_mean":
        return _embed_mean_batch(x_flat, offsets, self.embedding_)

    h = _extract_features_batch(
        x_flat,
        offsets,
        self.lut_,
        self.conv_b_,
        self.k_sizes_,
        self.feature_dim,
    )
    nf = self.num_filters
    if layer in ("penultimate", "full"):
        return h
    if layer == "conv_max":
        return np.ascontiguousarray(h[:, :nf])
    if layer == "conv_avg":
        return np.ascontiguousarray(h[:, nf : 2 * nf])
    if layer == "task_weighted":
        w_norms = np.maximum(
            np.linalg.norm(self.dense_w_[: self.feature_dim], axis=1).astype(np.float32), np.float32(1e-6)
        )
        return np.ascontiguousarray(h * w_norms[None, :])
    if layer == "task_weighted_conv_max":
        w_norms = np.maximum(np.linalg.norm(self.dense_w_[:nf], axis=1).astype(np.float32), np.float32(1e-6))
        return np.ascontiguousarray(h[:, :nf] * w_norms[None, :])
    if layer.startswith("conv_k"):
        try:
            k_val = int(layer[6:])
        except ValueError as err:
            raise ValueError(f"Invalid kernel width in layer={layer!r}.") from err
        k_mask = np.where(self.k_sizes_ == k_val)[0]
        if len(k_mask) == 0:
            raise ValueError(
                f"No filters with kernel size {k_val} in model (available: {self.unique_ks_.tolist()})."
            )
        idx = np.concatenate([k_mask, nf + k_mask])
        return np.ascontiguousarray(h[:, idx])
    raise ValueError(
        f"Unsupported layer {layer!r}. Expected 'penultimate', 'conv_max', 'conv_avg', "
        "'task_weighted', 'task_weighted_conv_max', 'embed_mean', 'logits', 'proba', or 'conv_k<N>'."
    )

decision_function ¶

decision_function(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
) -> ndarray[tuple[int, int], dtype[float32]]

Compute raw pre-softmax class logits (n_samples, n_classes) for each sequence in X.

Source code in src/seroref/genes/cnn.py

def decision_function(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
    """Compute raw pre-softmax class logits `(n_samples, n_classes)` for each sequence in `X`."""
    self._check_is_fitted()
    x_flat, offsets = self._pack_ragged(X)
    return _logits_batch(
        x_flat,
        offsets,
        self.lut_,
        self.conv_b_,
        self.dense_w_,
        self.dense_b_,
        self.k_sizes_,
    )

predict_proba ¶

predict_proba(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
) -> ndarray[tuple[int, int], dtype[float32]]

Predict class probabilities (n_samples, n_classes) for each sequence in X.

Source code in src/seroref/genes/cnn.py

def predict_proba(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
) -> np.ndarray[tuple[int, int], np.dtype[np.float32]]:
    """Predict class probabilities `(n_samples, n_classes)` for each sequence in `X`."""
    self._check_is_fitted()
    x_flat, offsets = self._pack_ragged(X)
    return _forward_batch(
        x_flat,
        offsets,
        self.lut_,
        self.conv_b_,
        self.dense_w_,
        self.dense_b_,
        self.k_sizes_,
    )

predict_source_proba ¶

predict_source_proba(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
) -> ndarray[tuple[int], dtype[float32]]

Predict probability P(phage | x) in [0.0, 1.0] for each sequence in X.

Source code in src/seroref/genes/cnn.py

def predict_source_proba(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
) -> np.ndarray[tuple[int], np.dtype[np.float32]]:
    """Predict probability `P(phage | x)` in `[0.0, 1.0]` for each sequence in `X`."""
    self._check_is_fitted()
    x_flat, offsets = self._pack_ragged(X)
    if len(offsets) <= 1:
        return np.empty(0, dtype=np.float32)
    h = _extract_features_batch(
        x_flat,
        offsets,
        self.lut_,
        self.conv_b_,
        self.k_sizes_,
        self.feature_dim,
    )
    sw = getattr(self, "source_w_", None)
    sb = getattr(self, "source_b_", None)
    if sw is None or sb is None or len(sw) != h.shape[1] or not np.any(sw != 0.0):
        return np.full(h.shape[0], 0.5, dtype=np.float32)
    logits = np.clip(h @ sw + sb[0], np.float32(-30.0), np.float32(30.0))
    return (np.float32(1.0) / (np.float32(1.0) + np.exp(-logits))).astype(np.float32)

predict_source ¶

predict_source(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
    threshold: float = 0.5,
) -> ndarray[tuple[int], dtype[uint8]]

Predict binary genomic source (0 = bacterial, 1 = phage/prophage) for each sequence in X.

Source code in src/seroref/genes/cnn.py

def predict_source(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    threshold: float = 0.5,
) -> np.ndarray[tuple[int], np.dtype[np.uint8]]:
    """Predict binary genomic source (`0` = bacterial, `1` = phage/prophage) for each sequence in `X`."""
    probs = self.predict_source_proba(X)
    return (probs >= np.float32(threshold)).astype(np.uint8)

predict ¶

predict(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
    min_confidence: float = 0.0,
    unknown_label: int | None = None,
) -> ndarray[tuple[int], dtype[Any]]

Predict most likely class label (e.g. GeneRole integer index) for each sequence in X.

Source code in src/seroref/genes/cnn.py

def predict(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    min_confidence: float = 0.0,
    unknown_label: int | None = None,
) -> np.ndarray[tuple[int], np.dtype[Any]]:
    """Predict most likely class label (e.g. `GeneRole` integer index) for each sequence in `X`."""
    probs = self.predict_proba(X)
    out_dtype = self.classes_.dtype
    if len(probs) == 0:
        return np.empty(0, dtype=out_dtype)
    top_indices = np.argmax(probs, axis=1)
    top_classes = self.classes_[top_indices].astype(out_dtype)
    if min_confidence > 0.0:
        unk = (
            (int(GeneRole.UNKNOWN) if out_dtype == np.int8 else -1) if unknown_label is None else int(unknown_label)
        )
        max_probs = np.max(probs, axis=1)
        return np.where(max_probs >= min_confidence, top_classes, unk).astype(out_dtype)
    return top_classes

score ¶

score(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
    y: Sequence[int | GeneRole] | ndarray,
) -> float

Return classification accuracy on (X, y).

Source code in src/seroref/genes/cnn.py

def score(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    y: Sequence[int | GeneRole] | np.ndarray,
) -> float:
    """Return classification accuracy on `(X, y)`."""
    preds = self.predict(X)
    y_arr = np.asarray(y, dtype=self.classes_.dtype)
    return float(np.mean(preds == y_arr))

top_k_accuracy ¶

top_k_accuracy(
    X: Genes
    | Sequences
    | Sequence[Any]
    | tuple[ndarray, ndarray],
    y: Sequence[int | GeneRole] | ndarray,
    k: int = 3,
) -> float

Compute Top-k classification accuracy on (X, y).

Source code in src/seroref/genes/cnn.py

def top_k_accuracy(
    self,
    X: Genes | Sequences | Sequence[Any] | tuple[np.ndarray, np.ndarray],
    y: Sequence[int | GeneRole] | np.ndarray,
    k: int = 3,
) -> float:
    """Compute Top-`k` classification accuracy on `(X, y)`."""
    probs = self.predict_proba(X)
    y_arr = np.asarray(y, dtype=self.classes_.dtype)
    if len(y_arr) == 0 or len(probs) == 0:
        return 0.0
    k_eff = min(max(1, int(k)), self.num_classes)
    top_k_idx = np.argsort(-probs, axis=1)[:, :k_eff]
    top_k_classes = self.classes_[top_k_idx]
    hits = np.any(top_k_classes == y_arr[:, None], axis=1)
    return float(np.mean(hits))

save ¶

save(path: Path | str) -> None

Save model weights, classes, and hyperparameters to a compressed .npz archive.

Source code in src/seroref/genes/cnn.py

def save(self, path: Path | str) -> None:
    """Save model weights, classes, and hyperparameters to a compressed `.npz` archive."""
    self._check_is_fitted()
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)

    cw_flag = 1 if self.class_weight == "balanced" else 0
    first_k = int(self.k_sizes_[0])
    meta = np.array(
        [
            self.vocab_size,
            self.embed_dim,
            self.num_filters,
            first_k,
            self.batch_size,
            self.epochs,
            cw_flag,
        ],
        dtype=np.int64,
    )
    float_meta = np.array(
        [
            self.learning_rate,
            self.weight_decay,
            -1.0 if self.random_state is None else float(self.random_state),
        ],
        dtype=np.float64,
    )
    sw = getattr(self, "source_w_", np.zeros(self.feature_dim, dtype=np.float32))
    sb = getattr(self, "source_b_", np.zeros(1, dtype=np.float32))
    np.savez_compressed(
        p,
        meta=meta,
        float_meta=float_meta,
        classes=self.classes_,
        embedding=self.embedding_,
        conv_w=self.conv_w_,
        conv_b=self.conv_b_,
        dense_w=self.dense_w_,
        dense_b=self.dense_b_,
        source_w=sw,
        source_b=sb,
        k_sizes=self.k_sizes_,
        loss_history=np.array(getattr(self, "loss_history_", []), dtype=np.float32),
    )

load `classmethod` ¶

load(path: Path | str) -> Self

Load model weights, classes, and hyperparameters from a compressed .npz archive.

Source code in src/seroref/genes/cnn.py

@classmethod
def load(cls, path: Path | str) -> Self:
    """Load model weights, classes, and hyperparameters from a compressed `.npz` archive."""
    p = Path(path)
    if not p.suffix:
        p = p.with_suffix(".npz")

    with np.load(p) as data:
        meta = data["meta"]
        float_meta = data["float_meta"]
        rs = None if float_meta[2] < 0 else int(float_meta[2])
        cw = "balanced" if (len(meta) > 6 and meta[6] == 1) else None

        num_filters = int(meta[2])
        if "k_sizes" in data:
            k_sizes = data["k_sizes"].astype(np.int32)
            ordered_ks = tuple(dict.fromkeys(int(x) for x in k_sizes))
            ks_param: int | tuple[int, ...] = ordered_ks[0] if len(ordered_ks) == 1 else ordered_ks
        else:
            ks_param = int(meta[3])
            k_sizes = np.full(num_filters, ks_param, dtype=np.int32)

        inst = cls(
            vocab_size=int(meta[0]),
            embed_dim=int(meta[1]),
            num_filters=num_filters,
            kernel_size=ks_param,
            batch_size=int(meta[4]),
            epochs=int(meta[5]),
            learning_rate=float(float_meta[0]),
            weight_decay=float(float_meta[1]),
            random_state=rs,
            class_weight=cw,
        )
        raw_classes = data["classes"]
        if (
            len(raw_classes) > 0
            and int(np.min(raw_classes)) >= -128
            and int(np.max(raw_classes)) <= 127
            and raw_classes.dtype == np.int8
        ):
            inst.classes_ = raw_classes.astype(np.int8)
        else:
            inst.classes_ = raw_classes.astype(np.int64)
        inst.embedding_ = data["embedding"].astype(np.float32)
        inst.conv_w_ = data["conv_w"].astype(np.float32)
        inst.conv_b_ = data["conv_b"].astype(np.float32)
        inst.dense_w_ = data["dense_w"].astype(np.float32)
        inst.dense_b_ = data["dense_b"].astype(np.float32)
        f_dim = int(inst.dense_w_.shape[0])
        inst.source_w_ = (
            data["source_w"].astype(np.float32) if "source_w" in data else np.zeros(f_dim, dtype=np.float32)
        )
        inst.source_b_ = (
            data["source_b"].astype(np.float32) if "source_b" in data else np.zeros(1, dtype=np.float32)
        )
        inst.k_sizes_ = k_sizes
        inst.unique_ks_ = np.unique(k_sizes).astype(np.int32)
        inst.loss_history_ = [float(x) for x in data["loss_history"].tolist()]
        inst.lut_ = _compute_lut(inst.embedding_, inst.conv_w_, inst.k_sizes_)
        return inst

pretrained `cached` `classmethod` ¶

pretrained() -> Self

Load the packaged pre-trained GeneRoleClassifier from serogen.data/model.npz.

Source code in src/seroref/genes/cnn.py

@classmethod
@cache
def pretrained(cls) -> Self:
    """Load the packaged pre-trained `GeneRoleClassifier` from `serogen.data/model.npz`."""
    resource = files("seroref.data").joinpath("model.npz")
    with as_file(resource) as model_path:
        return cls.load(model_path)

seroref.genes¶

ORF Prediction (seroref.genes.orf.predictor)¶

predictor ¶

OrfPredictor ¶

predict ¶

Gene Containers & HMM Parameters (seroref.genes.models)¶

models ¶

IndelKind ¶

IndelRecord dataclass ¶

Indels dataclass ¶

empty classmethod ¶

concat classmethod ¶

Gene dataclass ¶

Genes dataclass ¶

has_frameshift property ¶

frameshift_count property ¶

translations property ¶

translate ¶

empty classmethod ¶

concat classmethod ¶

HMM dataclass ¶

load classmethod ¶

get_hmm_model cached ¶

Functional Gene Role Ontology (seroref.genes.roles)¶

roles ¶

RoleCategory ¶

label property ¶

from_name classmethod ¶

resolve classmethod ¶

RoleSpec dataclass ¶

is_mobile_element property ¶

is_biosynthetic property ¶

RoleSpecs dataclass ¶

biosynthetic_mask property ¶

machinery_roles property ¶

empty classmethod ¶

from_list classmethod ¶

concat classmethod ¶

GeneRole ¶

spec property ¶

abbreviation property ¶

category property ¶

description property ¶

aliases property ¶

bitmask property ¶

is_mobile_element property ¶

is_biosynthetic property ¶

is_machinery property ¶

specs classmethod ¶

from_name classmethod ¶

resolve classmethod ¶

names_array classmethod ¶

abbreviations_array classmethod ¶

categories_array classmethod ¶

feature_matrix classmethod ¶

1D CNN Role Classifier (seroref.genes.cnn)¶

cnn ¶

GeneRoleClassifier ¶

is_fitted property ¶

num_classes property ¶

feature_dim property ¶

fit ¶

transform ¶

decision_function ¶

predict_proba ¶

predict_source_proba ¶

predict_source ¶

predict ¶

score ¶

top_k_accuracy ¶

save ¶

load classmethod ¶

pretrained cached classmethod ¶

Comments

`seroref.genes`¶

ORF Prediction (`seroref.genes.orf.predictor`)¶

Gene Containers & HMM Parameters (`seroref.genes.models`)¶

IndelRecord `dataclass` ¶

Indels `dataclass` ¶

empty `classmethod` ¶

concat `classmethod` ¶

Gene `dataclass` ¶

Genes `dataclass` ¶

has_frameshift `property` ¶

frameshift_count `property` ¶

translations `property` ¶

empty `classmethod` ¶

concat `classmethod` ¶

HMM `dataclass` ¶

load `classmethod` ¶

get_hmm_model `cached` ¶

Functional Gene Role Ontology (`seroref.genes.roles`)¶

label `property` ¶

from_name `classmethod` ¶

resolve `classmethod` ¶

RoleSpec `dataclass` ¶

is_mobile_element `property` ¶

is_biosynthetic `property` ¶

RoleSpecs `dataclass` ¶

biosynthetic_mask `property` ¶

machinery_roles `property` ¶

empty `classmethod` ¶

from_list `classmethod` ¶

concat `classmethod` ¶

spec `property` ¶

abbreviation `property` ¶

category `property` ¶

description `property` ¶

aliases `property` ¶

bitmask `property` ¶

is_mobile_element `property` ¶

is_biosynthetic `property` ¶

is_machinery `property` ¶

specs `classmethod` ¶

from_name `classmethod` ¶

resolve `classmethod` ¶

names_array `classmethod` ¶

abbreviations_array `classmethod` ¶

categories_array `classmethod` ¶

feature_matrix `classmethod` ¶

1D CNN Role Classifier (`seroref.genes.cnn`)¶

is_fitted `property` ¶

num_classes `property` ¶

feature_dim `property` ¶

load `classmethod` ¶

pretrained `cached` `classmethod` ¶