`serogen.genes`¶

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

ORF Prediction (`serogen.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, Sequences, or GenomeAssembly containers using a unified flat window table and Numba prange parallelism, returning batched Genes. Supports bacterial and viral/phage translation tables and single-pass co-prediction of bacterial and prophage/phage ORFs.

Attributes:

Name	Type	Description
`model`	`HMM`	Baked HMM model parameter container.
`genetic_code`	`GeneticCode`	Translation table defining start/stop codons and amino acid mappings.
`include_phage`	`bool`	If True, enables single-pass phage/prophage ORF co-prediction (sliding local GC adaptation, overlapping operon transitions, and small ORF recovery).
`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/serogen/genes/orf/predictor.py

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

    Executes Numba JIT-compiled FragGeneScan HMM decoding over single sequences,
    [`Sequences`][serogen.core.seq.seq.Sequences], or [`GenomeAssembly`][serogen.core.genome.GenomeAssembly]
    containers using a unified flat window table and Numba `prange` parallelism, returning
    batched [`Genes`][serogen.genes.models.Genes]. Supports bacterial and viral/phage translation
    tables and single-pass co-prediction of bacterial and prophage/phage ORFs.

    Attributes:
        model (HMM): Baked HMM model parameter container.
        genetic_code (GeneticCode): Translation table defining start/stop codons and amino acid mappings.
        include_phage (bool): If True, enables single-pass phage/prophage ORF co-prediction
            (sliding local GC adaptation, overlapping operon transitions, and small ORF recovery).
        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 | str = GeneticCode.bacterial,
        include_phage: bool = False,
        min_gene_len: int | None = None,
        local_gc_window: int = 4_000,
    ) -> None:
        r"""Initialize an [`OrfPredictor`][serogen.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 | str): Translation table (instance, NCBI table ID, or name).
                Defaults to `GeneticCode.bacterial` (Table 11).
            include_phage (bool): If True, enables single-pass bacterial + phage ORF co-prediction.
                Defaults to False.
            min_gene_len (int | None): Optional minimum ORF length threshold in base pairs. If None,
                defaults to 90 bp when `include_phage=True` or 210 bp when `include_phage=False`.
            local_gc_window (int): Sliding window size in base pairs for local GC bin adaptation
                when `include_phage=True`. Defaults to 4_000.
        """
        self.model = model if model is not None else get_hmm_model()
        self.window_size = window_size
        self.overlap = overlap
        self.genetic_code = (
            genetic_code if isinstance(genetic_code, GeneticCode) else GeneticCode.from_table(genetic_code)
        )
        self.include_phage = include_phage
        if min_gene_len is not None:
            self.min_gene_len = int(min_gene_len)
        else:
            self.min_gene_len = 90 if include_phage else 210
        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.include_phage,
            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.include_phage,
            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`][serogen.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/serogen/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`][serogen.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 (`serogen.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/serogen/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/serogen/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/serogen/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/serogen/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/serogen/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 (`Interval`).
`seq`	`Sequence \| None`	Coding nucleotide or protein sequence (`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/serogen/genes/models.py

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

    Attributes:
        interval (Interval): Genomic interval coordinate ([`Interval`][serogen.core.interval.Interval]).
        seq (Sequence | None): Coding nucleotide or protein sequence ([`Sequence`][serogen.core.seq.seq.Sequence]).
        contig_index (int): 0-based contig index offset corresponding to parent contig name.
        indels (IndelRecord): Scalar frameshift indel record ([`IndelRecord`][serogen.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 BatchedContainer.

Attributes:

Name	Type	Description
`intervals`	`Intervals`	Genomic interval coordinates for predicted ORFs (`Intervals`).
`seqs`	`Sequences`	In-frame coding nucleotide sequences oriented 5' to 3' (`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/serogen/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
    [`BatchedContainer`][serogen.core.collections.BatchedContainer].

    Attributes:
        intervals (Intervals): Genomic interval coordinates for predicted ORFs
            ([`Intervals`][serogen.core.interval.Intervals]).
        seqs (Sequences): In-frame coding nucleotide sequences oriented 5' to 3'
            ([`Sequences`][serogen.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 | str = GeneticCode.bacterial,
    ) -> None:
        r"""Initialize a [`Genes`][serogen.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_),
        )
        object.__setattr__(
            self,
            "genetic_code",
            genetic_code if isinstance(genetic_code, GeneticCode) else GeneticCode.from_table(genetic_code),
        )

    @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`][serogen.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 | str | None = None, to_stop: bool = True) -> Sequences:
        r"""Translate coding nucleotide sequences into protein sequences.

        Args:
            code (GeneticCode | int | 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
        resolved = (
            self.genetic_code
            if code is None
            else (code if isinstance(code, GeneticCode) else GeneticCode.from_table(code))
        )
        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`][serogen.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`][serogen.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.

Returns:

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

translate ¶

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

Translate coding nucleotide sequences into protein sequences.

Parameters:

Name	Type	Description	Default
`code`	`GeneticCode \| int \| 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/serogen/genes/models.py

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

    Args:
        code (GeneticCode | int | 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
    resolved = (
        self.genetic_code
        if code is None
        else (code if isinstance(code, GeneticCode) else GeneticCode.from_table(code))
    )
    return self.seqs.translate(code=resolved, to_stop=to_stop)

empty `classmethod` ¶

empty() -> Self

Create an empty Genes container.

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

@classmethod
def empty(cls) -> Self:
    r"""Create an empty [`Genes`][serogen.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/serogen/genes/models.py

@classmethod
def concat(cls, batches: Iterable[Self]) -> Self:
    r"""Concatenate multiple [`Genes`][serogen.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/serogen/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/serogen/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/serogen/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 (`serogen.genes.roles`)¶

roles ¶

Canonical biosynthetic gene roles, metadata specifications, and 4-tier regex classification.

Classes:

Name	Description
`RoleCategory`	Functional category of a biosynthetic gene role.
`RoleSpec`	Immutable metadata and classification specification for a canonical gene role.
`GeneRole`	Canonical biosynthetic gene roles (0..29) for surface polysaccharide loci.

RoleCategory ¶

Bases: StrEnum

Functional category of a biosynthetic gene role.

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

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

    TRANSPORT = "Transport"
    POLYMER_ASSEMBLY = "PolymerAssembly"
    MODIFIER = "Modifier"
    PRECURSOR_SYNTHESIS = "PrecursorSynthesis"
    MOBILE_ELEMENT = "MobileElement"
    BACKGROUND = "Background"

RoleSpec `dataclass` ¶

Immutable metadata and classification specification for a canonical gene role.

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

@dataclass(frozen=True, slots=True)
class RoleSpec:
    """Immutable metadata and classification specification for a canonical gene role."""

    name: str
    abbreviation: str
    category: RoleCategory
    symbol_regex: str
    desc_regex: str
    description: str
    aliases: tuple[str, ...] = ()
    trusted_prefixes: tuple[str, ...] = ()

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 SciPy CSR matrices.

Methods:

Name	Description
`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 string array of canonical role names.
`abbreviations_array`	Contiguous 1D NumPy string array of role abbreviations.
`categories_array`	Contiguous 1D NumPy string array of role categories.
`feature_matrix`	Return precomputed `(30, D)` float32 one-hot category feature matrix for embeddings.
`label`	Assign gene role indices for a single gene or batch of genes using the 4-tier strategy.

Attributes:

Name	Type	Description
`spec`	`RoleSpec`	Return the immutable `RoleSpec` metadata for this role.
`abbreviation`	`str`	Short display abbreviation (e.g. `'Wzx'`, `'GT'`).
`abbr`	`str`	Alias for `abbreviation`.
`category`	`RoleCategory`	Functional `RoleCategory` of this gene role.
`description`	`str`	Human-readable biochemical description.
`aliases`	`tuple[str, ...]`	Known gene symbol aliases for this role.
`trusted_prefixes`	`tuple[str, ...]`	Trusted canonical gene symbol prefixes for Tier-1 classification.
`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.

Source code in src/serogen/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 SciPy 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

    # ------------------------------------------------------------------------------------------------------------------
    # Instance Metadata Properties
    # ------------------------------------------------------------------------------------------------------------------
    @property
    def spec(self) -> RoleSpec:
        """Return the immutable `RoleSpec` metadata for this role."""
        return _ROLE_SPECS[self]

    @property
    def abbreviation(self) -> str:
        """Short display abbreviation (e.g. `'Wzx'`, `'GT'`)."""
        return _ROLE_SPECS[self].abbreviation

    @property
    def abbr(self) -> str:
        """Alias for `abbreviation`."""
        return _ROLE_SPECS[self].abbreviation

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

    @property
    def description(self) -> str:
        """Human-readable biochemical description."""
        return _ROLE_SPECS[self].description

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

    @property
    def trusted_prefixes(self) -> tuple[str, ...]:
        """Trusted canonical gene symbol prefixes for Tier-1 classification."""
        return _ROLE_SPECS[self].trusted_prefixes

    @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 _ROLE_SPECS[self].category == RoleCategory.MOBILE_ELEMENT

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

    # ------------------------------------------------------------------------------------------------------------------
    # Lookups & Vectorized Class Views
    # ------------------------------------------------------------------------------------------------------------------
    @classmethod
    def from_name(cls, key: str | bytes | int | GeneRole) -> GeneRole:
        """Resolve a role from an integer index, canonical name, abbreviation, or alias.

        Args:
            key: Role identifier (`GeneRole`, `int`, `str`, or `bytes`).

        Returns:
            GeneRole: Resolved canonical role enum member.

        Raises:
            KeyError: If a string/bytes key does not match any known role name, abbreviation, or alias.
            ValueError: If an integer index is out of range.
        """
        if isinstance(key, cls):
            return key
        if isinstance(key, (int, np.integer)):
            return cls(int(key))
        s = key.decode("utf-8", errors="replace") if isinstance(key, (bytes, bytearray)) else str(key)
        s_clean = s.strip()
        if s_clean in _LOOKUP_MAP:
            return _LOOKUP_MAP[s_clean]
        s_lower = s_clean.lower()
        if s_lower in _LOOKUP_MAP:
            return _LOOKUP_MAP[s_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.str_]]:
        """Contiguous 1D NumPy string array of canonical role names."""
        return _NAMES_ARRAY

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

    @classmethod
    def categories_array(cls) -> np.ndarray[tuple[int], np.dtype[np.str_]]:
        """Contiguous 1D NumPy string array of role categories."""
        return _CATEGORIES_ARRAY

    @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 _FEATURE_MATRIX

    # ------------------------------------------------------------------------------------------------------------------
    # Single-Pass 4-Tier Classification Engine
    # ------------------------------------------------------------------------------------------------------------------
    @classmethod
    def _classify_one(cls, symbol_clean: str, desc_clean: str) -> GeneRole:
        """Core 4-tier single-item classification engine."""
        # Tier 1: Canonical Trusted Core Symbol Prefixes
        if symbol_clean and symbol_clean.startswith(_TRUSTED_PREFIXES):
            for p in _TRUSTED_PREFIXES:
                if symbol_clean.startswith(p):
                    return _TRUSTED_MAP[p]

        # Tier 2: Informative Product Description Regex
        if desc_clean and _DESC_PATTERN is not None:
            match = _DESC_PATTERN.search(desc_clean)
            if match and match.lastgroup:
                return _GROUP_TO_ROLE.get(match.lastgroup, cls.UNKNOWN)

        # Tier 3: Symbol Regex & Aliases Fallback
        if symbol_clean:
            if symbol_clean in _ALIAS_MAP:
                return _ALIAS_MAP[symbol_clean]
            if _SYMBOL_PATTERN is not None:
                match = _SYMBOL_PATTERN.search(symbol_clean)
                if match and match.lastgroup:
                    return _GROUP_TO_ROLE.get(match.lastgroup, cls.UNKNOWN)
            stripped = _LOCUS_SUFFIX_PATTERN.sub("", symbol_clean)
            if stripped != symbol_clean and stripped:
                if stripped in _ALIAS_MAP:
                    return _ALIAS_MAP[stripped]
                if _SYMBOL_PATTERN is not None:
                    match = _SYMBOL_PATTERN.search(stripped)
                    if match and match.lastgroup:
                        return _GROUP_TO_ROLE.get(match.lastgroup, cls.UNKNOWN)

        # Tier 4: Fallback for empty symbol when description starts with a candidate gene identifier
        if not symbol_clean and desc_clean:
            candidate = desc_clean.split()[0].rstrip("*")
            if candidate.startswith(_TRUSTED_PREFIXES):
                for p in _TRUSTED_PREFIXES:
                    if candidate.startswith(p):
                        return _TRUSTED_MAP[p]
            if candidate in _ALIAS_MAP:
                return _ALIAS_MAP[candidate]
            if _SYMBOL_PATTERN is not None:
                match = _SYMBOL_PATTERN.search(candidate)
                if match and match.lastgroup:
                    return _GROUP_TO_ROLE.get(match.lastgroup, cls.UNKNOWN)
                stripped = _LOCUS_SUFFIX_PATTERN.sub("", candidate)
                if stripped != candidate and stripped:
                    if stripped in _ALIAS_MAP:
                        return _ALIAS_MAP[stripped]
                    m_strip = _SYMBOL_PATTERN.search(stripped)
                    if m_strip and m_strip.lastgroup:
                        return _GROUP_TO_ROLE.get(m_strip.lastgroup, cls.UNKNOWN)

        return cls.UNKNOWN

    @classmethod
    def label(
        cls,
        symbol: str | bytes | Sequence[str | bytes] | np.ndarray | None = None,
        description: str | bytes | Sequence[str | bytes] | np.ndarray | None = None,
    ) -> GeneRole | np.ndarray[tuple[int], np.dtype[np.int8]]:
        """Assign gene role indices for a single gene or batch of genes using the 4-tier strategy.

        1. Tier 1: Canonical trusted core symbol prefixes.
        2. Tier 2: Informative product description regex.
        3. Tier 3: Symbol regex and aliases fallback (with locus suffix stripping).
        4. Tier 4: Gene identifier extraction from product description fallback.

        Returns:
            `GeneRole` if inputs are scalar, or `np.ndarray[int8]` if inputs are a sequence/array.
        """
        if isinstance(symbol, (str, bytes, bytearray)) or (
            symbol is None and isinstance(description, (str, bytes, bytearray))
        ):
            s = (
                (symbol.decode("utf-8", errors="replace") if isinstance(symbol, (bytes, bytearray)) else symbol)
                .strip()
                .lower()
                if symbol
                else ""
            )
            d = (
                (
                    description.decode("utf-8", errors="replace")
                    if isinstance(description, (bytes, bytearray))
                    else str(description)
                )
                .strip()
                .lower()
                if description
                else ""
            )
            return cls._classify_one(s, d)

        if symbol is not None:
            sym_arr = symbol if isinstance(symbol, (list, tuple, np.ndarray)) else list(symbol)
            n_items = len(sym_arr)
        elif description is not None:
            desc_arr = description if isinstance(description, (list, tuple, np.ndarray)) else list(description)
            n_items = len(desc_arr)
            sym_arr = ()
        else:
            return cls.UNKNOWN

        if n_items == 0:
            return np.empty(0, dtype=np.int8)

        out = np.empty(n_items, dtype=np.int8)
        classify = cls._classify_one
        has_sym = len(sym_arr) > 0

        if description is None:
            for i in range(n_items):
                s_raw = sym_arr[i]
                if s_raw:
                    s = (
                        (
                            s_raw.decode("utf-8", errors="replace")
                            if isinstance(s_raw, (bytes, bytearray))
                            else str(s_raw)
                        )
                        .strip()
                        .lower()
                    )
                else:
                    s = ""
                out[i] = classify(s, "")
        else:
            desc_arr = description if isinstance(description, (list, tuple, np.ndarray)) else list(description)
            if len(desc_arr) != n_items:
                raise ValueError(f"descriptions length {len(desc_arr)} does not match symbols length {n_items}")
            for i in range(n_items):
                s_raw = sym_arr[i] if has_sym else None
                d_raw = desc_arr[i]
                s = (
                    (s_raw.decode("utf-8", errors="replace") if isinstance(s_raw, (bytes, bytearray)) else str(s_raw))
                    .strip()
                    .lower()
                    if s_raw
                    else ""
                )
                d = (
                    (d_raw.decode("utf-8", errors="replace") if isinstance(d_raw, (bytes, bytearray)) else str(d_raw))
                    .strip()
                    .lower()
                    if d_raw
                    else ""
                )
                out[i] = classify(s, d)

        return out

spec `property` ¶

spec: RoleSpec

Return the immutable RoleSpec metadata for this role.

abbreviation `property` ¶

abbreviation: str

Short display abbreviation (e.g. 'Wzx', 'GT').

abbr `property` ¶

abbr: str

Alias for abbreviation.

category `property` ¶

category: RoleCategory

Functional RoleCategory of this gene role.

description `property` ¶

description: str

Human-readable biochemical description.

aliases `property` ¶

aliases: tuple[str, ...]

Known gene symbol aliases for this role.

trusted_prefixes `property` ¶

trusted_prefixes: tuple[str, ...]

Trusted canonical gene symbol prefixes for Tier-1 classification.

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.

from_name `classmethod` ¶

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

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

Parameters:

Name	Type	Description	Default
`key`	`str \| bytes \| int \| GeneRole`	Role identifier (`GeneRole`, `int`, `str`, or `bytes`).	required

Returns:

Name	Type	Description
`GeneRole`	`GeneRole`	Resolved canonical role enum member.

Raises:

Type	Description
`KeyError`	If a string/bytes key does not match any known role name, abbreviation, or alias.
`ValueError`	If an integer index is out of range.

Source code in src/serogen/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.

    Args:
        key: Role identifier (`GeneRole`, `int`, `str`, or `bytes`).

    Returns:
        GeneRole: Resolved canonical role enum member.

    Raises:
        KeyError: If a string/bytes key does not match any known role name, abbreviation, or alias.
        ValueError: If an integer index is out of range.
    """
    if isinstance(key, cls):
        return key
    if isinstance(key, (int, np.integer)):
        return cls(int(key))
    s = key.decode("utf-8", errors="replace") if isinstance(key, (bytes, bytearray)) else str(key)
    s_clean = s.strip()
    if s_clean in _LOOKUP_MAP:
        return _LOOKUP_MAP[s_clean]
    s_lower = s_clean.lower()
    if s_lower in _LOOKUP_MAP:
        return _LOOKUP_MAP[s_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/serogen/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[str_]]

Contiguous 1D NumPy string array of canonical role names.

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

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

abbreviations_array `classmethod` ¶

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

Contiguous 1D NumPy string array of role abbreviations.

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

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

categories_array `classmethod` ¶

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

Contiguous 1D NumPy string array of role categories.

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

@classmethod
def categories_array(cls) -> np.ndarray[tuple[int], np.dtype[np.str_]]:
    """Contiguous 1D NumPy string array of role categories."""
    return _CATEGORIES_ARRAY

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/serogen/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 _FEATURE_MATRIX

label `classmethod` ¶

label(symbol: str | bytes | Sequence[str | bytes] | ndarray | None = None, description: str | bytes | Sequence[str | bytes] | ndarray | None = None) -> GeneRole | ndarray[tuple[int], dtype[int8]]

Assign gene role indices for a single gene or batch of genes using the 4-tier strategy.

Tier 1: Canonical trusted core symbol prefixes.
Tier 2: Informative product description regex.
Tier 3: Symbol regex and aliases fallback (with locus suffix stripping).
Tier 4: Gene identifier extraction from product description fallback.

Returns:

Type	Description
`GeneRole \| ndarray[tuple[int], dtype[int8]]`	`GeneRole` if inputs are scalar, or `np.ndarray[int8]` if inputs are a sequence/array.

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

@classmethod
def label(
    cls,
    symbol: str | bytes | Sequence[str | bytes] | np.ndarray | None = None,
    description: str | bytes | Sequence[str | bytes] | np.ndarray | None = None,
) -> GeneRole | np.ndarray[tuple[int], np.dtype[np.int8]]:
    """Assign gene role indices for a single gene or batch of genes using the 4-tier strategy.

    1. Tier 1: Canonical trusted core symbol prefixes.
    2. Tier 2: Informative product description regex.
    3. Tier 3: Symbol regex and aliases fallback (with locus suffix stripping).
    4. Tier 4: Gene identifier extraction from product description fallback.

    Returns:
        `GeneRole` if inputs are scalar, or `np.ndarray[int8]` if inputs are a sequence/array.
    """
    if isinstance(symbol, (str, bytes, bytearray)) or (
        symbol is None and isinstance(description, (str, bytes, bytearray))
    ):
        s = (
            (symbol.decode("utf-8", errors="replace") if isinstance(symbol, (bytes, bytearray)) else symbol)
            .strip()
            .lower()
            if symbol
            else ""
        )
        d = (
            (
                description.decode("utf-8", errors="replace")
                if isinstance(description, (bytes, bytearray))
                else str(description)
            )
            .strip()
            .lower()
            if description
            else ""
        )
        return cls._classify_one(s, d)

    if symbol is not None:
        sym_arr = symbol if isinstance(symbol, (list, tuple, np.ndarray)) else list(symbol)
        n_items = len(sym_arr)
    elif description is not None:
        desc_arr = description if isinstance(description, (list, tuple, np.ndarray)) else list(description)
        n_items = len(desc_arr)
        sym_arr = ()
    else:
        return cls.UNKNOWN

    if n_items == 0:
        return np.empty(0, dtype=np.int8)

    out = np.empty(n_items, dtype=np.int8)
    classify = cls._classify_one
    has_sym = len(sym_arr) > 0

    if description is None:
        for i in range(n_items):
            s_raw = sym_arr[i]
            if s_raw:
                s = (
                    (
                        s_raw.decode("utf-8", errors="replace")
                        if isinstance(s_raw, (bytes, bytearray))
                        else str(s_raw)
                    )
                    .strip()
                    .lower()
                )
            else:
                s = ""
            out[i] = classify(s, "")
    else:
        desc_arr = description if isinstance(description, (list, tuple, np.ndarray)) else list(description)
        if len(desc_arr) != n_items:
            raise ValueError(f"descriptions length {len(desc_arr)} does not match symbols length {n_items}")
        for i in range(n_items):
            s_raw = sym_arr[i] if has_sym else None
            d_raw = desc_arr[i]
            s = (
                (s_raw.decode("utf-8", errors="replace") if isinstance(s_raw, (bytes, bytearray)) else str(s_raw))
                .strip()
                .lower()
                if s_raw
                else ""
            )
            d = (
                (d_raw.decode("utf-8", errors="replace") if isinstance(d_raw, (bytes, bytearray)) else str(d_raw))
                .strip()
                .lower()
                if d_raw
                else ""
            )
            out[i] = classify(s, d)

    return out

1D CNN Role Classifier (`serogen.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, 9, 13), ReLU activation, dual pooling (global max pooling + global average pooling) concatenated with normalized log-length and amino acid composition, and a dense classification layer. 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`	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/serogen/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, 9, 13), ReLU activation, dual pooling
    (global max pooling + global average pooling) concatenated with normalized log-length and
    amino acid composition, and a dense classification layer. 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]]
    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 + global composition feature vector."""
        if getattr(self, "dense_w_", None) is not None:
            return int(self.dense_w_.shape[0])
        return 2 * int(self.num_filters) + 2 + int(self.vocab_size)

    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,
        *,
        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 int(np.min(y_i64)) >= -128 and int(np.max(y_i64)) <= 127:
            y_arr = y_i64.astype(np.int8)
            self.classes_ = np.unique(y_arr).astype(np.int8)
        else:
            y_arr = y_i64
            self.classes_ = np.unique(y_arr).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.searchsorted(self.classes_, y_arr).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, weights=sw_arr, minlength=n_classes).astype(np.float32)
            counts_safe = np.maximum(counts, 1.0)
            weights = float(np.sum(sw_arr)) / (float(n_classes) * counts_safe)
            class_weights = (weights / np.mean(weights)).astype(np.float32)
        else:
            class_weights = np.ones(n_classes, 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 + 2 + self.vocab_size
        cnn_feat_dim = 2 * self.num_filters
        self.dense_w_ = np.zeros((feat_dim, n_classes), dtype=np.float32)
        self.dense_w_[:cnn_feat_dim, :] = rng.normal(
            0.0,
            np.sqrt(2.0 / cnn_feat_dim),
            (cnn_feat_dim, n_classes),
        ).astype(np.float32)
        self.dense_b_ = np.zeros(n_classes, 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_,
            loss_hist,
        ) = _train_adam(
            x_flat,
            offsets,
            y_indexed,
            class_weights,
            sw_arr,
            self.embedding_,
            self.conv_w_,
            self.conv_b_,
            self.dense_w_,
            self.dense_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)
        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_, 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 == "embed_mean":
            aa_start = 2 * nf + 2
            aa_freqs = h[:, aa_start : aa_start + self.vocab_size] * np.float32(4.0)
            return np.ascontiguousarray(aa_freqs @ self.embedding_)
        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(
        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,
        )
        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_,
            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)
            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("serogen.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 + global composition feature vector.

fit ¶

fit(X: Genes | Sequences | Sequence[Any] | tuple[ndarray, ndarray], y: Sequence[int | GeneRole] | ndarray, sample_weight: Sequence[float] | ndarray | None = None, *, 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/serogen/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,
    *,
    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 int(np.min(y_i64)) >= -128 and int(np.max(y_i64)) <= 127:
        y_arr = y_i64.astype(np.int8)
        self.classes_ = np.unique(y_arr).astype(np.int8)
    else:
        y_arr = y_i64
        self.classes_ = np.unique(y_arr).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.searchsorted(self.classes_, y_arr).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, weights=sw_arr, minlength=n_classes).astype(np.float32)
        counts_safe = np.maximum(counts, 1.0)
        weights = float(np.sum(sw_arr)) / (float(n_classes) * counts_safe)
        class_weights = (weights / np.mean(weights)).astype(np.float32)
    else:
        class_weights = np.ones(n_classes, 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 + 2 + self.vocab_size
    cnn_feat_dim = 2 * self.num_filters
    self.dense_w_ = np.zeros((feat_dim, n_classes), dtype=np.float32)
    self.dense_w_[:cnn_feat_dim, :] = rng.normal(
        0.0,
        np.sqrt(2.0 / cnn_feat_dim),
        (cnn_feat_dim, n_classes),
    ).astype(np.float32)
    self.dense_b_ = np.zeros(n_classes, 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_,
        loss_hist,
    ) = _train_adam(
        x_flat,
        offsets,
        y_indexed,
        class_weights,
        sw_arr,
        self.embedding_,
        self.conv_w_,
        self.conv_b_,
        self.dense_w_,
        self.dense_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/serogen/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)
    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_, 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 == "embed_mean":
        aa_start = 2 * nf + 2
        aa_freqs = h[:, aa_start : aa_start + self.vocab_size] * np.float32(4.0)
        return np.ascontiguousarray(aa_freqs @ self.embedding_)
    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/serogen/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/serogen/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 ¶

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/serogen/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/serogen/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/serogen/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/serogen/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,
    )
    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_,
        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/serogen/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)
        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/serogen/genes/cnn.py

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

serogen.genes¶

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

predictor ¶

OrfPredictor ¶

predict ¶

Gene Containers & HMM Parameters (serogen.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 (serogen.genes.roles)¶

roles ¶

RoleCategory ¶

RoleSpec dataclass ¶

GeneRole ¶

spec property ¶

abbreviation property ¶

abbr property ¶

category property ¶

description property ¶

aliases property ¶

trusted_prefixes property ¶

bitmask property ¶

is_mobile_element property ¶

is_biosynthetic property ¶

from_name classmethod ¶

resolve classmethod ¶

names_array classmethod ¶

abbreviations_array classmethod ¶

categories_array classmethod ¶

feature_matrix classmethod ¶

label classmethod ¶

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

cnn ¶

GeneRoleClassifier ¶

is_fitted property ¶

num_classes property ¶

feature_dim property ¶

fit ¶

transform ¶

decision_function ¶

predict_proba ¶

predict ¶

score ¶

top_k_accuracy ¶

save ¶

load classmethod ¶

pretrained cached classmethod ¶

Comments

`serogen.genes`¶

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

Gene Containers & HMM Parameters (`serogen.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 (`serogen.genes.roles`)¶

RoleSpec `dataclass` ¶

spec `property` ¶

abbreviation `property` ¶

abbr `property` ¶

category `property` ¶

description `property` ¶

aliases `property` ¶

trusted_prefixes `property` ¶

bitmask `property` ¶

is_mobile_element `property` ¶

is_biosynthetic `property` ¶

from_name `classmethod` ¶

resolve `classmethod` ¶

names_array `classmethod` ¶

abbreviations_array `classmethod` ¶

categories_array `classmethod` ¶

feature_matrix `classmethod` ¶

label `classmethod` ¶

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

is_fitted `property` ¶

num_classes `property` ¶

feature_dim `property` ¶

load `classmethod` ¶

pretrained `cached` `classmethod` ¶