API Reference

siRNAforge - Comprehensive siRNA design toolkit for gene silencing. Comprehensive gene silencing design and analysis.

This module exposes package metadata (author/email/version) in a single place. The version is resolved from installed package metadata (importlib.metadata). When running from a source checkout (not installed), it falls back to reading pyproject.toml if available, otherwise uses a conservative placeholder.

Command Line Interface

Modern CLI for siRNAforge using Typer and Rich.

sirnaforge.cli.patched_init(self, *args, **kwargs)

Force simplified terminal capabilities for deterministic CI output.

Parameters:

• self (Console)

• args (Any)

• kwargs (Any)

Return type:

None

sirnaforge.cli.filter_transcripts(transcripts, include_types=None, exclude_types=None, canonical_only=False)

Filter transcript records by type and canonical status.

Parameters:

• transcripts – Iterable of transcript-like objects that expose transcript_type and is_canonical attributes.

• include_types – Optional iterable of transcript types to keep.

• exclude_types – Optional iterable of transcript types to drop.

• canonical_only – When True, keep only canonical isoforms.

Returns:

A list of transcripts that match the requested filters.

sirnaforge.cli.extract_canonical_transcripts(transcripts, gene_name, output_dir=None)

Write canonical isoforms to a separate FASTA file.

Parameters:

• transcripts – Iterable of transcript-like objects (must expose is_canonical and sequence attributes used by the underlying save routine).

• gene_name – Name used to derive the output FASTA filename.

• output_dir – Directory to write the FASTA file into (defaults to CWD).

Returns:

A tuple of (canonical_fasta_path, count) where the path is None when no canonical isoforms are available.

sirnaforge.cli.search(query=<typer.models.ArgumentInfo object>, output=<typer.models.OptionInfo object>, database=<typer.models.OptionInfo object>, all_databases=<typer.models.OptionInfo object>, fallback=<typer.models.OptionInfo object>, no_sequence=<typer.models.OptionInfo object>, canonical_only=<typer.models.OptionInfo object>, extract_canonical=<typer.models.OptionInfo object>, transcript_types=<typer.models.OptionInfo object>, exclude_types=<typer.models.OptionInfo object>, verbose=<typer.models.OptionInfo object>)

Search transcript references and optionally fetch sequences.

This command queries Ensembl/RefSeq/Gencode (depending on flags) for a gene or transcript identifier. When sequences are fetched, it writes them to a FASTA file and can optionally also emit a canonical-only FASTA.

Parameters:

• query (str)

• output (Path)

• database (str)

• all_databases (bool)

• fallback (bool)

• no_sequence (bool)

• canonical_only (bool)

• extract_canonical (bool)

• transcript_types (str)

• exclude_types (str)

• verbose (bool)

Return type:

None

sirnaforge.cli.workflow(gene_query=<typer.models.ArgumentInfo object>, input_fasta=<typer.models.OptionInfo object>, output_dir=<typer.models.OptionInfo object>, database=<typer.models.OptionInfo object>, design_mode=<typer.models.OptionInfo object>, top_n_candidates=<typer.models.OptionInfo object>, species=<typer.models.OptionInfo object>, mirna_db=<typer.models.OptionInfo object>, mirna_species=<typer.models.OptionInfo object>, transcriptome_fasta=<typer.models.OptionInfo object>, transcriptome_filter=<typer.models.OptionInfo object>, offtarget_indices=<typer.models.OptionInfo object>, gc_min=<typer.models.OptionInfo object>, gc_max=<typer.models.OptionInfo object>, sirna_length=<typer.models.OptionInfo object>, modification_pattern=<typer.models.OptionInfo object>, overhang=<typer.models.OptionInfo object>, skip_off_targets=<typer.models.OptionInfo object>, snp=<typer.models.OptionInfo object>, snp_file=<typer.models.OptionInfo object>, variant_mode=<typer.models.OptionInfo object>, min_af=<typer.models.OptionInfo object>, clinvar_filter_levels=<typer.models.OptionInfo object>, variant_assembly=<typer.models.OptionInfo object>, verbose=<typer.models.OptionInfo object>, log_file=<typer.models.OptionInfo object>, nextflow_docker_image=<typer.models.OptionInfo object>, json_summary=<typer.models.OptionInfo object>)

Run the end-to-end workflow: transcripts → siRNA design → off-target.

This is the main orchestration command. It resolves transcriptome and miRNA reference policies, designs candidates, and then runs off-target analysis on the selected top candidates.

Parameters:

• gene_query (str)

• input_fasta (str | None)

• output_dir (Path)

• database (str)

• design_mode (str)

• top_n_candidates (int)

• species (str)

• mirna_db (str)

• mirna_species (str | None)

• transcriptome_fasta (str | None)

• transcriptome_filter (str | None)

• offtarget_indices (str | None)

• gc_min (float)

• gc_max (float)

• sirna_length (int)

• modification_pattern (str)

• overhang (str)

• skip_off_targets (bool)

• snp (list[str])

• snp_file (Path | None)

• variant_mode (VariantMode)

• min_af (float)

• clinvar_filter_levels (str)

• variant_assembly (str)

• verbose (bool)

• log_file (Path | None)

• nextflow_docker_image (str | None)

• json_summary (bool)

Return type:

None

sirnaforge.cli.offtarget(input_candidates_fasta=<typer.models.OptionInfo object>, output_dir=<typer.models.OptionInfo object>, species=<typer.models.OptionInfo object>, mirna_db=<typer.models.OptionInfo object>, mirna_species=<typer.models.OptionInfo object>, transcriptome_fasta=<typer.models.OptionInfo object>, transcriptome_filter=<typer.models.OptionInfo object>, offtarget_indices=<typer.models.OptionInfo object>, verbose=<typer.models.OptionInfo object>, log_file=<typer.models.OptionInfo object>, nextflow_docker_image=<typer.models.OptionInfo object>)

Run off-target analysis on pre-designed siRNA candidates.

This command accepts a FASTA file containing pre-designed siRNA guide sequences of any length and runs comprehensive off-target analysis including: - Transcriptome alignment (BWA-MEM2) - miRNA seed match analysis - Off-target hit classification and scoring

The embedded Nextflow pipeline is used for parallel processing across species.

Notes

• --species drives transcriptome fetching and miRNA lookup.

• --offtarget-indices can override the indices used for alignment using species:/abs/path/index_prefix entries.

Parameters:

• input_candidates_fasta (Path)

• output_dir (Path)

• species (str)

• mirna_db (str)

• mirna_species (str | None)

• transcriptome_fasta (str | None)

• transcriptome_filter (str | None)

• offtarget_indices (str | None)

• verbose (bool)

• log_file (Path | None)

• nextflow_docker_image (str | None)

Return type:

None

sirnaforge.cli.design(input_file=<typer.models.ArgumentInfo object>, output=<typer.models.OptionInfo object>, design_mode=<typer.models.OptionInfo object>, length=<typer.models.OptionInfo object>, top_n=<typer.models.OptionInfo object>, gc_min=<typer.models.OptionInfo object>, gc_max=<typer.models.OptionInfo object>, max_poly_runs=<typer.models.OptionInfo object>, genome_index=<typer.models.OptionInfo object>, snp_file=<typer.models.OptionInfo object>, skip_structure=<typer.models.OptionInfo object>, skip_off_targets=<typer.models.OptionInfo object>, modification_pattern=<typer.models.OptionInfo object>, overhang=<typer.models.OptionInfo object>, verbose=<typer.models.OptionInfo object>)

Design siRNA candidates from a transcript FASTA file.

Outputs a TSV/CSV-like table of candidates, optionally including secondary structure scoring, off-target checks, and chemical modification annotations.

Parameters:

• input_file (Path)

• output (Path)

• design_mode (str)

• length (int)

• top_n (int)

• gc_min (float)

• gc_max (float)

• max_poly_runs (int)

• genome_index (Path | None)

• snp_file (Path | None)

• skip_structure (bool)

• skip_off_targets (bool)

• modification_pattern (str)

• overhang (str)

• verbose (bool)

Return type:

None

sirnaforge.cli.validate(input_file=<typer.models.ArgumentInfo object>)

Validate a FASTA file and report basic statistics.

This performs lightweight validation (parseable FASTA, presence of sequences, and common issues like short/ambiguous sequences).

Parameters:

input_file (Path)

Return type:

None

sirnaforge.cli.version()

Show CLI version and author information.

Return type:

None

sirnaforge.cli.config()

Print the default design parameter values.

Return type:

None

sirnaforge.cli.cache(clear=<typer.models.OptionInfo object>, clear_mirna=<typer.models.OptionInfo object>, clear_transcriptome=<typer.models.OptionInfo object>, dry_run=<typer.models.OptionInfo object>, info=<typer.models.OptionInfo object>)

Inspect and clear the unified reference cache.

This command can display cache statistics and/or delete cached assets for miRNA databases and transcriptomes.

Parameters:

• clear (bool)

• clear_mirna (bool)

• clear_transcriptome (bool)

• dry_run (bool)

• info (bool)

Return type:

None

exception sirnaforge.cli.SequencesShowError

Bases: RuntimeError

Raised when sequence display/formatting input is invalid.

sirnaforge.cli.sequences_show(input_file=<typer.models.ArgumentInfo object>, sequence_id=<typer.models.OptionInfo object>, format=<typer.models.OptionInfo object>)

Show sequences from a FASTA file in table, JSON, or FASTA format.

Use --id to select a single record. --format controls output: table (default), json (header metadata only), or fasta.

Parameters:

• input_file (Path)

• sequence_id (str | None)

• format (str)

Return type:

None

sirnaforge.cli.sequences_annotate(input_fasta=<typer.models.ArgumentInfo object>, metadata_json=<typer.models.ArgumentInfo object>, output=<typer.models.OptionInfo object>, verbose=<typer.models.OptionInfo object>)

Merge metadata from a JSON file into FASTA headers.

The JSON is expected to conform to the project metadata schema used by the modification/annotation utilities.

Parameters:

• input_fasta (Path)

• metadata_json (Path)

• output (Path | None)

• verbose (bool)

Return type:

None

Core Modules

Design Engine

Core siRNA design algorithms and functionality.

class sirnaforge.core.design.SiRNADesigner(parameters)

Bases: object

Main siRNA design engine following the algorithm specification.

Parameters:

parameters (DesignParameters)

__init__(parameters)

Initialize designer with given parameters.

Parameters:

parameters (DesignParameters)

design_from_file(input_file)

Design siRNAs from input FASTA file.

Parameters:

input_file (str)

Return type:

DesignResult

design_from_sequence(sequence, transcript_id='seq1')

Design siRNAs from a single sequence.

Parameters:

• sequence (str)

• transcript_id (str)

Return type:

DesignResult

class sirnaforge.core.design.MiRNADesigner(parameters)

Bases: SiRNADesigner

miRNA-biogenesis-aware siRNA designer with specialized scoring.

Extends SiRNADesigner with scoring rules optimized for miRNA-like processing: - Argonaute selection preferences (pos1 A/U, mismatch at pos1) - 3’ supplementary pairing analysis (positions 13-16) - Conservative thermodynamic thresholds - Seed region quality assessment

Parameters:

parameters (DesignParameters)

__init__(parameters)

Initialize miRNA designer with miRNA-specific config validation.

Parameters:

parameters (DesignParameters)

Thermodynamics Analysis

Thermodynamic calculations for siRNA design using ViennaRNA.

class sirnaforge.core.thermodynamics.ThermodynamicCalculator(temperature=37.0)

Bases: object

Calculate thermodynamic properties for siRNA candidates using ViennaRNA.

Parameters:

temperature (float)

__init__(temperature=37.0)

Initialize thermodynamic calculator.

Parameters:

temperature (float) – Temperature in Celsius for calculations

calculate_duplex_stability(guide, passenger)

Calculate duplex stability (deltaG) using ViennaRNA.

Parameters:

• guide (str)

• passenger (str)

Return type:

float

calculate_asymmetry_score(candidate)

Calculate thermodynamic asymmetry score using ViennaRNA.

Return type:

tuple[float, float, float]

Returns:

Tuple of (5’ end stability, 3’ end stability, asymmetry score)

Parameters:

candidate (SiRNACandidate)

calculate_target_accessibility(target_sequence, start_pos, sirna_length)

Calculate target site accessibility using ViennaRNA.

Parameters:

• target_sequence (str) – Full target mRNA sequence

• start_pos (int) – Start position of siRNA target site (0-based)

• sirna_length (int) – Length of siRNA

Return type:

tuple[float, float]

Returns:

Tuple of (average_unpaired_probability, mfe)

calculate_melting_temperature(guide, passenger)

Calculate melting temperature using ViennaRNA thermodynamics.

Parameters:

• guide (str)

• passenger (str)

Return type:

float

is_thermodynamically_favorable(candidate, threshold=0.5)

Check if candidate meets thermodynamic asymmetry threshold.

Parameters:

• candidate (SiRNACandidate)

• threshold (float)

Return type:

bool

calculate_secondary_structure(sequence)

Calculate secondary structure for a sequence.

Return type:

tuple[str, float, float]

Returns:

Tuple of (structure, mfe, paired_fraction)

Parameters:

sequence (str)

Off-Target Prediction

Off-target analysis for siRNA design.

This module provides comprehensive off-target analysis functionality for siRNA design, including both miRNA seed match analysis and transcriptome off-target detection. Uses BWA-MEM2 for both short and long sequence alignments. Optimized for both standalone use and parallelized Nextflow workflows.

class sirnaforge.core.off_target.BwaAnalyzer(index_prefix, mode='transcriptome', seed_length=12, min_score=15, max_hits=10000, seed_start=2, seed_end=8)

Bases: object

BWA-MEM2 based analyzer for both transcriptome and miRNA seed off-target search.

Parameters:

• index_prefix (str | Path)

• mode (str)

• seed_length (int)

• min_score (int)

• max_hits (int)

• seed_start (int)

• seed_end (int)

__init__(index_prefix, mode='transcriptome', seed_length=12, min_score=15, max_hits=10000, seed_start=2, seed_end=8)

Initialize BWA-MEM2 analyzer.

Parameters:

• index_prefix (str | Path) – Path to BWA index

• mode (str) – Analysis mode - “transcriptome” for long targets, “mirna_seed” for short targets

• seed_length (int) – BWA seed length parameter

• min_score (int) – Minimum alignment score

• max_hits (int) – Maximum hits to return

• seed_start (int) – Seed region start (1-based)

• seed_end (int) – Seed region end (1-based)

analyze_sequences(sequences)

Run BWA-MEM2 analysis on sequences.

Parameters:

sequences (dict[str, str]) – Dictionary of sequence name -> sequence

Return type:

list[dict[str, Any]]

Returns:

List of alignment dictionaries

class sirnaforge.core.off_target.OffTargetAnalysisManager(species, transcriptome_path=None, mirna_path=None, transcriptome_index=None, mirna_index=None)

Bases: object

Manager class for comprehensive off-target analysis using BWA-MEM2.

Parameters:

• species (str)

• transcriptome_path (str | Path | None)

• mirna_path (str | Path | None)

• transcriptome_index (str | Path | None)

• mirna_index (str | Path | None)

__init__(species, transcriptome_path=None, mirna_path=None, transcriptome_index=None, mirna_index=None)

Initialize the off-target analysis manager.

Parameters:

• species (str)

• transcriptome_path (str | Path | None)

• mirna_path (str | Path | None)

• transcriptome_index (str | Path | None)

• mirna_index (str | Path | None)

analyze_mirna_off_targets(sequences, output_prefix)

Analyze miRNA off-targets using BWA-MEM2 in miRNA seed mode.

Parameters:

• sequences (dict[str, str] | str | Path)

• output_prefix (str | Path)

Return type:

tuple[Path, Path]

analyze_transcriptome_off_targets(sequences, output_prefix)

Analyze transcriptome off-targets using BWA-MEM2 in transcriptome mode.

Parameters:

• sequences (dict[str, str] | str | Path)

• output_prefix (str | Path)

Return type:

tuple[Path, Path]

analyze_sirna_candidate(candidate)

Analyze a single siRNA candidate for off-targets.

Parameters:

candidate (SiRNACandidate)

Return type:

dict[str, Any]

sirnaforge.core.off_target.create_temp_fasta(sequences)

Create temporary FASTA file from sequences.

Parameters:

sequences (dict[str, str])

Return type:

str

sirnaforge.core.off_target.validate_and_write_sequences(input_file, output_file, expected_length=21)

Validate siRNA sequences and write valid ones to output file.

Parameters:

• input_file (str)

• output_file (str)

• expected_length (int)

Return type:

tuple[int, int, list[str]]

sirnaforge.core.off_target.build_bwa_index(fasta_file, index_prefix)

Build BWA-MEM2 index for both transcriptome and miRNA off-target analysis.

Parameters:

• fasta_file (str | Path)

• index_prefix (str | Path)

Return type:

Path

sirnaforge.core.off_target.validate_sirna_sequences(sequences, expected_length=21)

Validate siRNA sequences using existing FastaUtils.

Parameters:

• sequences (dict[str, str])

• expected_length (int)

Return type:

tuple[dict[str, str], dict[str, str], list[str]]

sirnaforge.core.off_target.parse_fasta_file(fasta_file)

Parse FASTA file using existing FastaUtils.

Parameters:

fasta_file (str | Path)

Return type:

dict[str, str]

sirnaforge.core.off_target.write_fasta_file(sequences, output_file)

Write sequences to FASTA file using existing FastaUtils.

Parameters:

• sequences (dict[str, str])

• output_file (str)

Return type:

None

sirnaforge.core.off_target.check_tool_availability(tool)

Check if external tool is available.

Parameters:

tool (str)

Return type:

bool

sirnaforge.core.off_target.validate_index_files(index_prefix, tool='bwa')

Validate that index files exist for given tool.

Parameters:

• index_prefix (str | Path)

• tool (str)

Return type:

bool

sirnaforge.core.off_target.run_bwa_alignment_analysis(candidates_file, index_prefix, species, output_dir, max_hits=10000, bwa_k=12, bwa_T=15, seed_start=2, seed_end=8)

Run BWA-MEM2 alignment analysis for candidate sequences using Pydantic models.

This is the main function called by OFFTARGET_ANALYSIS Nextflow module.

Parameters:

• candidates_file (str | Path) – Path to FASTA file with candidate sequences

• index_prefix (str | Path) – Path to BWA-MEM2 index prefix

• species (str) – Species identifier

• output_dir (str | Path) – Directory to write results

• max_hits (int) – Maximum hits to report per candidate

• bwa_k (int) – BWA seed length parameter

• bwa_T (int) – BWA minimum score threshold

• seed_start (int) – Seed region start position (1-based)

• seed_end (int) – Seed region end position (1-based)

Return type:

Path

Returns:

Path to output directory containing results

sirnaforge.core.off_target.aggregate_offtarget_results(results_dir, output_dir, genome_species)

Aggregate transcriptome off-target analysis results using Pandera.

Uses pandas + Pandera for efficient bulk reading and validation instead of manual line-by-line parsing with Pydantic models.

NOTE: This function ONLY aggregates genome/transcriptome hits. miRNA results are aggregated separately by aggregate_mirna_results() to keep output files distinct and properly typed.

Parameters:

• results_dir (str | Path) – Directory containing individual analysis results

• output_dir (str | Path) – Directory to write aggregated results

• genome_species (str) – Comma-separated list of genome species analyzed

Return type:

Path

Returns:

Path to output directory containing aggregated results

sirnaforge.core.off_target.run_mirna_seed_analysis(candidates_file, candidate_id, mirna_db, mirna_species, output_dir)

Run miRNA seed match analysis for candidate sequences.

This function uses the MiRNADatabaseManager to download and cache miRNA databases, builds BWA indices if needed, and performs seed match analysis.

Parameters:

• candidates_file (str | Path) – Path to FASTA file with candidate sequences

• candidate_id (str) – Candidate identifier

• mirna_db (str) – miRNA database name (mirgenedb, mirbase, etc.)

• mirna_species (list[str]) – List of species to analyze against

• output_dir (str | Path) – Directory to write results

Return type:

Path

Returns:

Path to output directory containing results

sirnaforge.core.off_target.aggregate_mirna_results(results_dir, output_dir, mirna_db, mirna_species)

Aggregate miRNA seed analysis results from multiple candidates using pandas.

Uses pandas + Pandera for efficient bulk reading and validation instead of manual line-by-line parsing with Pydantic models.

Parameters:

• results_dir (str | Path) – Directory containing individual miRNA analysis results

• output_dir (str | Path) – Directory to write aggregated results

• mirna_db (str) – miRNA database used for analysis

• mirna_species (str) – Comma-separated list of species analyzed

Return type:

Path

Returns:

Path to output directory containing aggregated results

Data Models

SiRNA Models

Pydantic models for siRNA design data structures.

class sirnaforge.models.sirna.FilterCriteria(**data)

Bases: BaseModel

Quality filters for siRNA candidate selection based on thermodynamic and empirical criteria.

Parameters:

data (Any)

gc_min: float

gc_max: float

max_poly_runs: int

max_paired_fraction: float

min_asymmetry_score: float

mfe_min: float | None

mfe_max: float | None

duplex_stability_min: float | None

duplex_stability_max: float | None

melting_temp_min: float | None

melting_temp_max: float | None

delta_dg_end_min: float | None

delta_dg_end_max: float | None

max_off_target_count: int | None

classmethod gc_max_greater_than_min(v, info)

Validate that gc_max is greater than or equal to gc_min.

Parameters:

• v (float)

• info (ValidationInfo)

Return type:

float

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.sirna.OffTargetFilterCriteria(**data)

Bases: BaseModel

Filtering criteria for off-target analysis results.

Controls which siRNA candidates fail due to excessive off-target potential.

Parameters:

data (Any)

max_transcriptome_hits_0mm: int | None

max_transcriptome_hits_1mm: int | None

max_transcriptome_hits_2mm: int | None

max_transcriptome_seed_perfect: int | None

max_mirna_perfect_seed: int | None

max_mirna_1mm_seed: int | None

fail_on_high_risk_mirna: bool

max_total_offtarget_hits: int | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.sirna.ScoringWeights(**data)

Bases: BaseModel

Relative weights for composite siRNA scoring components.

Parameters:

data (Any)

asymmetry: float

gc_content: float

accessibility: float

off_target: float

empirical: float

classmethod weights_sum_to_one(v, info)

Validate that scoring weights sum to approximately 1.0.

Parameters:

• v (float)

• info (ValidationInfo)

Return type:

float

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.sirna.DesignMode(*values)

Bases: str, Enum

Design mode for siRNA/miRNA-biogenesis-aware workflows.

SIRNA = 'sirna'

MIRNA = 'mirna'

class sirnaforge.models.sirna.MiRNADesignConfig(**data)

Bases: BaseModel

Configuration preset for miRNA-biogenesis-aware siRNA design.

This config encapsulates thresholds, defaults, and scoring weights optimized for miRNA-like processing (Drosha/Dicer recognition, Argonaute loading preferences, seed-based off-target analysis).

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

gc_min: float

gc_max: float

asymmetry_min: float

max_homopolymer: int

overhang: str

modifications: str

off_target_preset: str

scoring_weights: dict[str, float]

enable_pri_hairpin_validation: bool

class sirnaforge.models.sirna.DesignParameters(**data)

Bases: BaseModel

Complete configuration parameters for siRNA design workflow.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

design_mode: DesignMode

sirna_length: int

top_n: int

filters: FilterCriteria

scoring: ScoringWeights

avoid_snps: bool

check_off_targets: bool

predict_structure: bool

apply_modifications: bool

modification_pattern: str

default_overhang: str

snp_file: str | None

genome_index: str | None

class sirnaforge.models.sirna.SequenceType(*values)

Bases: str, Enum

Categories of input sequence types for siRNA design.

TRANSCRIPT = 'transcript'

GENOMIC = 'genomic'

CDS = 'cds'

UTR = 'utr'

class sirnaforge.models.sirna.SiRNACandidate(**data)

Bases: BaseModel

Individual siRNA candidate with computed thermodynamic and efficacy properties.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str

transcript_id: str

position: int

guide_sequence: str

passenger_sequence: str

gc_content: float

length: int

asymmetry_score: float

duplex_stability: float | None

structure: str | None

mfe: float | None

paired_fraction: float

off_target_count: int

off_target_penalty: float

transcriptome_hits_total: int

transcriptome_hits_0mm: int

transcriptome_hits_1mm: int

transcriptome_hits_2mm: int

transcriptome_hits_seed_0mm: int

mirna_hits_total: int

mirna_hits_0mm_seed: int

mirna_hits_1mm_seed: int

mirna_hits_high_risk: int

guide_pos1_base: str | None

pos1_pairing_state: str | None

seed_class: str | None

supp_13_16_score: float | None

seed_7mer_hits: int | None

seed_8mer_hits: int | None

seed_hits_weighted: float | None

off_target_seed_risk_class: str | None

transcript_hit_count: int

transcript_hit_fraction: float

component_scores: dict[str, float]

composite_score: float

class FilterStatus(*values)

Bases: str, Enum

Filter status codes for quality control.

PASS = 'PASS'

GC_OUT_OF_RANGE = 'GC_OUT_OF_RANGE'

POLY_RUNS = 'POLY_RUNS'

EXCESS_PAIRING = 'EXCESS_PAIRING'

LOW_ASYMMETRY = 'LOW_ASYMMETRY'

DIRTY_CONTROL = 'DIRTY_CONTROL'

passes_filters: bool | FilterStatus

quality_issues: list[str]

overlapped_variants: list[dict[str, Any]]

allele_specific: bool

targeted_alleles: list[str]

variant_mode: str | None

guide_metadata: StrandMetadata | None

passenger_metadata: StrandMetadata | None

classmethod validate_nucleotide_sequence(v)

Validate that sequence contains only valid nucleotides.

Parameters:

v (str)

Return type:

str

classmethod sequences_same_length(v, info)

Validate that passenger sequence is same length as guide sequence.

Parameters:

• v (str)

• info (ValidationInfo)

Return type:

str

to_fasta(include_metadata=False)

Return FASTA format representation of the guide sequence.

Parameters:

include_metadata (bool) – If True and guide_metadata is present, include it in the header

Return type:

str

Returns:

FASTA-formatted string with candidate ID as header and guide sequence.

class sirnaforge.models.sirna.DesignResult(**data)

Bases: BaseModel

Complete results from siRNA design workflow with metadata and statistics.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

input_file: str

parameters: DesignParameters

candidates: list[SiRNACandidate]

top_candidates: list[SiRNACandidate]

total_sequences: int

total_candidates: int

filtered_candidates: int

processing_time: float

tool_versions: dict[str, str]

rejected_candidates: list[SiRNACandidate]

save_csv(filepath)

Save siRNA candidates to CSV file with comprehensive validation.

Exports all candidates to CSV format with full thermodynamic metrics. The DataFrame is validated against SiRNACandidateSchema before saving to ensure data integrity and proper column types.

Parameters:

filepath (str) – Output CSV file path

Return type:

DataFrame[SiRNACandidateSchema]

Returns:

Validated DataFrame conforming to SiRNACandidateSchema

Raises:

pandera.errors.SchemaError – If data validation fails

get_summary()

Generate summary statistics for the design results.

Return type:

dict[str, Any]

Returns:

Dictionary containing key metrics including sequence counts, processing time, best score, and tool versions used.

Chemical Modifications

Data models for siRNA chemical modifications and metadata.

This module provides structured representations for chemical modifications, overhangs, and provenance metadata associated with siRNA strands.

class sirnaforge.models.modifications.ConfirmationStatus(*values)

Bases: str, Enum

Confirmation status for siRNA sequence data.

PENDING = 'pending'

CONFIRMED = 'confirmed'

class sirnaforge.models.modifications.SourceType(*values)

Bases: str, Enum

Source type for siRNA provenance.

PATENT = 'patent'

PUBLICATION = 'publication'

CLINICAL_TRIAL = 'clinical_trial'

DATABASE = 'database'

DESIGNED = 'designed'

OTHER = 'other'

class sirnaforge.models.modifications.Provenance(**data)

Bases: BaseModel

Provenance information for siRNA sequences.

Tracks the origin and validation status of siRNA sequences.

Parameters:

data (Any)

source_type: SourceType

identifier: str

url: str | None

to_header_string()

Convert provenance to FASTA header format.

Returns:

US10060921B2”

Return type:

Formatted string like “Patent

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.modifications.ChemicalModification(**data)

Bases: BaseModel

Chemical modification annotation for siRNA strands.

Represents a specific type of chemical modification and the positions where it occurs in the sequence.

Parameters:

data (Any)

type: str

positions: list[int]

classmethod validate_type(v)

Validate modification type is not empty.

Parameters:

v (str)

Return type:

str

classmethod validate_positions(v)

Validate positions are positive integers.

Parameters:

v (list[int])

Return type:

list[int]

to_header_string()

Convert modification to FASTA header format.

Return type:

str

Returns:

Formatted string like “2OMe(1,4,6,11,13,16,19)” or “2F()” for no positions

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.modifications.StrandRole(*values)

Bases: str, Enum

Role of the siRNA strand in the duplex.

GUIDE = 'guide'

SENSE = 'sense'

ANTISENSE = 'antisense'

PASSENGER = 'passenger'

class sirnaforge.models.modifications.StrandMetadata(**data)

Bases: BaseModel

Complete metadata for a single siRNA strand.

This model captures all relevant information about a siRNA strand including sequence, modifications, overhangs, and provenance.

Parameters:

data (Any)

id: str

sequence: str

overhang: str | None

chem_mods: list[ChemicalModification]

notes: str | None

provenance: Provenance | None

confirmation_status: ConfirmationStatus

classmethod validate_sequence(v)

Validate sequence contains only valid nucleotides.

Parameters:

v (str)

Return type:

str

validate_modification_positions()

Validate that modification positions don’t exceed sequence length.

Return type:

StrandMetadata

to_fasta_header(target_gene=None, strand_role=None)

Generate FASTA header with embedded metadata.

Parameters:

• target_gene (str | None) – Target gene name

• strand_role (StrandRole | None) – Role of this strand in the duplex

Return type:

str

Returns:

FASTA header string with key-value pairs

__getitem__(item)

Get item by key.

Parameters:

item (str)

Return type:

Any

get(item, default=None)

Get item by key with default.

Parameters:

• item (str)

• default (Any)

Return type:

Any

__contains__(item)

Check if item is in fields.

Parameters:

item (object)

Return type:

bool

keys()

Get field keys.

Return type:

Iterable[str]

items()

Get field items.

Return type:

Iterable[tuple[str, Any]]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.modifications.SequenceRecord(**data)

Bases: BaseModel

Complete sequence record with strand metadata.

Associates a strand with its target and role information.

Parameters:

data (Any)

target_gene: str

strand_role: StrandRole

metadata: StrandMetadata

to_fasta()

Generate complete FASTA record.

Return type:

str

Returns:

Multi-line FASTA string with header and sequence

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Off-Target Models

Pydantic models for off-target analysis data structures.

This module provides validated data models for: - BWA alignment results (both genome and miRNA) - Aggregated off-target summaries - Analysis metadata and statistics

Using Pydantic ensures type safety, automatic validation, and clean serialization to JSON/TSV formats.

class sirnaforge.models.off_target.AlignmentStrand(*values)

Bases: str, Enum

Genomic strand orientation.

FORWARD = '+'

REVERSE = '-'

class sirnaforge.models.off_target.AnalysisMode(*values)

Bases: str, Enum

BWA alignment analysis mode.

MIRNA_SEED = 'mirna_seed'

TRANSCRIPTOME = 'transcriptome'

class sirnaforge.models.off_target.MiRNADatabase(*values)

Bases: str, Enum

Supported miRNA database sources.

Values correspond to database identifiers used by MiRNADatabaseManager. Using str enum allows seamless string comparison while providing validation.

MIRGENEDB = 'mirgenedb'

MIRBASE = 'mirbase'

MIRBASE_HIGH_CONF = 'mirbase_high_conf'

MIRBASE_HAIRPIN = 'mirbase_hairpin'

TARGETSCAN = 'targetscan'

class sirnaforge.models.off_target.BaseAlignmentHit(**data)

Bases: BaseModel, ABC

Base class for alignment hits with common fields and validators.

This abstract base class contains all shared fields and validation logic for both off-target and miRNA alignment hits.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'frozen': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

qname: str

qseq: str

coord: int

strand: AlignmentStrand

cigar: str

mapq: int

as_score: int | None

nm: int

seed_mismatches: int

offtarget_score: float

classmethod validate_sequence(v)

Ensure sequence contains only valid nucleotide characters.

Parameters:

v (str)

Return type:

str

classmethod validate_cigar(v)

Basic CIGAR string validation.

Parameters:

v (str)

Return type:

str

abstractmethod to_dict()

Convert to dictionary for TSV/JSON serialization.

Return type:

dict[str, Any]

to_tsv_row()

Convert to TSV row format.

Return type:

str

abstractmethod classmethod tsv_header()

Get TSV header line.

Return type:

str

class sirnaforge.models.off_target.OffTargetHit(**data)

Bases: BaseAlignmentHit

Single off-target alignment hit from BWA analysis.

Represents one potential off-target binding site identified by sequence alignment against a reference genome or transcriptome.

Parameters:

data (Any)

species: str

rname: str

to_dict()

Convert to dictionary for TSV/JSON serialization.

Return type:

dict[str, Any]

classmethod tsv_header()

Get TSV header line.

Return type:

str

model_config: ClassVar[ConfigDict] = {'frozen': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.off_target.MiRNAHit(**data)

Bases: BaseAlignmentHit

Single miRNA seed match hit from BWA analysis.

Represents a potential miRNA-like seed match identified by alignment against miRNA databases.

Parameters:

data (Any)

species: str

database: MiRNADatabase | str

mirna_id: str

to_dict()

Convert to dictionary for TSV/JSON serialization.

Return type:

dict[str, Any]

classmethod tsv_header()

Get TSV header line.

Return type:

str

model_config: ClassVar[ConfigDict] = {'frozen': False, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.off_target.BaseSummary(**data)

Bases: BaseModel

Base class for analysis summary statistics with common metadata fields.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'frozen': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

candidate_id: str

total_sequences: int

total_hits: int

timestamp: str

status: str

class sirnaforge.models.off_target.AnalysisSummary(**data)

Bases: BaseSummary

Summary statistics for a single candidate’s off-target analysis.

Parameters:

data (Any)

species: str

mode: AnalysisMode

mean_mapq: float | None

mean_mismatches: float | None

mean_seed_mismatches: float | None

model_config: ClassVar[ConfigDict] = {'frozen': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.off_target.MiRNASummary(**data)

Bases: BaseSummary

Summary statistics for miRNA seed match analysis.

Note: total_hits represents validated, high-quality seed region matches. hits_per_species represents raw alignment counts (may include low-quality matches).

Parameters:

data (Any)

mirna_database: MiRNADatabase | str

species_analyzed: list[str]

hits_per_species: dict[str, int]

total_raw_alignments: int

parameters: dict[str, Any]

model_config: ClassVar[ConfigDict] = {'frozen': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.off_target.BaseAggregatedSummary(**data)

Bases: BaseModel

Base class for aggregated analysis summaries with common fields.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'frozen': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

species_analyzed: list[str]

analysis_files_processed: int

combined_tsv: Path | None

combined_json: Path | None

summary_file: Path | None

timestamp: str

serialize_path(path)

Serialize Path objects to strings for JSON compatibility.

Parameters:

path (Path | None)

Return type:

str | None

class sirnaforge.models.off_target.AggregatedOffTargetSummary(**data)

Bases: BaseAggregatedSummary

Summary of aggregated off-target results across multiple candidates and genomes.

Parameters:

data (Any)

total_results: int

hits_per_species: dict[str, int]

human_hits: int

other_species_hits: int

species_file_counts: dict[str, int]

missing_species: list[str]

status: str

model_config: ClassVar[ConfigDict] = {'frozen': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.models.off_target.AggregatedMiRNASummary(**data)

Bases: BaseAggregatedSummary

Summary of aggregated miRNA results across multiple candidates.

Parameters:

data (Any)

total_mirna_hits: int

mirna_database: MiRNADatabase | str

hits_per_species: dict[str, int]

hits_per_candidate: dict[str, int]

human_hits: int

other_species_hits: int

total_candidates: int

model_config: ClassVar[ConfigDict] = {'frozen': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Transcript Annotation Models

Pydantic models for transcript annotation data structures.

class sirnaforge.models.transcript_annotation.Interval(**data)

Bases: BaseModel

Genomic interval with start, end, and optional strand information.

Parameters:

data (Any)

seq_region_name: str

start: int

end: int

strand: int | None

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

__hash__()

Make Interval hashable for use in sets/dicts.

Return type:

int

__eq__(other)

Compare Interval instances for equality.

Parameters:

other (object)

Return type:

bool

class sirnaforge.models.transcript_annotation.TranscriptAnnotation(**data)

Bases: BaseModel

Comprehensive transcript annotation from genomic databases.

Contains transcript metadata, genomic coordinates, exon/CDS structure, and source provenance for reproducibility.

Parameters:

data (Any)

transcript_id: str

gene_id: str

symbol: str | None

biotype: str | None

seq_region_name: str

start: int

end: int

strand: int

gene_interval: Interval | None

exons: list[Interval]

cds: list[Interval]

provider: str

endpoint: str | None

reference_choice: str | None

model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property chr: str

Return a chr-prefixed chromosome label (e.g., ‘chr17’).

class sirnaforge.models.transcript_annotation.TranscriptAnnotationBundle(**data)

Bases: BaseModel

Collection of transcript annotations with resolution tracking.

Bundles multiple transcript annotations from a single query, tracks which IDs were successfully resolved, and maintains reference provenance.

Parameters:

data (Any)

transcripts: dict[str, TranscriptAnnotation]

unresolved: list[str]

reference_choice: ReferenceChoice

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property resolved_count: int

Number of successfully resolved transcripts.

property unresolved_count: int

Number of IDs that could not be resolved.

property total_requested: int

Total number of IDs requested (resolved + unresolved).

Validation Schemas

Pandera schemas for siRNAforge data validation.

This module defines pandera schemas for validating the structure and content of various table-like outputs from the siRNAforge pipeline.

Modern schemas using class-based approach with type annotations for improved type safety, error reporting, and maintainability.

Use schemas: MySchema.validate(df) - validation errors provide detailed feedback.

class sirnaforge.models.schemas.SchemaConfig

Bases: object

Common configuration settings for all pandera schemas.

Provides consistent validation behavior across all siRNAforge data schemas with type coercion, strict column checking, and flexible column ordering.

coerce = True

strict = True

ordered = False

class sirnaforge.models.schemas.SiRNACandidateSchema(*args, **kwargs)

Bases: DataFrameModel

Validation schema for siRNA candidate results (CSV output).

Ensures data integrity and biological validity of siRNA design results with comprehensive checks for sequence composition, thermodynamic parameters, and scoring metrics. Includes optimal value ranges for key metrics based on research-backed thermodynamic principles.

Expected columns include sequences, thermodynamic scores (asymmetry, MFE, duplex stability), off-target counts, and composite quality scores.

class Config

Bases: SchemaConfig

Schema configuration with improved error reporting.

description = 'siRNA candidate validation schema'

title = 'SiRNA Design Results'

add_missing_columns = True

strict = False

name = 'SiRNACandidateSchema'

id: Series[str] = 'id'

transcript_id: Series[str] = 'transcript_id'

position: Series[int] = 'position'

guide_sequence: Series[str] = 'guide_sequence'

passenger_sequence: Series[str] = 'passenger_sequence'

gc_content: Series[float] = 'gc_content'

asymmetry_score: Series[float] = 'asymmetry_score'

paired_fraction: Series[float] = 'paired_fraction'

structure: Series[Any] = 'structure'

mfe: Series[float] = 'mfe'

duplex_stability_dg: Series[float] = 'duplex_stability_dg'

duplex_stability_score: Series[float] = 'duplex_stability_score'

dg_5p: Series[float] = 'dg_5p'

dg_3p: Series[float] = 'dg_3p'

delta_dg_end: Series[float] = 'delta_dg_end'

melting_temp_c: Series[float] = 'melting_temp_c'

off_target_count: Series[int] = 'off_target_count'

guide_pos1_base: Series[str] = 'guide_pos1_base'

pos1_pairing_state: Series[str] = 'pos1_pairing_state'

seed_class: Series[str] = 'seed_class'

supp_13_16_score: Series[float] = 'supp_13_16_score'

seed_7mer_hits: Series[Int64Dtype] = 'seed_7mer_hits'

seed_8mer_hits: Series[Int64Dtype] = 'seed_8mer_hits'

seed_hits_weighted: Series[float] = 'seed_hits_weighted'

off_target_seed_risk_class: Series[str] = 'off_target_seed_risk_class'

transcript_hit_count: Series[int] = 'transcript_hit_count'

transcript_hit_fraction: Series[float] = 'transcript_hit_fraction'

composite_score: Series[float] = 'composite_score'

passes_filters: Series[Any] = 'passes_filters'

guide_overhang: Series[str] = 'guide_overhang'

guide_modifications: Series[str] = 'guide_modifications'

passenger_overhang: Series[str] = 'passenger_overhang'

passenger_modifications: Series[str] = 'passenger_modifications'

variant_mode: Series[str] = 'variant_mode'

allele_specific: Series[bool] = 'allele_specific'

targeted_alleles: Series[str] = 'targeted_alleles'

overlapped_variants: Series[str] = 'overlapped_variants'

classmethod check_passes_filters_values(df)

Ensure passes_filters contains allowed filter status values.

Parameters:

df (DataFrame)

Return type:

bool

classmethod check_sequence_lengths(df)

Validate siRNA sequences are in functional range (19-23 nt).

Parameters:

df (DataFrame)

Return type:

bool

classmethod check_nucleotide_sequences(df)

Validate sequences contain only valid RNA/DNA bases.

Parameters:

df (DataFrame)

Return type:

bool

class sirnaforge.models.schemas.ORFValidationSchema(*args, **kwargs)

Bases: DataFrameModel

Validation schema for open reading frame analysis results (tab-delimited output).

Validates ORF detection and characterization results with proper handling of nullable fields for cases where no valid ORF is found. Includes metrics for transcript composition, ORF boundaries, codon usage, and GC content within coding regions.

Used to validate outputs from ORF analysis tools and ensure data consistency for downstream siRNA target validation.

class Config

Bases: SchemaConfig

Schema configuration.

description = 'ORF validation analysis schema'

title = 'ORF Analysis Results'

strict = False

name = 'ORFValidationSchema'

transcript_id: Series[str] = 'transcript_id'

sequence_length: Series[int] = 'sequence_length'

gc_content: Series[float] = 'gc_content'

orfs_found: Series[int] = 'orfs_found'

has_valid_orf: Series[bool] = 'has_valid_orf'

longest_orf_start: Series[Any] = 'longest_orf_start'

longest_orf_end: Series[Any] = 'longest_orf_end'

longest_orf_length: Series[Any] = 'longest_orf_length'

longest_orf_frame: Series[Any] = 'longest_orf_frame'

start_codon: Series[Any] = 'start_codon'

stop_codon: Series[Any] = 'stop_codon'

orf_gc_content: Series[Any] = 'orf_gc_content'

class sirnaforge.models.schemas.OffTargetHitsSchema(*args, **kwargs)

Bases: DataFrameModel

DEPRECATED: Use MiRNAAlignmentSchema or GenomeAlignmentSchema instead.

Legacy validation schema for off-target analysis results (TSV output). This schema is too generic and doesn’t match actual BWA output format.

Migration Guide: - For miRNA seed analysis → Use MiRNAAlignmentSchema - For genome/transcriptome → Use GenomeAlignmentSchema

Will be removed in v0.3.0.

class Config

Bases: SchemaConfig

Schema configuration with relaxed strictness for external tool outputs.

description = 'DEPRECATED: Generic off-target schema'

title = 'Off-target Prediction Results (DEPRECATED)'

strict = False

name = 'OffTargetHitsSchema'

qname: Series[str] = 'qname'

target_id: Series[Any] = 'target_id'

species: Series[Any] = 'species'

chromosome: Series[Any] = 'chromosome'

position: Series[Any] = 'position'

strand: Series[Any] = 'strand'

mismatches: Series[Any] = 'mismatches'

alignment_score: Series[Any] = 'alignment_score'

offtarget_score: Series[Any] = 'offtarget_score'

target_sequence: Series[Any] = 'target_sequence'

class sirnaforge.models.schemas.MiRNAAlignmentSchema(*args, **kwargs)

Bases: DataFrameModel

Pandera schema for miRNA seed match alignment results (TSV/DataFrame).

Validates tabular data from BWA-MEM2 miRNA seed analysis. Each row represents one alignment between an siRNA candidate and a miRNA seed region.

Use this for: - Reading *_mirna_analysis.tsv files - Validating pandas DataFrames from miRNA analysis - Bulk operations on miRNA alignment results

Corresponding Pydantic model: models.off_target.MiRNAHit (for single rows)

class Config

Bases: SchemaConfig

Schema configuration.

description = 'miRNA seed match alignment results'

title = 'miRNA Alignment DataFrame'

strict = True

coerce = True

name = 'MiRNAAlignmentSchema'

qname: Series[str] = 'qname'

qseq: Series[str] = 'qseq'

species: Series[str] = 'species'

database: Series[str] = 'database'

mirna_id: Series[str] = 'mirna_id'

coord: Series[int] = 'coord'

strand: Series[str] = 'strand'

cigar: Series[str] = 'cigar'

mapq: Series[int] = 'mapq'

as_score: Series[Int64Dtype] = 'as_score'

nm: Series[int] = 'nm'

seed_mismatches: Series[int] = 'seed_mismatches'

offtarget_score: Series[float] = 'offtarget_score'

classmethod validate_seed_mismatches(df)

Ensure seed_mismatches <= nm (total mismatches).

Parameters:

df (DataFrame)

Return type:

bool

classmethod validate_perfect_match_score(df)

Perfect matches (nm=0) should have offtarget_score == 0.0 (highest risk).

Parameters:

df (DataFrame)

Return type:

bool

class sirnaforge.models.schemas.GenomeAlignmentSchema(*args, **kwargs)

Bases: DataFrameModel

Pandera schema for genome/transcriptome off-target alignment results (TSV/DataFrame).

Validates tabular data from BWA-MEM2 genome/transcriptome analysis. Each row represents one potential off-target alignment in the genome.

Use this for: - Reading *_analysis.tsv files from genome alignment - Validating pandas DataFrames from transcriptome off-target analysis - Bulk operations on genome alignment results

Corresponding Pydantic model: models.off_target.OffTargetHit (for single rows)

class Config

Bases: SchemaConfig

Schema configuration.

description = 'Genome/transcriptome off-target alignment results'

title = 'Genome Alignment DataFrame'

strict = True

coerce = True

name = 'GenomeAlignmentSchema'

qname: Series[str] = 'qname'

qseq: Series[str] = 'qseq'

species: Series[str] = 'species'

rname: Series[str] = 'rname'

coord: Series[int] = 'coord'

strand: Series[str] = 'strand'

cigar: Series[str] = 'cigar'

mapq: Series[int] = 'mapq'

as_score: Series[Int64Dtype] = 'as_score'

nm: Series[int] = 'nm'

seed_mismatches: Series[int] = 'seed_mismatches'

offtarget_score: Series[float] = 'offtarget_score'

classmethod validate_seed_mismatches(df)

Ensure seed_mismatches <= nm (total mismatches).

Parameters:

df (DataFrame)

Return type:

bool

classmethod validate_score_consistency(df)

Perfect matches should have offtarget_score == 0.0 (highest risk).

Parameters:

df (DataFrame)

Return type:

bool

Data Access

Base Data Classes

Shared base classes and utilities for genomic data analysis.

exception sirnaforge.data.base.DatabaseError(message, database=None)

Bases: Exception

Base exception for database-related errors.

Parameters:

• message (str)

• database (str | None)

__init__(message, database=None)

Initialize database error.

Parameters:

• message (str)

• database (str | None)

exception sirnaforge.data.base.DatabaseAccessError(message, database=None)

Bases: DatabaseError

Exception for network/access issues (firewall, timeout, server down).

Parameters:

• message (str)

• database (str | None)

exception sirnaforge.data.base.GeneNotFoundError(query, database=None)

Bases: DatabaseError

Exception for when a gene is not found in the database.

Parameters:

• query (str)

• database (str | None)

__init__(query, database=None)

Initialize gene not found error.

Parameters:

• query (str)

• database (str | None)

class sirnaforge.data.base.DatabaseType(*values)

Bases: str, Enum

Supported genomic databases.

ENSEMBL = 'ensembl'

REFSEQ = 'refseq'

GENCODE = 'gencode'

class sirnaforge.data.base.SequenceType(*values)

Bases: str, Enum

Types of sequence data that can be retrieved.

CDNA = 'cdna'

CDS = 'cds'

PROTEIN = 'protein'

GENOMIC = 'genomic'

class sirnaforge.data.base.GeneInfo(**data)

Bases: BaseModel

Gene information model.

Parameters:

data (Any)

gene_id: str

gene_name: str | None

gene_type: str | None

chromosome: str | None

start: int | None

end: int | None

strand: int | None

description: str | None

database: DatabaseType

model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.data.base.TranscriptInfo(**data)

Bases: BaseModel

Transcript information model.

Parameters:

data (Any)

transcript_id: str

transcript_name: str | None

transcript_type: str | None

gene_id: str

gene_name: str | None

sequence: str | None

length: int | None

database: DatabaseType

is_canonical: bool

model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_sequence(v)

Validate RNA sequence.

Parameters:

v (str | None)

Return type:

str | None

class sirnaforge.data.base.AbstractDatabaseClient(timeout=30)

Bases: ABC

Abstract base class for database clients.

Parameters:

timeout (int)

__init__(timeout=30)

Initialize database client.

Parameters:

timeout (int)

abstractmethod async search_gene(query, include_sequence=True)

Search for a gene and return gene info and transcripts.

Parameters:

• query (str) – Gene ID, gene name, or transcript ID

• include_sequence (bool) – Whether to fetch transcript sequences

Return type:

tuple[GeneInfo | None, list[TranscriptInfo]]

Returns:

Tuple of (gene_info, transcripts)

Raises:

• DatabaseAccessError – For network/server access issues

• GeneNotFoundError – When gene is not found in database

abstractmethod async get_sequence(identifier, sequence_type=SequenceType.CDNA)

Get sequence for a specific identifier.

Parameters:

• identifier (str) – Gene ID, transcript ID, etc.

• sequence_type (SequenceType) – Type of sequence to retrieve

Return type:

str

Returns:

Sequence string

Raises:

• DatabaseAccessError – For network/server access issues

• GeneNotFoundError – When identifier is not found in database

abstract property database_type: DatabaseType

Return the database type this client handles.

class sirnaforge.data.base.AbstractTranscriptAnnotationClient(timeout=30)

Bases: ABC

Abstract base class for transcript annotation clients.

Purpose and Scope: Provides genomic annotation metadata (exon/CDS structure, coordinates, biotype) WITHOUT fetching full transcript sequences. This is complementary to, not overlapping with, AbstractDatabaseClient which focuses on sequence retrieval.

Key Differences from GeneSearcher/AbstractDatabaseClient:

1. Focus: Structural annotations (exons, CDS intervals, genomic coordinates) vs. sequence data (cDNA, CDS, protein sequences)

2. Use Case: Enriching existing transcript metadata with genomic context vs. discovering and retrieving transcripts with sequences

3. Query Patterns: - By stable IDs: fetch_by_ids([‘ENST00000269305’]) - By genomic regions: fetch_by_regions([‘17:7661779-7687550’]) vs. GeneSearcher which queries by gene name/symbol

4. Caching Strategy: In-memory LRU cache with TTL for transient annotation data vs. ReferenceManager’s persistent file cache for large sequence datasets

When to Use: - Need exon/CDS boundaries for visualization or analysis - Need genomic coordinates for variant mapping - Need biotype information without full sequence download - Need to query multiple transcripts in a genomic region

When to Use GeneSearcher Instead: - Need transcript sequences for siRNA design - Need to discover transcripts by gene name/symbol - Need protein sequences or translations

Parameters:

timeout (int)

__init__(timeout=30)

Initialize transcript annotation client.

Parameters:

timeout (int) – Request timeout in seconds

abstractmethod async fetch_by_ids(ids, *, species, reference)

Fetch transcript annotations by stable IDs.

Parameters:

• ids (list[str]) – List of transcript or gene IDs (e.g., ENST00000269305, TP53)

• species (str) – Species name (e.g., ‘homo_sapiens’, ‘human’)

• reference (ReferenceChoice) – Reference assembly/release choice

Return type:

TranscriptAnnotationBundle

Returns:

TranscriptAnnotationBundle containing resolved annotations

Raises:

DatabaseAccessError – For network/server access issues

abstractmethod async fetch_by_regions(regions, *, species, reference)

Fetch transcript annotations by genomic regions.

Parameters:

• regions (list[str]) – List of genomic regions in format ‘chr:start-end’ (e.g., ‘17:7661779-7687550’)

• species (str) – Species name (e.g., ‘homo_sapiens’, ‘human’)

• reference (ReferenceChoice) – Reference assembly/release choice

Return type:

TranscriptAnnotationBundle

Returns:

TranscriptAnnotationBundle containing all transcripts overlapping regions

Raises:

DatabaseAccessError – For network/server access issues

class sirnaforge.data.base.EnsemblClient(timeout=30, base_url='https://rest.ensembl.org')

Bases: AbstractDatabaseClient

Client for Ensembl REST API interactions.

Parameters:

• timeout (int)

• base_url (str)

__init__(timeout=30, base_url='https://rest.ensembl.org')

Initialize Ensembl client.

Parameters:

• timeout (int)

• base_url (str)

property database_type: DatabaseType

Return the database type this client handles.

async search_gene(query, include_sequence=True)

Search for a gene and return gene info and transcripts.

Parameters:

• query (str)

• include_sequence (bool)

Return type:

tuple[GeneInfo | None, list[TranscriptInfo]]

async get_sequence(identifier, sequence_type=SequenceType.CDNA, headers=None)

Get sequence from Ensembl REST API.

Parameters:

• identifier (str) – Gene ID, transcript ID, etc.

• sequence_type (SequenceType) – Type of sequence to retrieve

• headers (dict | None) – Optional HTTP headers

Return type:

str

Returns:

Sequence string

Raises:

• DatabaseAccessError – For network/server access issues

• GeneNotFoundError – When identifier is not found in database

class sirnaforge.data.base.RefSeqClient(timeout=30, base_url='https://eutils.ncbi.nlm.nih.gov/entrez/eutils')

Bases: AbstractDatabaseClient

Client for RefSeq database via NCBI E-utilities API.

Parameters:

• timeout (int)

• base_url (str)

__init__(timeout=30, base_url='https://eutils.ncbi.nlm.nih.gov/entrez/eutils')

Initialize RefSeq client.

Parameters:

• timeout (int)

• base_url (str)

property database_type: DatabaseType

Return the database type this client handles.

async search_gene(query, include_sequence=True)

Search for a gene and return gene info and transcripts.

Parameters:

• query (str)

• include_sequence (bool)

Return type:

tuple[GeneInfo | None, list[TranscriptInfo]]

async get_sequence(identifier, _sequence_type=SequenceType.CDNA)

Get sequence for a specific identifier from NCBI.

Parameters:

• identifier (str)

• _sequence_type (SequenceType)

Return type:

str

class sirnaforge.data.base.GencodeClient(timeout=30)

Bases: AbstractDatabaseClient

Client for GENCODE database.

Parameters:

timeout (int)

__init__(timeout=30)

Initialize GENCODE client.

Parameters:

timeout (int)

property database_type: DatabaseType

Return the database type this client handles.

async search_gene(query, include_sequence=True)

Search for a gene and return gene info and transcripts.

Parameters:

• query (str)

• include_sequence (bool)

Return type:

tuple[GeneInfo | None, list[TranscriptInfo]]

async get_sequence(_identifier, _sequence_type=SequenceType.CDNA)

Get sequence for a specific identifier from GENCODE.

Parameters:

• _identifier (str)

• _sequence_type (SequenceType)

Return type:

str

class sirnaforge.data.base.SequenceUtils

Bases: object

Utility functions for sequence analysis.

static calculate_gc_content(sequence)

Calculate GC content of a sequence.

Parameters:

sequence (str)

Return type:

float

static reverse_complement(sequence)

Get reverse complement of DNA sequence.

Parameters:

sequence (str)

Return type:

str

static transcribe_dna_to_rna(sequence)

Convert DNA sequence to RNA (T -> U).

Parameters:

sequence (str)

Return type:

str

static reverse_transcribe_rna_to_dna(sequence)

Convert RNA sequence to DNA (U -> T).

Parameters:

sequence (str)

Return type:

str

class sirnaforge.data.base.FastaUtils

Bases: object

Utility functions for FASTA file operations.

static save_sequences_fasta(sequences, output_path, line_length=80)

Save sequences to FASTA format.

Parameters:

• sequences (list[tuple[str, str]]) – List of (header, sequence) tuples

• output_path (str | Path) – Output file path

• line_length (int) – Maximum line length for sequence

Return type:

None

static read_fasta(file_path)

Read sequences from FASTA file.

Parameters:

file_path (str | Path) – Path to FASTA file

Return type:

list[tuple[str, str]]

Returns:

List of (header, sequence) tuples

static parse_fasta_to_dict(file_path)

Parse FASTA file into a dictionary.

Parameters:

file_path (str | Path) – Path to FASTA file

Return type:

dict[str, str]

Returns:

Dictionary mapping sequence names to sequences

static write_dict_to_fasta(sequences, output_path)

Write sequences dictionary to FASTA format.

Parameters:

• sequences (dict[str, str]) – Dictionary of sequence name -> sequence

• output_path (str | Path) – Output file path

Return type:

None

static validate_sirna_sequences(sequences, expected_length=21)

Validate siRNA sequences for correct length and nucleotide content.

Parameters:

• sequences (dict[str, str]) – Dictionary of sequence name -> sequence

• expected_length (int) – Expected siRNA length

Return type:

dict[str, str]

Returns:

Dictionary of valid sequences

sirnaforge.data.base.get_database_display_name(database)

Get display name for database, handling both enum and string values.

Parameters:

database (DatabaseType)

Return type:

str

Gene Search

Gene search and sequence retrieval from multiple databases.

class sirnaforge.data.gene_search.GeneSearchResult(**data)

Bases: BaseModel

Complete gene search result.

Parameters:

data (Any)

query: str

database: DatabaseType

gene_info: GeneInfo | None

transcripts: list[TranscriptInfo]

error: str | None

is_access_error: bool

model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property success: bool

Check if search was successful.

class sirnaforge.data.gene_search.GeneSearcher(timeout=30, max_retries=3)

Bases: object

Search genes and retrieve sequences from genomic databases using multiple clients.

Parameters:

• timeout (int)

• max_retries (int)

__init__(timeout=30, max_retries=3)

Initialize gene searcher with database clients.

Parameters:

• timeout (int) – Request timeout in seconds

• max_retries (int) – Maximum retry attempts

get_client(database)

Get the client for a specific database.

Parameters:

database (DatabaseType)

Return type:

AbstractDatabaseClient

async search_gene_with_fallback(query, include_sequence=True)

Search for a gene with automatic fallback to other databases.

Tries databases in order: Ensembl -> RefSeq -> GENCODE Falls back to next database only if access is blocked (not if gene is not found).

Parameters:

• query (str) – Gene ID, gene name, or transcript ID

• include_sequence (bool) – Whether to fetch transcript sequences

Return type:

GeneSearchResult

Returns:

GeneSearchResult from the first accessible database

async search_gene(query, database=None, include_sequence=True)

Search for a gene and retrieve its isoforms.

Parameters:

• query (str) – Gene ID, gene name, or transcript ID

• database (DatabaseType | None) – Database to search (defaults to Ensembl)

• include_sequence (bool) – Whether to fetch transcript sequences

Return type:

GeneSearchResult

Returns:

GeneSearchResult with gene info and transcripts

async search_multiple_databases(query, databases=None, include_sequence=True)

Search across multiple databases.

Parameters:

• query (str) – Gene ID, gene name, or transcript ID

• databases (list[DatabaseType] | None) – List of databases to search

• include_sequence (bool) – Whether to fetch sequences

Return type:

list[GeneSearchResult]

Returns:

List of search results from each database

save_transcripts_fasta(transcripts, output_path, include_metadata=True)

Save transcripts to FASTA format using shared utility.

Parameters:

• transcripts (list[TranscriptInfo]) – List of transcript information

• output_path (str | Path) – Output file path

• include_metadata (bool) – Include metadata in FASTA headers

Return type:

None

sirnaforge.data.gene_search.search_gene_sync(query, database=DatabaseType.ENSEMBL, include_sequence=True)

Synchronous wrapper for gene search.

Parameters:

• query (str)

• database (DatabaseType)

• include_sequence (bool)

Return type:

GeneSearchResult

sirnaforge.data.gene_search.search_gene_with_fallback_sync(query, include_sequence=True)

Synchronous wrapper for gene search with fallback.

Parameters:

• query (str)

• include_sequence (bool)

Return type:

GeneSearchResult

sirnaforge.data.gene_search.search_multiple_databases_sync(query, databases=None, include_sequence=True)

Synchronous wrapper for multi-database search.

Parameters:

• query (str)

• databases (list[DatabaseType] | None)

• include_sequence (bool)

Return type:

list[GeneSearchResult]

ORF Analysis

ORF analysis and sequence validation for transcript sequences.

class sirnaforge.data.orf_analysis.ORFInfo(**data)

Bases: BaseModel

Information about an Open Reading Frame.

Parameters:

data (Any)

start_pos: int

end_pos: int

length: int

reading_frame: int

start_codon: str

stop_codon: str

has_valid_start: bool

has_valid_stop: bool

is_complete: bool

gc_content: float

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.data.orf_analysis.SequenceAnalysis(**data)

Bases: BaseModel

Complete sequence analysis including ORF information.

Parameters:

data (Any)

transcript_id: str

sequence_type: SequenceType

sequence_length: int

gc_content: float

orfs: list[ORFInfo]

longest_orf: ORFInfo | None

has_valid_orf: bool

cds_sequence: str | None

protein_sequence: str | None

cds_start: int | None

cds_end: int | None

utr5_length: int | None

utr3_length: int | None

sequence_region: str | None

model_config: ClassVar[ConfigDict] = {'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.data.orf_analysis.ORFAnalyzer(database_client=None)

Bases: object

Analyze ORFs in transcript sequences and validate sequence types.

Parameters:

database_client (AbstractDatabaseClient | None)

__init__(database_client=None)

Initialize ORF analyzer.

Parameters:

database_client (AbstractDatabaseClient | None) – Optional database client for retrieving additional sequence types

calculate_gc_content(sequence)

Calculate GC content using shared utility.

Parameters:

sequence (str)

Return type:

float

find_orfs(sequence, min_length=150)

Find all ORFs in a sequence (all 3 reading frames).

Parameters:

• sequence (str)

• min_length (int)

Return type:

list[ORFInfo]

translate_sequence(sequence)

Translate DNA sequence to protein.

Parameters:

sequence (str)

Return type:

str

async get_additional_sequence(transcript_id, sequence_type)

Retrieve specific sequence type using the database client if available.

Parameters:

• transcript_id (str) – Transcript identifier

• sequence_type (SequenceType) – Type of sequence to retrieve

Return type:

str | None

Returns:

Sequence string or None if not available or client not provided

async analyze_transcript(transcript)

Perform complete ORF analysis of a transcript.

Parameters:

transcript (TranscriptInfo)

Return type:

SequenceAnalysis

async analyze_transcripts(transcripts)

Analyze multiple transcripts.

Parameters:

transcripts (list[TranscriptInfo])

Return type:

dict[str, SequenceAnalysis]

sirnaforge.data.orf_analysis.create_orf_analyzer(database_client=None)

Create an ORF analyzer with optional database client.

Parameters:

database_client (AbstractDatabaseClient | None) – Optional database client for retrieving additional sequence types

Return type:

ORFAnalyzer

Returns:

ORFAnalyzer instance

async sirnaforge.data.orf_analysis.analyze_multiple_transcript_orfs(transcripts, database_client=None)

Analyze ORFs in multiple transcripts.

Parameters:

• transcripts (list[TranscriptInfo]) – List of transcripts to analyze

• database_client (AbstractDatabaseClient | None) – Optional database client for additional sequence retrieval

Return type:

dict[str, SequenceAnalysis]

Returns:

Dictionary mapping transcript IDs to SequenceAnalysis results

Transcript Annotations

Transcript annotation providers using Ensembl REST and optional VEP enrichment.

This module provides clients for fetching genomic transcript annotations (exon/CDS structure, coordinates, biotype) separate from sequence retrieval.

Architecture Overview:

• EnsemblTranscriptModelClient: Primary implementation using Ensembl REST API

• VepConsequenceClient: Optional enrichment client (placeholder for future development)

Caching Strategy:

Uses in-memory LRU cache with TTL rather than ReferenceManager’s persistent file cache. This design choice is intentional because:

1. Data Size: Annotation JSON responses are small (KB) vs. sequence files (GB)

2. Volatility: Annotations may update with new releases; TTL provides freshness

3. Access Pattern: High frequency, low latency requirements during workflow execution

4. Scope: Transient metadata enrichment vs. permanent reference datasets

The cache automatically evicts oldest entries when reaching max_cache_entries, and entries expire after cache_ttl seconds.

Relationship to GeneSearcher:

• GeneSearcher: Discovers transcripts by gene name, fetches cDNA/protein sequences

• This module: Enriches known transcript IDs with genomic structural metadata

• Both can use Ensembl, but query different API endpoints for different purposes

• No redundancy: complementary data types that don’t overlap

class sirnaforge.data.transcript_annotation.EnsemblTranscriptModelClient(timeout=30, base_url='https://rest.ensembl.org', cache_ttl=3600, max_cache_entries=1000)

Bases: AbstractTranscriptAnnotationClient

Ensembl REST-based transcript annotation client.

Retrieves transcript metadata including genomic coordinates, exon/CDS structure, and biotype information using Ensembl’s public REST API.

API Endpoints Used:

1. Lookup by ID (/lookup/id/:id?expand=1): - Fetches detailed annotation for single transcript/gene ID - Returns exon coordinates, CDS intervals, biotype - Example: /lookup/id/ENST00000269305?expand=1

2. Overlap by Region (/overlap/region/:species/:region): - Fetches all transcripts overlapping genomic region - Useful for region-based queries - Example: /overlap/region/human/17:7661779-7687550?feature=transcript

Caching Implementation:

• Cache key format: “id:{species}:{identifier}:{reference}” or “region:{species}:{region}:{reference}”

• TTL: Configurable, default 1 hour (3600 seconds)

• Eviction: LRU when max_cache_entries reached (default 1000)

• Thread-safe: Single-process use only (workflow orchestration context)

Error Handling:

• 404: ID not found → added to unresolved list, no exception raised

• 403/503: Server unavailable → DatabaseAccessError raised

• Network errors: Wrapped in DatabaseAccessError with context

• Timeout: Configurable via timeout parameter

Example Usage:

>>> client = EnsemblTranscriptModelClient()
>>> reference = ReferenceChoice.explicit("GRCh38", reason="user-specified")
>>> bundle = await client.fetch_by_ids(
...     ids=["ENST00000269305"],
...     species="human",
...     reference=reference
... )
>>> print(f"Resolved: {bundle.resolved_count}, Unresolved: {bundle.unresolved_count}")

Parameters:

• timeout (int)

• base_url (str)

• cache_ttl (int)

• max_cache_entries (int)

__init__(timeout=30, base_url='https://rest.ensembl.org', cache_ttl=3600, max_cache_entries=1000)

Initialize Ensembl transcript annotation client.

Parameters:

• timeout (int) – Request timeout in seconds

• base_url (str) – Ensembl REST API base URL

• cache_ttl (int) – Cache time-to-live in seconds (default: 1 hour)

• max_cache_entries (int) – Maximum number of cached entries (default: 1000)

async fetch_by_ids(ids, *, species, reference)

Fetch transcript annotations by stable IDs using Ensembl lookup endpoint.

Parameters:

• ids (list[str]) – List of transcript or gene IDs

• species (str) – Species name (e.g., ‘homo_sapiens’, ‘human’)

• reference (ReferenceChoice) – Reference assembly/release choice

Return type:

TranscriptAnnotationBundle

Returns:

TranscriptAnnotationBundle with resolved annotations

async fetch_by_regions(regions, *, species, reference)

Fetch transcript annotations by genomic regions using Ensembl overlap endpoint.

Parameters:

• regions (list[str]) – List of regions in format ‘chr:start-end’ (e.g., ‘17:7661779-7687550’)

• species (str) – Species name (e.g., ‘homo_sapiens’, ‘human’)

• reference (ReferenceChoice) – Reference assembly/release choice

Return type:

TranscriptAnnotationBundle

Returns:

TranscriptAnnotationBundle with all transcripts overlapping regions

class sirnaforge.data.transcript_annotation.VepConsequenceClient(timeout=30, base_url='https://rest.ensembl.org')

Bases: object

Optional VEP (Variant Effect Predictor) consequence enrichment client.

Provides additional functional annotation for transcript variants. This is an optional enhancement and not required for base functionality.

Current Status: PLACEHOLDER

This client exists as a stub for future VEP integration. The enrich_annotations method currently returns the input bundle unchanged.

Future Implementation:

When activated (via config flag), this client will:

1. Query Ensembl VEP REST API for consequence predictions

2. Enrich TranscriptAnnotation objects with variant consequence types (missense, nonsense, etc.), conservation scores, regulatory feature overlaps, and population frequency data

3. Maintain consistent caching strategy with EnsemblTranscriptModelClient

Design Rationale:

Separated from EnsemblTranscriptModelClient because:

• VEP queries are expensive (rate-limited, slower)

• Not all workflows need consequence predictions

• Allows independent caching strategies

• Can be enabled/disabled via configuration

Parameters:

• timeout (int)

• base_url (str)

__init__(timeout=30, base_url='https://rest.ensembl.org')

Initialize VEP client.

Parameters:

• timeout (int) – Request timeout in seconds

• base_url (str) – Ensembl REST API base URL

async enrich_annotations(bundle, _species='homo_sapiens')

Enrich transcript annotations with VEP consequence data.

Parameters:

• bundle (TranscriptAnnotationBundle) – Existing transcript annotation bundle

• species – Species name for VEP queries

• _species (str)

Return type:

TranscriptAnnotationBundle

Returns:

Enriched bundle (currently returns input unchanged - placeholder for future VEP integration)

miRNA Management

miRNA Database Manager with multi-species support.

This module provides a clean interface for downloading, caching, and managing miRNA databases from multiple sources (MirGeneDB, miRBase, TargetScan) with automatic cache management and species-specific organization.

class sirnaforge.data.mirna_manager.MiRNASource(name, url, species, format='fasta', compressed=False, description='')

Bases: ReferenceSource

miRNA-specific database source configuration.

Inherits from ReferenceSource and can add miRNA-specific fields if needed.

Parameters:

• name (str)

• url (str)

• species (str)

• format (str)

• compressed (bool)

• description (str)

__init__(name, url, species, format='fasta', compressed=False, description='')

Parameters:

• name (str)

• url (str)

• species (str)

• format (str)

• compressed (bool)

• description (str)

class sirnaforge.data.mirna_manager.MiRNADatabaseManager(cache_dir=None, cache_ttl_days=30)

Bases: ReferenceManager[MiRNASource]

Elegant miRNA database manager with caching and multi-species support.

Parameters:

• cache_dir (str | Path | None)

• cache_ttl_days (int)

SOURCES = {'mirbase': {'human': MiRNASource(name='mirbase_mature', url='https://www.mirbase.org/download/CURRENT/mature.fa', species='human', format='fasta', compressed=False, description='miRBase mature miRNA sequences (all species, filtered for Homo sapiens - hsa)'), 'mouse': MiRNASource(name='mirbase_mature', url='https://www.mirbase.org/download/CURRENT/mature.fa', species='mouse', format='fasta', compressed=False, description='miRBase mature miRNA sequences (all species, filtered for Mus musculus - mmu)'), 'rat': MiRNASource(name='mirbase_mature', url='https://www.mirbase.org/download/CURRENT/mature.fa', species='rat', format='fasta', compressed=False, description='miRBase mature miRNA sequences (all species, filtered for Rattus norvegicus - rno)')}, 'mirbase_hairpin': {'human': MiRNASource(name='mirbase_hairpin', url='https://www.mirbase.org/download/CURRENT/hairpin.fa', species='human', format='fasta', compressed=False, description='miRBase hairpin precursor miRNA sequences (Homo sapiens - hsa)'), 'mouse': MiRNASource(name='mirbase_hairpin', url='https://www.mirbase.org/download/CURRENT/hairpin.fa', species='mouse', format='fasta', compressed=False, description='miRBase hairpin precursor miRNA sequences (Mus musculus - mmu)'), 'rat': MiRNASource(name='mirbase_hairpin', url='https://www.mirbase.org/download/CURRENT/hairpin.fa', species='rat', format='fasta', compressed=False, description='miRBase hairpin precursor miRNA sequences (Rattus norvegicus - rno)')}, 'mirbase_high_conf': {'human': MiRNASource(name='mirbase_mature_hc', url='https://www.mirbase.org/download/CURRENT/mature_high_conf.fa', species='human', format='fasta', compressed=False, description='miRBase high-confidence mature miRNA sequences (Homo sapiens - hsa)'), 'mouse': MiRNASource(name='mirbase_mature_hc', url='https://www.mirbase.org/download/CURRENT/mature_high_conf.fa', species='mouse', format='fasta', compressed=False, description='miRBase high-confidence mature miRNA sequences (Mus musculus - mmu)'), 'rat': MiRNASource(name='mirbase_mature_hc', url='https://www.mirbase.org/download/CURRENT/mature_high_conf.fa', species='rat', format='fasta', compressed=False, description='miRBase high-confidence mature miRNA sequences (Rattus norvegicus - rno)')}, 'mirgenedb': {'aga': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/aga?mat=1', species='aga', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Anopheles gambiae, NCBI:7165) [Mosquito]'), 'bta': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/bta?mat=1', species='bta', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Bos taurus, NCBI:9913) [Cow]'), 'cel': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/cel?mat=1', species='cel', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Caenorhabditis elegans, NCBI:6239) [C. elegans]'), 'cfa': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/cfa?mat=1', species='cfa', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Canis lupus familiaris, NCBI:9615) [Dog]'), 'dme': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/dme?mat=1', species='dme', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Drosophila melanogaster, NCBI:7227) [Fruit fly]'), 'dre': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/dre?mat=1', species='dre', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Danio rerio, NCBI:7955) [Zebrafish]'), 'eca': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/eca?mat=1', species='eca', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Equus caballus, NCBI:9796) [Horse]'), 'fca': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/fca?mat=1', species='fca', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Felis catus, NCBI:9685)'), 'gac': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/gac?mat=1', species='gac', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Gasterosteus aculeatus, NCBI:69293) [Stickleback]'), 'gga': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/gga?mat=1', species='gga', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Gallus gallus, NCBI:9031) [Chicken]'), 'ggo': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/ggo?mat=1', species='ggo', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Gorilla gorilla, NCBI:9593)'), 'hsa': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/hsa?mat=1', species='hsa', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Homo sapiens, NCBI:9606) [Human]'), 'mml': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/mml?mat=1', species='mml', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Macaca mulatta, NCBI:9544) [Rhesus macaque]'), 'mmu': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/mmu?mat=1', species='mmu', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Mus musculus, NCBI:10090) [Mouse]'), 'oar': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/oar?mat=1', species='oar', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Ovis aries, NCBI:9940) [Sheep]'), 'ola': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/ola?mat=1', species='ola', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Oryzias latipes, NCBI:8090) [Medaka]'), 'pma': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/pma?mat=1', species='pma', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Petromyzon marinus, NCBI:7757) [Sea lamprey]'), 'ptr': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/ptr?mat=1', species='ptr', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Pan troglodytes, NCBI:9598) [Chimpanzee]'), 'rno': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/rno?mat=1', species='rno', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Rattus norvegicus, NCBI:10116)'), 'spur': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/spur?mat=1', species='spur', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Strongylocentrotus purpuratus, NCBI:7668) [Purple sea urchin]'), 'ssc': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/ssc?mat=1', species='ssc', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Sus scrofa, NCBI:9823) [Pig]'), 'tgu': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/tgu?mat=1', species='tgu', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Meleagris gallopavo, NCBI:9103) [Turkey]'), 'xla': MiRNASource(name='mirgenedb', url='https://www.mirgenedb.org/fasta/xla?mat=1', species='xla', format='fasta', compressed=False, description='MirGeneDB high-confidence miRNAs (Xenopus laevis, NCBI:8355) [African clawed frog]')}, 'targetscan': {'human': MiRNASource(name='targetscan', url='https://www.targetscan.org/vert_80/vert_80_data_download/miR_Family_Info.txt.zip', species='human', format='tsv', compressed=True, description='TargetScan miRNA family data')}}

classmethod get_available_sources()

Return sorted list of supported database sources.

Return type:

list[str]

classmethod get_all_species()

Return sorted list of all species across sources.

Return type:

list[str]

classmethod get_species_for_source(source_name)

Return sorted list of species supported by a given source.

Parameters:

source_name (str)

Return type:

list[str]

classmethod get_species_aliases(source_name)

Return mapping of canonical species identifiers to their known aliases.

Parameters:

source_name (str)

Return type:

dict[str, list[str]]

classmethod get_canonical_species()

Return sorted list of canonical species keys.

Return type:

list[str]

classmethod canonicalize_species_name(species)

Normalize a raw species identifier to a canonical key.

Parameters:

species (str)

Return type:

str | None

classmethod canonicalize_species_list(species_list)

Normalize a list of species identifiers to canonical keys, preserving order.

Parameters:

species_list (Sequence[str])

Return type:

list[str]

classmethod get_genome_species_for_canonical(canonical_species)

Return genome species identifiers for canonical species keys.

Note: These are used for miRNA database lookups, not genomic DNA alignment. The term ‘genome’ here refers to the organism’s miRNA annotation set.

Parameters:

canonical_species (Sequence[str])

Return type:

list[str]

classmethod get_mirna_slugs_for_canonical(canonical_species, source_name)

Return normalized miRNA identifiers for canonical species.

Parameters:

• canonical_species (Sequence[str])

• source_name (str)

Return type:

list[str]

classmethod get_supported_canonical_species_for_source(source_name)

Return canonical species supported by a given source.

Parameters:

source_name (str)

Return type:

list[str]

classmethod resolve_species_selection(requested_species, source_name, mirna_overrides=None)

Resolve canonical, genome, and miRNA identifiers for the requested species.

Parameters:

• requested_species (Sequence[str])

• source_name (str)

• mirna_overrides (Sequence[str] | None)

Return type:

dict[str, list[str]]

classmethod normalize_species(source_name, species)

Normalize user-provided species identifiers to canonical keys.

Parameters:

• source_name (str)

• species (str)

Return type:

str | None

classmethod get_source_configuration(source_name, species)

Retrieve the MiRNASource configuration for a given source/species.

Parameters:

• source_name (str)

• species (str)

Return type:

MiRNASource | None

classmethod is_supported_source(source_name)

Check if a source is supported.

Parameters:

source_name (str)

Return type:

bool

classmethod is_supported_species(source_name, species)

Check if a species is supported for the given source.

Parameters:

• source_name (str)

• species (str)

Return type:

bool

classmethod get_mirgenedb_species_metadata()

Expose the MirGeneDB species metadata table.

Return type:

dict[str, dict[str, Any]]

__init__(cache_dir=None, cache_ttl_days=30)

Initialize the miRNA database manager.

Parameters:

• cache_dir (str | Path | None) – Directory for caching databases (default: ~/.cache/sirnaforge/mirna)

• cache_ttl_days (int) – Cache time-to-live in days

get_database(source_name, species, force_refresh=False)

Get miRNA database, downloading and filtering if needed.

Simplified caching: each species+source combination gets its own cache file.

Parameters:

• source_name (str) – Database source (“mirbase”, “mirbase_high_conf”, etc.)

• species (str) – Species name (“human”, “mouse”, “rat”)

• force_refresh (bool) – Force re-download even if cached

Return type:

Path | None

Returns:

Path to cached FASTA file, or None if failed

get_combined_database(sources, species, output_name=None)

Combine multiple databases into a single file.

Parameters:

• sources (list[str]) – List of source names to combine

• species (str) – Target species

• output_name (str | None) – Custom output filename (default: auto-generated)

Return type:

Path | None

Returns:

Path to combined FASTA file

list_available_databases()

List all available database sources and species.

Return type:

dict[str, dict[str, MiRNASource]]

clear_cache(confirm=False)

Clear the miRNA cache directory.

Parameters:

confirm (bool) – If True, actually delete files. If False, just return what would be deleted.

Return type:

dict[str, Any]

Returns:

Dictionary with information about files deleted or that would be deleted.

sirnaforge.data.mirna_manager.main()

CLI interface for the miRNA database manager.

Return type:

None

Species Registry

Canonical species registry and metadata for miRNA and genome mappings.

sirnaforge.data.species_registry.normalize_species_name(species)

Normalize species name to canonical form.

Parameters:

species (str) – Species name in any recognized form (e.g., ‘hsa’, ‘human’, ‘Homo sapiens’)

Return type:

str

Returns:

Canonical species name (e.g., ‘human’), or original string if not recognized

Examples

>>> normalize_species_name('hsa')
'human'
>>> normalize_species_name('Mus musculus')
'mouse'
>>> normalize_species_name('macaque')
'macaque'
>>> normalize_species_name('unknown')
'unknown'

Pipeline Integration

Nextflow CLI

Command-line entry points used by embedded Nextflow modules.

sirnaforge.pipeline.nextflow_cli.build_bwa_index_cli(fasta_file, species, output_dir='.')

Build BWA-MEM2 index for genome/transcriptome.

Parameters:

• fasta_file (str) – Path to input FASTA file

• species (str) – Species identifier

• output_dir (str) – Directory to write index files

Return type:

dict[str, Any]

Returns:

Dictionary with index prefix path

sirnaforge.pipeline.nextflow_cli.aggregate_results_cli(genome_species, output_dir='.', mirna_db=None, mirna_species=None)

Aggregate off-target analysis results from multiple candidates and genomes.

Parameters:

• genome_species (str) – Comma-separated list of species

• output_dir (str) – Directory to write aggregated results

• mirna_db (str | None) – The database that provided the reference

• mirna_species (str | None) – The species code for the matching miRNA

Return type:

dict[str, Any]

Returns:

Dictionary with aggregation statistics

sirnaforge.pipeline.nextflow_cli.aggregate_mirna_results_cli(mirna_db, mirna_species, results_dir='.', output_dir='.')

Aggregate miRNA seed analysis results from multiple candidates.

Parameters:

• mirna_db (str) – miRNA database name used for analysis

• mirna_species (str) – Comma-separated list of species analyzed

• results_dir (str) – Directory containing individual miRNA results

• output_dir (str) – Directory to write aggregated results

Return type:

dict[str, Any]

Returns:

Dictionary with aggregation statistics

Nextflow Configuration

Nextflow Configuration Management.

This module handles configuration for Nextflow workflows, including Docker settings, resource management, and parameter validation.

Simple Usage Examples:

# Auto-configure based on environment (easiest) config = NextflowConfig.auto_configure()

# Production settings config = NextflowConfig.for_production()

# Testing settings config = NextflowConfig.for_testing()

# Local Docker testing (uses local image built by ‘make docker’) config = NextflowConfig.for_local_docker_testing()

class sirnaforge.pipeline.nextflow.config.EnvironmentInfo(**data)

Bases: BaseModel

Information about the current execution environment.

Parameters:

data (Any)

running_in_docker: bool

docker_available: bool

requested_profile: str

recommended_profile: str

docker_image: str | None

profile_override_reason: str | None

is_profile_overridden()

Check if the recommended profile differs from the requested profile.

Return type:

bool

get_execution_summary()

Get a human-readable summary of the execution environment.

Return type:

str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.pipeline.nextflow.config.NextflowConfig(docker_image='ghcr.io/austin-s-h/sirnaforge:latest', profile='docker', work_dir=None, nxf_home=None, max_cpus=16, max_memory='128.GB', max_time='240.h', **kwargs)

Bases: object

Configuration manager for Nextflow workflows.

Parameters:

• docker_image (str)

• profile (str)

• work_dir (Path | None)

• nxf_home (Path | None)

• max_cpus (int)

• max_memory (str)

• max_time (str)

• kwargs (Any)

DEFAULT_SIRNAFORGE_DOCKER_IMAGE = 'ghcr.io/austin-s-h/sirnaforge:latest'

MEMORY_BUFFER_GB = 0.5

MIN_MEMORY_GB = 1

__init__(docker_image='ghcr.io/austin-s-h/sirnaforge:latest', profile='docker', work_dir=None, nxf_home=None, max_cpus=16, max_memory='128.GB', max_time='240.h', **kwargs)

Initialize Nextflow configuration.

Parameters:

• docker_image (str) – Docker container image to use

• profile (str) – Nextflow profile (docker, singularity, conda, local)

• work_dir (Path | None) – Working directory for Nextflow execution

• nxf_home (Path | None) – Nextflow home cache

• max_cpus (int) – Maximum CPU cores

• max_memory (str) – Maximum memory allocation

• max_time (str) – Maximum execution time

• **kwargs (Any) – Additional configuration parameters

get_nextflow_args(input_file, output_dir, genome_species, additional_params=None, include_test_profile=False)

Generate Nextflow command arguments.

Parameters:

• input_file (Path) – Input FASTA file path

• output_dir (Path) – Output directory

• genome_species (list[str]) – List of species for miRNA genome lookups (not genomic DNA)

• additional_params (dict[str, Any] | None) – Additional parameters to pass

• include_test_profile (bool) – Whether to include ‘test’ profile for integration testing

Return type:

list[str]

Returns:

List of command arguments for Nextflow

create_config_file(config_path)

Create a custom Nextflow configuration file.

Parameters:

config_path (Path) – Path where to create the config file

Return type:

Path

Returns:

Path to the created configuration file

validate_docker_available()

Check if Docker is available for Nextflow execution.

This checks if Docker can be used by Nextflow to run containers. Note: This is different from running tests inside Docker containers.

Return type:

bool

Returns:

True if Docker is available and accessible for Nextflow

is_running_in_docker()

Check if we’re currently running inside a Docker container.

This is useful for determining the appropriate execution profile when running tests or workflows.

Return type:

bool

Returns:

True if running inside a Docker container

get_execution_profile()

Get the appropriate execution profile based on available tools and environment.

This method considers: 1. Environment variables (SIRNAFORGE_USE_LOCAL_EXECUTION) 2. Whether we’re running inside a Docker container (for testing) 3. Whether Docker is available for Nextflow execution 4. Availability of Singularity or Conda as fallbacks 5. The requested profile

Return type:

str

Returns:

Recommended execution profile

get_environment_info()

Get information about the current execution environment.

This provides structured information about Docker availability, profile selection, and environment detection.

Return type:

EnvironmentInfo

Returns:

EnvironmentInfo model with environment details

classmethod for_testing()

Create a configuration optimized for testing.

This automatically detects if we’re running in Docker and adjusts accordingly. Uses uv/conda for environment management when available.

Return type:

NextflowConfig

Returns:

NextflowConfig instance with test-friendly settings

classmethod for_production(**kwargs)

Create a configuration optimized for production use.

This uses Docker by default for reproducible execution with full resources.

Parameters:

**kwargs (Any) – Additional configuration parameters to override defaults

Return type:

NextflowConfig

Returns:

NextflowConfig instance with production settings

classmethod auto_configure(**kwargs)

Auto-configure Nextflow settings based on environment detection.

This method automatically detects available tools and selects the best profile.

Parameters:

**kwargs (Any) – Additional configuration parameters to override defaults

Return type:

NextflowConfig

Returns:

NextflowConfig instance with auto-detected settings

Nextflow Runner

Nextflow Pipeline Runner.

This module provides a Python interface to execute Nextflow workflows for siRNA off-target analysis with proper Docker integration.

class sirnaforge.pipeline.nextflow.runner.NextflowRunner(config=None)

Bases: object

Execute Nextflow workflows from Python with proper error handling.

Parameters:

config (NextflowConfig | None)

__init__(config=None)

Initialize Nextflow runner.

Parameters:

config (NextflowConfig | None) – NextflowConfig instance, creates auto-configured if None

get_main_workflow()

Get path to the main Nextflow workflow.

Return type:

Path

get_pipeline_revision()

Expose the detected pipeline revision identifier.

Return type:

str

async run(input_file, output_dir, genome_species=None, **kwargs)

Simple method to run Nextflow workflow with auto-validation and defaults.

Parameters:

• input_file (Path) – Path to input FASTA file

• output_dir (Path) – Output directory for results

• genome_species (list[str] | None) – List of species for miRNA genome lookups (defaults to [“human”, “rat”, “rhesus”])

• **kwargs (Any) – Additional parameters passed to run_offtarget_analysis

Return type:

dict[str, Any]

Returns:

Dictionary containing execution results and metadata

Raises:

NextflowExecutionError – If workflow execution fails

run_sync(input_file, output_dir, genome_species=None, **kwargs)

Synchronous version of run() for simpler usage without async/await.

Parameters:

• input_file (Path) – Path to input FASTA file

• output_dir (Path) – Output directory for results

• genome_species (list[str] | None) – List of species for miRNA genome lookups (defaults to [“human”, “rat”, “rhesus”])

• **kwargs (Any) – Additional parameters passed to run_offtarget_analysis

Return type:

dict[str, Any]

Returns:

Dictionary containing execution results and metadata

async run_offtarget_analysis(input_file, output_dir, genome_species, additional_params=None, show_progress=True)

Run the off-target analysis Nextflow workflow.

Parameters:

• input_file (Path) – Path to siRNA candidates FASTA file

• output_dir (Path) – Output directory for results

• genome_species (list[str]) – List of species for miRNA genome lookups

• additional_params (dict[str, Any] | None) – Additional parameters for the workflow

• show_progress (bool) – Whether to show progress indicators

Return type:

dict[str, Any]

Returns:

Dictionary containing execution results and metadata

Raises:

NextflowExecutionError – If workflow execution fails

validate_installation()

Validate that Nextflow and required tools are available.

Return type:

dict[str, bool | str]

Returns:

Dictionary of tool availability status

classmethod for_testing()

Create a runner configured for testing.

This uses the test configuration which automatically detects if we’re running in Docker and adjusts accordingly.

Return type:

NextflowRunner

Returns:

NextflowRunner configured for testing

classmethod create(**config_kwargs)

Create a NextflowRunner with auto-configured settings.

Parameters:

**config_kwargs (Any) – Configuration parameters for NextflowConfig

Return type:

NextflowRunner

Returns:

NextflowRunner with auto-configured NextflowConfig

exception sirnaforge.pipeline.nextflow.runner.NextflowExecutionError(message, stdout='', stderr='')

Bases: Exception

Exception raised when Nextflow execution fails.

Parameters:

• message (str)

• stdout (str)

• stderr (str)

__init__(message, stdout='', stderr='')

Initialize the exception with message and optional stdout/stderr.

Parameters:

• message (str) – Error message describing what went wrong

• stdout (str) – Standard output from the failed command

• stderr (str) – Standard error from the failed command

Workflow Orchestration

siRNAforge Workflow Orchestrator.

Coordinates the complete siRNA design pipeline: 1. Transcript retrieval and validation 2. ORF validation and reporting 3. siRNA candidate generation and scoring 4. Top-N candidate selection and reporting 5. Off-target analysis with Nextflow pipeline

class sirnaforge.workflow.WorkflowConfig(output_dir, gene_query, input_fasta=None, database=DatabaseType.ENSEMBL, design_params=None, nextflow_config=None, genome_indices_override=None, genome_species=None, mirna_database='mirgenedb', mirna_species=None, transcriptome_fasta=None, transcriptome_filter=None, transcriptome_selection=None, validation_config=None, log_file=None, write_json_summary=True, num_threads=None, input_source=None, keep_nextflow_work=False, variant_config=None)

Bases: object

Configuration for the complete siRNA design workflow.

Parameters:

• output_dir (Path)

• gene_query (str)

• input_fasta (Path | None)

• database (DatabaseType)

• design_params (DesignParameters | None)

• nextflow_config (Mapping[str, Any] | None)

• genome_indices_override (str | None)

• genome_species (list[str] | None)

• mirna_database (str)

• mirna_species (Sequence[str] | None)

• transcriptome_fasta (str | None)

• transcriptome_filter (str | None)

• transcriptome_selection (ReferenceSelection | None)

• validation_config (ValidationConfig | None)

• log_file (str | None)

• write_json_summary (bool)

• num_threads (int | None)

• input_source (InputSource | None)

• keep_nextflow_work (bool)

• variant_config (VariantWorkflowConfig | None)

__init__(output_dir, gene_query, input_fasta=None, database=DatabaseType.ENSEMBL, design_params=None, nextflow_config=None, genome_indices_override=None, genome_species=None, mirna_database='mirgenedb', mirna_species=None, transcriptome_fasta=None, transcriptome_filter=None, transcriptome_selection=None, validation_config=None, log_file=None, write_json_summary=True, num_threads=None, input_source=None, keep_nextflow_work=False, variant_config=None)

Initialize workflow configuration.

Parameters:

• output_dir (Path)

• gene_query (str)

• input_fasta (Path | None)

• database (DatabaseType)

• design_params (DesignParameters | None)

• nextflow_config (Mapping[str, Any] | None)

• genome_indices_override (str | None)

• genome_species (list[str] | None)

• mirna_database (str)

• mirna_species (Sequence[str] | None)

• transcriptome_fasta (str | None)

• transcriptome_filter (str | None)

• transcriptome_selection (ReferenceSelection | None)

• validation_config (ValidationConfig | None)

• log_file (str | None)

• write_json_summary (bool)

• num_threads (int | None)

• input_source (InputSource | None)

• keep_nextflow_work (bool)

• variant_config (VariantWorkflowConfig | None)

class sirnaforge.workflow.SiRNAWorkflow(config)

Bases: object

Main workflow orchestrator for siRNA design pipeline.

Parameters:

config (WorkflowConfig)

__init__(config)

Initialize the siRNA workflow orchestrator.

Parameters:

config (WorkflowConfig)

async run_complete_workflow()

Run the complete siRNA design workflow.

Return type:

dict[str, Any]

async step1_retrieve_transcripts(progress)

Step 1: Retrieve and validate transcript sequences.

Parameters:

progress (Progress)

Return type:

list[TranscriptInfo]

async resolve_variants_step(progress)

Resolve variants for targeting or avoidance (optional workflow step).

This step runs after transcript retrieval and before siRNA design, resolving and filtering variants based on the workflow configuration.

This step is run after transcript retrieval and before ORF validation and siRNA design. Variants are resolved using ClinVar, Ensembl Variation, and/or VCF files.

Parameters:

progress (Progress) – Rich progress tracker

Return type:

list[VariantRecord]

Returns:

List of resolved VariantRecords that passed filters

async step2_validate_orfs(transcripts, progress)

Step 2: Validate ORFs and generate validation report.

Parameters:

• transcripts (list[TranscriptInfo])

• progress (Progress)

Return type:

dict[str, Any]

async step3_design_sirnas(transcripts, progress)

Step 3: Design siRNA candidates for valid transcripts.

Parallelizes per-transcript design when not running from a user-provided input FASTA, to preserve backward-compatibility with tests and monkeypatching of design_from_file. Set env SIRNAFORGE_PARALLEL_DESIGN=1 to force parallel mode.

Parameters:

• transcripts (list[TranscriptInfo])

• progress (Progress)

Return type:

DesignResult

async step4_generate_reports(design_results)

Step 4: Generate comprehensive reports.

Parameters:

design_results (DesignResult)

Return type:

None

async step5_offtarget_analysis(design_results)

Step 5: Run off-target analysis using embedded Nextflow pipeline.

Parameters:

design_results (DesignResult)

Return type:

dict[str, Any]

async sirnaforge.workflow.run_sirna_workflow(gene_query, output_dir, input_fasta=None, database='ensembl', design_mode='sirna', top_n_candidates=20, genome_species=None, genome_indices_override=None, mirna_database='mirgenedb', mirna_species=None, transcriptome_fasta=None, transcriptome_filter=None, transcriptome_selection=None, gc_min=30.0, gc_max=52.0, sirna_length=21, modification_pattern='standard_2ome', overhang='dTdT', check_off_targets=True, variant_ids=None, variant_vcf_file=None, variant_mode='avoid', variant_min_af=0.01, variant_clinvar_filters='Pathogenic,Likely pathogenic', variant_assembly='GRCh38', log_file=None, write_json_summary=True, num_threads=None, allow_transcriptome_with_input_fasta=False, default_transcriptome_sources=('ensembl_human_cdna', 'ensembl_mouse_cdna', 'ensembl_rat_cdna', 'ensembl_macaque_cdna'), keep_nextflow_work=False, nextflow_docker_image=None)

Run complete siRNA design workflow.

Parameters:

• gene_query (str) – Gene name or ID to search for

• output_dir (str) – Directory for output files

• input_fasta (str | None) – Local path or remote URI to an input FASTA file

• database (str) – Database to search (ensembl, refseq, gencode)

• design_mode (str) – Design mode (sirna or mirna)

• top_n_candidates (int) – Number of top candidates to generate

• genome_species (list[str] | None) – Species genomes for off-target analysis

• genome_indices_override (str | None) – Comma-separated species:/index_prefix overrides for off-target analysis

• mirna_database (str) – miRNA reference database identifier

• mirna_species (Sequence[str] | None) – miRNA reference species identifiers

• transcriptome_fasta (str | None) – Path or URL to transcriptome FASTA for off-target analysis

• transcriptome_filter (str | None) – Comma-separated filter names (protein_coding, canonical_only)

• transcriptome_selection (ReferenceSelection | None) – Pre-resolved transcriptome selection metadata

• gc_min (float) – Minimum GC content percentage

• gc_max (float) – Maximum GC content percentage

• sirna_length (int) – siRNA length in nucleotides

• modification_pattern (str) – Chemical modification pattern

• overhang (str) – Overhang sequence (dTdT for DNA, UU for RNA)

• check_off_targets (bool) – Perform off-target analysis stage (default: True)

• variant_ids (list[str] | None) – List of variant identifiers (rsID, chr:pos:ref:alt, or HGVS) to target or avoid

• variant_vcf_file (Path | None) – Path to VCF file containing variants to target or avoid

• variant_mode (str) – How to handle variants (avoid/target/both) - default is avoid

• variant_min_af (float) – Minimum allele frequency threshold for variant filtering (default: 0.01)

• variant_clinvar_filters (str) – Comma-separated ClinVar significance levels to include (default: Pathogenic,Likely pathogenic)

• variant_assembly (str) – Reference genome assembly for variants (only GRCh38 supported)

• log_file (str | None) – Path to centralized log file

• write_json_summary (bool) – Write logs/workflow_summary.json

• num_threads (int | None) – Optional override for design parallelism

• allow_transcriptome_with_input_fasta (bool) – Force transcriptome analysis even when using input FASTA

• default_transcriptome_sources (Sequence[str]) – Ordered list of transcriptome identifiers evaluated by default

• keep_nextflow_work (bool) – Keep Nextflow work directory symlink in output

• nextflow_docker_image (str | None) – Override Docker image used by the embedded Nextflow pipeline

Return type:

dict[str, Any]

Returns:

Dictionary with complete workflow results

async sirnaforge.workflow.run_offtarget_only_workflow(input_candidates_fasta, output_dir, genome_species=None, genome_indices_override=None, mirna_database='mirgenedb', mirna_species=None, transcriptome_fasta=None, transcriptome_filter=None, transcriptome_selection=None, log_file=None, nextflow_docker_image=None)

Run off-target-only workflow for pre-designed siRNA candidates.

This is a simplified workflow that only runs the off-target analysis stage without transcript retrieval, ORF validation, or siRNA design. It accepts pre-designed 21-nt siRNA guide sequences and runs comprehensive off-target analysis using the embedded Nextflow pipeline.

Parameters:

• input_candidates_fasta (str) – Path to FASTA file with 21-nt siRNA guide sequences

• output_dir (str) – Directory for output files

• genome_species (list[str] | None) – Species genomes for off-target analysis

• genome_indices_override (str | None) – Comma-separated species:/index_prefix overrides

• mirna_database (str) – miRNA reference database identifier

• mirna_species (Sequence[str] | None) – miRNA reference species identifiers

• transcriptome_fasta (str | None) – Path or URL to transcriptome FASTA for off-target analysis

• transcriptome_filter (str | None) – Comma-separated filter names (protein_coding, canonical_only)

• transcriptome_selection (ReferenceSelection | None) – Pre-resolved transcriptome selection metadata

• log_file (str | None) – Path to centralized log file

• nextflow_docker_image (str | None) – Override Docker image used by the embedded Nextflow pipeline

Return type:

dict[str, Any]

Returns:

Dictionary with off-target analysis results

Validation

Validation Configuration

Validation configuration and settings.

class sirnaforge.validation.config.ValidationLevel(*values)

Bases: str, Enum

Validation strictness levels.

STRICT = 'strict'

WARNING = 'warning'

DISABLED = 'disabled'

class sirnaforge.validation.config.ValidationStage(*values)

Bases: str, Enum

Pipeline stages where validation can be applied.

INPUT = 'input'

TRANSCRIPT_RETRIEVAL = 'transcript_retrieval'

ORF_ANALYSIS = 'orf_analysis'

DESIGN = 'design'

FILTERING = 'filtering'

SCORING = 'scoring'

OFF_TARGET = 'off_target'

OUTPUT = 'output'

class sirnaforge.validation.config.ValidationConfig(**data)

Bases: BaseModel

Configuration for validation system.

Parameters:

data (Any)

default_level: ValidationLevel

stage_levels: dict[ValidationStage, ValidationLevel]

validate_sequences: bool

validate_ranges: bool

validate_consistency: bool

validate_biology: bool

max_validation_errors: int

collect_all_errors: bool

batch_size: int

enable_caching: bool

get_level_for_stage(stage)

Get validation level for a specific stage.

Parameters:

stage (ValidationStage)

Return type:

ValidationLevel

is_enabled_for_stage(stage)

Check if validation is enabled for a stage.

Parameters:

stage (ValidationStage)

Return type:

bool

should_fail_on_error(stage)

Check if validation errors should cause failures.

Parameters:

stage (ValidationStage)

Return type:

bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sirnaforge.validation.config.ValidationPresets

Bases: object

Predefined validation configurations.

static development()

Configuration for development environment.

Return type:

ValidationConfig

static production()

Configuration for production environment.

Return type:

ValidationConfig

static testing()

Configuration for testing environment.

Return type:

ValidationConfig

static performance()

Configuration optimized for performance.

Return type:

ValidationConfig

Validation Middleware

Validation middleware for integrating validation into the siRNA design workflow.

class sirnaforge.validation.middleware.ValidationReport(stage)

Bases: object

Comprehensive validation report for a workflow stage.

Parameters:

stage (ValidationStage)

__init__(stage)

Initialize validation report.

Parameters:

stage (ValidationStage)

add_item_result(result)

Add validation result for an individual item.

Parameters:

result (ValidationResult)

Return type:

None

finalize()

Finalize the report and calculate summary statistics.

Return type:

None

to_dict()

Convert report to dictionary.

Return type:

dict[str, Any]

class sirnaforge.validation.middleware.ValidationMiddleware(config)

Bases: object

Middleware for integrating validation throughout the workflow.

Parameters:

config (ValidationConfig)

__init__(config)

Initialize validation middleware.

Parameters:

config (ValidationConfig)

validate_input_parameters(params)

Validate input design parameters.

Parameters:

params (DesignParameters)

Return type:

ValidationReport

validate_transcripts(transcripts)

Validate transcript data after retrieval.

Parameters:

transcripts (list[TranscriptInfo])

Return type:

ValidationReport

validate_design_results(design_result)

Validate siRNA design results.

Parameters:

design_result (DesignResult)

Return type:

ValidationReport

validate_dataframe_output(df, schema_type)

Validate DataFrame output against pandera schemas.

Parameters:

• df (DataFrame)

• schema_type (str)

Return type:

ValidationReport

validate_transcript_id_consistency(transcripts, candidates, orf_data=None)

Validate consistency of transcript IDs across datasets.

Parameters:

• transcripts (list[TranscriptInfo])

• candidates (list[SiRNACandidate])

• orf_data (DataFrame | None)

Return type:

ValidationReport

save_validation_report(output_path)

Save comprehensive validation report.

Parameters:

output_path (Path)

Return type:

None

Validation Utilities

Validation utilities for data consistency and cross-validation.

class sirnaforge.validation.utils.ValidationResult(is_valid=True)

Bases: object

Container for validation results.

Parameters:

is_valid (bool)

__init__(is_valid=True)

Initialize validation result container.

Parameters:

is_valid (bool)

add_error(message)

Add a validation error.

Parameters:

message (str)

Return type:

None

add_warning(message)

Add a validation warning.

Parameters:

message (str)

Return type:

None

add_metadata(key, value)

Add metadata to the result.

Parameters:

• key (str)

• value (Any)

Return type:

None

merge(other)

Merge another validation result into this one.

Parameters:

other (ValidationResult)

Return type:

None

summary()

Get a summary of validation results.

Return type:

dict[str, Any]

class sirnaforge.validation.utils.ValidationUtils

Bases: object

Utility functions for data validation.

static validate_nucleotide_sequence(sequence, allow_ambiguous=True)

Validate nucleotide sequence composition.

Parameters:

• sequence (str)

• allow_ambiguous (bool)

Return type:

ValidationResult

static validate_sirna_length(sequence)

Validate siRNA sequence length.

Parameters:

sequence (str)

Return type:

ValidationResult

static validate_parameter_consistency(params)

Validate design parameter consistency.

Parameters:

params (DesignParameters)

Return type:

ValidationResult

static validate_candidate_consistency(candidate)

Validate siRNA candidate internal consistency.

Parameters:

candidate (SiRNACandidate)

Return type:

ValidationResult

static validate_dataframe_schema(df, schema_type)

Validate DataFrame against appropriate pandera schema.

Parameters:

• df (DataFrame)

• schema_type (str)

Return type:

ValidationResult

static validate_transcript_ids_consistency(candidate_ids, orf_ids, transcript_ids)

Validate consistency of transcript IDs across datasets.

Parameters:

• candidate_ids (set[str])

• orf_ids (set[str])

• transcript_ids (set[str])

Return type:

ValidationResult

static validate_biological_constraints(candidate)

Validate bioinformatics-specific constraints.

Parameters:

candidate (SiRNACandidate)

Return type:

ValidationResult

static cross_validate_pydantic_pandera()

Cross-validate Pydantic model constraints with Pandera schema constraints.

Return type:

ValidationResult

Utilities

Logging Utilities

Logging utilities for siRNAforge toolkit.

This module provides a single point to configure logging for both console and an optional centralized log file. Call configure_logging once at application startup (CLI entrypoint) to enable file logging. Individual modules should use get_logger(__name__) to obtain a configured logger.

sirnaforge.utils.logging_utils.configure_logging(level=None, log_file=None)

Configure root logger: console + optional rotating file handler.

Parameters:

• level (str | None) – Logging level name (DEBUG/INFO/WARNING/ERROR). If None, uses INFO.

• log_file (str | None) – Path to central log file. If None, will read env var SIRNAFORGE_LOG_FILE. If still None, no file handler is added.

Return type:

None

sirnaforge.utils.logging_utils.get_logger(name, level=None)

Get a logger with standard configuration.

This will return a child logger of the root logger configured by configure_logging. For scripts that don’t call configure_logging, get_logger will still set a console handler on first use.

Parameters:

• name (str)

• level (str | None)

Return type:

Logger

Modification Patterns

Utility functions for applying chemical modification patterns to siRNA candidates.

This module provides functions to apply standard modification patterns to siRNA candidates during the design workflow, enabling automated annotation of chemical modifications for downstream synthesis and analysis.

sirnaforge.utils.modification_patterns.apply_standard_2ome_pattern(sequence)

Apply standard alternating 2’-O-methyl pattern.

This is the industry-standard pattern providing balanced nuclease resistance and RISC loading efficiency.

Parameters:

sequence (str) – RNA sequence to modify

Return type:

list[ChemicalModification]

Returns:

List containing one ChemicalModification with alternating positions

sirnaforge.utils.modification_patterns.apply_minimal_terminal_pattern(sequence)

Apply minimal terminal modifications for cost-effective protection.

Modifies only the 3’ terminal positions to provide basic nuclease resistance while minimizing synthesis cost.

Parameters:

sequence (str) – RNA sequence to modify

Return type:

list[ChemicalModification]

Returns:

List containing one ChemicalModification with terminal positions

sirnaforge.utils.modification_patterns.apply_maximal_stability_pattern(sequence)

Apply maximal stability pattern for in vivo applications.

Fully modified pattern similar to FDA-approved therapeutics, providing maximum nuclease resistance and extended serum half-life.

Parameters:

sequence (str) – RNA sequence to modify

Return type:

list[ChemicalModification]

Returns:

List containing ChemicalModifications (2OMe on all positions, PS at terminals)

sirnaforge.utils.modification_patterns.get_modification_pattern(pattern_name, sequence)

Get modification pattern by name.

Parameters:

• pattern_name (str) – Name of the pattern (standard_2ome, minimal_terminal, maximal_stability, none)

• sequence (str) – RNA sequence to apply pattern to

Return type:

list[ChemicalModification]

Returns:

List of ChemicalModification objects

Raises:

ValueError – If pattern_name is not recognized

sirnaforge.utils.modification_patterns.apply_modifications_to_candidate(candidate, pattern_name='standard_2ome', overhang='dTdT', target_gene=None)

Apply chemical modifications to a siRNA candidate.

This function annotates both guide and passenger strands with the specified modification pattern and overhang, updating the candidate’s metadata fields.

Parameters:

• candidate (SiRNACandidate) – SiRNACandidate to annotate

• pattern_name (str) – Modification pattern to apply (default: standard_2ome)

• overhang (str) – Overhang sequence (default: dTdT)

• target_gene (str | None) – Optional target gene name for metadata

Return type:

SiRNACandidate

Returns:

Updated SiRNACandidate with modification metadata

sirnaforge.utils.modification_patterns.get_modification_summary(candidate)

Get a summary of modifications for a candidate.

Parameters:

candidate (SiRNACandidate) – SiRNACandidate with modification metadata

Return type:

dict[str, str]

Returns:

Dictionary with modification summary info

Resource Resolver

Utilities for resolving user-provided input resources.

Supports downloading transcript FASTA files from remote locations and normalises them into local paths that the workflow can consume.

class sirnaforge.utils.resource_resolver.InputSource(original, local_path, source_type, downloaded, size_bytes, sha256=None)

Bases: object

Normalized representation of a workflow input resource.

Parameters:

• original (str)

• local_path (Path)

• source_type (str)

• downloaded (bool)

• size_bytes (int)

• sha256 (str | None)

original: str

local_path: Path

source_type: str

downloaded: bool

size_bytes: int

sha256: str | None = None

property stem: str

Return the filesystem stem for the local resource.

__init__(original, local_path, source_type, downloaded, size_bytes, sha256=None)

Parameters:

• original (str)

• local_path (Path)

• source_type (str)

• downloaded (bool)

• size_bytes (int)

• sha256 (str | None)

sirnaforge.utils.resource_resolver.resolve_input_source(input_location, destination_root, *, timeout=30.0)

Resolve a workflow input location into a local path.

Parameters:

• input_location (str) – Raw string provided by the user (path or URI).

• destination_root (Path) – Directory where downloaded inputs should be stored.

• timeout (float) – Timeout for remote downloads in seconds.

Return type:

InputSource

Returns:

InputSource describing the normalized local resource.

Raises:

• FileNotFoundError – If a local file doesn’t exist.

• ValueError – If the URI scheme is unsupported.

• httpx.HTTPStatusError – If the remote download fails with non-2xx status.

Chemical Modifications

Helper functions for working with siRNA chemical modifications metadata.

This module provides utilities for: - Parsing FASTA headers to extract modification metadata - Loading metadata from JSON sidecar files - Encoding/decoding modification annotations

sirnaforge.modifications.parse_chem_mods(chem_mods_str)

Parse ChemMods field from FASTA header.

Parameters:

chem_mods_str (str) – String like “2OMe(1,4,6,11)+2F()”

Return type:

list[ChemicalModification]

Returns:

List of ChemicalModification objects

sirnaforge.modifications.parse_provenance(prov_str, url=None)

Parse Provenance field from FASTA header.

Parameters:

• prov_str (str) – String like “Patent:US10060921B2”

• url (str | None) – Optional URL string

Return type:

Provenance | None

Returns:

Provenance object or None

sirnaforge.modifications.parse_header(record)

Parse FASTA header to extract metadata.

Parameters:

record (SeqRecord) – BioPython SeqRecord from FASTA file

Return type:

dict[str, Any]

Returns:

Dictionary with parsed metadata fields

sirnaforge.modifications.load_metadata(json_path)

Load and validate metadata from JSON sidecar file using Pydantic.

Parameters:

json_path (str | Path) – Path to JSON file containing metadata

Return type:

dict[str, StrandMetadata]

Returns:

Dictionary mapping strand IDs to StrandMetadata objects

Raises:

ValidationError – If JSON data doesn’t match StrandMetadata schema

sirnaforge.modifications.merge_metadata_into_fasta(fasta_path, metadata_path, output_path)

Merge metadata from JSON into FASTA headers.

Uses Pydantic for automatic validation of metadata.

Parameters:

• fasta_path (str | Path) – Input FASTA file

• metadata_path (str | Path) – JSON file with metadata

• output_path (str | Path) – Output FASTA file with updated headers

Return type:

int

Returns:

Number of sequences with metadata applied

Raises:

ValidationError – If metadata doesn’t match StrandMetadata schema

sirnaforge.modifications.save_metadata_json(metadata_dict, output_path)

Save strand metadata to JSON file using Pydantic serialization.

Parameters:

• metadata_dict (dict[str, StrandMetadata]) – Dictionary mapping strand IDs to StrandMetadata objects

• output_path (str | Path) – Path to output JSON file

Return type:

None