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.

Configuration

Reference Policy

Reference/default resolution utilities for workflow inputs.

class sirnaforge.config.reference_policy.ReferenceChoice(value: str | None, state: ReferenceState, reason: str)

Bases: object

Normalized representation of a resolved reference input.

value: str | None

state: ReferenceState

reason: str

property enabled: bool

Return True when a usable reference has been selected.

static explicit(value: str, reason: str = 'user-provided') → ReferenceChoice

Create an explicit user-selected reference choice.

static default(value: str, reason: str = 'auto-selected') → ReferenceChoice

Create a default-sourced reference choice.

static disabled(reason: str) → ReferenceChoice

Create a disabled reference choice with context.

to_metadata() → dict[str, str | None | bool]

Return a serializable snapshot for logs/JSON summaries.

__init__(value: str | None, state: ReferenceState, reason: str) → None

class sirnaforge.config.reference_policy.ReferenceSelection(choices: tuple[~sirnaforge.config.reference_policy.ReferenceChoice, ...]=<factory>, disabled_reason: str | None = None)

Bases: object

Container describing zero or more resolved references.

choices: tuple[ReferenceChoice, ...]

disabled_reason: str | None = None

property enabled: bool

Return True when at least one reference is configured.

static disabled(reason: str) → ReferenceSelection

Create a disabled selection with a descriptive reason.

to_metadata() → dict[str, object]

Render selection metadata for logging.

__init__(choices: tuple[~sirnaforge.config.reference_policy.ReferenceChoice, ...]=<factory>, disabled_reason: str | None = None) → None

class sirnaforge.config.reference_policy.ReferencePolicyResolver(spec: WorkflowInputSpec)

Bases: object

Resolve workflow defaults while preserving intent metadata.

__init__(spec: WorkflowInputSpec)

Create a resolver for a specific workflow input specification.

resolve_transcriptomes() → ReferenceSelection

Return one or more transcriptome references.

class sirnaforge.config.reference_policy.ReferenceState(*values)

Bases: str, Enum

Describe how a reference input was selected.

EXPLICIT = 'explicit'

DEFAULT = 'default'

DISABLED = 'disabled'

class sirnaforge.config.reference_policy.WorkflowInputSpec(input_fasta: str | None = None, transcriptome_argument: str | None = None, default_transcriptomes: Sequence[str] = <factory>, design_only: bool = False, allow_transcriptome_for_input_fasta: bool = False)

Bases: object

Raw workflow inputs prior to policy resolution.

input_fasta: str | None = None

transcriptome_argument: str | None = None

default_transcriptomes: Sequence[str]

design_only: bool = False

allow_transcriptome_for_input_fasta: bool = False

__init__(input_fasta: str | None = None, transcriptome_argument: str | None = None, default_transcriptomes: Sequence[str] = <factory>, design_only: bool = False, allow_transcriptome_for_input_fasta: bool = False) → None

sirnaforge.config.reference_policy.render_reference_selection_label(selection: ReferenceSelection) → str

Render a stable, human-readable label for CLI/config summaries.

Command Line Interface

Modern CLI for siRNAforge using Typer and Rich.

sirnaforge.cli.patched_init(self: Console, *args: Any, **kwargs: Any) → None

Force simplified terminal capabilities for deterministic CI output.

class sirnaforge.cli.TranscriptLike(*args, **kwargs)

Bases: Protocol

Minimal transcript-like interface used by CLI filters.

transcript_type: str | None

is_canonical: bool

__init__(*args, **kwargs)

sirnaforge.cli.filter_transcripts(transcripts: list[TTranscript], include_types: list[str] | None = None, exclude_types: list[str] | None = None, canonical_only: bool = False) → list[TTranscript]

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: list[TranscriptInfo], gene_name: str, output_dir: Path | str | None = None) → tuple[Path | None, int]

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: str = <typer.models.ArgumentInfo object>, output: Path = <typer.models.OptionInfo object>, database: str = <typer.models.OptionInfo object>, all_databases: bool = <typer.models.OptionInfo object>, fallback: bool = <typer.models.OptionInfo object>, no_sequence: bool = <typer.models.OptionInfo object>, canonical_only: bool = <typer.models.OptionInfo object>, extract_canonical: bool = <typer.models.OptionInfo object>, transcript_types: str = <typer.models.OptionInfo object>, exclude_types: str = <typer.models.OptionInfo object>, verbose: bool = <typer.models.OptionInfo object>) → None

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.

sirnaforge.cli.workflow(gene_query: str = <typer.models.ArgumentInfo object>, input_fasta: str | None = <typer.models.OptionInfo object>, output_dir: Path = <typer.models.OptionInfo object>, database: str = <typer.models.OptionInfo object>, design_mode: str = <typer.models.OptionInfo object>, zfn_subfinger_mutation: list[str] = <typer.models.OptionInfo object>, zfn_max_mismatches_per_subfinger: int | None = <typer.models.OptionInfo object>, zfn_max_substitutions_overall: int | None = <typer.models.OptionInfo object>, zfn_left_half_site: str | None = <typer.models.OptionInfo object>, zfn_right_half_site: str | None = <typer.models.OptionInfo object>, zfn_search_space: str | None = <typer.models.OptionInfo object>, zfn_search_space_index: str | None = <typer.models.OptionInfo object>, zfn_search_backend: ZFNSearchBackend = <typer.models.OptionInfo object>, zfn_algorithm: ZFNAlgorithm = <typer.models.OptionInfo object>, zfn_dimer_mode: DimerMode = <typer.models.OptionInfo object>, zfn_spacer_lengths: str = <typer.models.OptionInfo object>, zfn_max_mismatches: int = <typer.models.OptionInfo object>, zfn_window_stride: int | None = <typer.models.OptionInfo object>, zfn_top_n_sites: int | None = <typer.models.OptionInfo object>, zfn_report_n_sites: int | None = <typer.models.OptionInfo object>, cores: int | None = <typer.models.OptionInfo object>, zfn_annotation: str | None = <typer.models.OptionInfo object>, top_n_candidates: int | None = <typer.models.OptionInfo object>, species: str = <typer.models.OptionInfo object>, query_species: str | None = <typer.models.OptionInfo object>, mirna_db: str = <typer.models.OptionInfo object>, mirna_species: str | None = <typer.models.OptionInfo object>, transcriptome_fasta: str | None = <typer.models.OptionInfo object>, transcriptome_filter: str | None = <typer.models.OptionInfo object>, offtarget_indices: str | None = <typer.models.OptionInfo object>, gc_min: float = <typer.models.OptionInfo object>, gc_max: float = <typer.models.OptionInfo object>, sirna_length: int = <typer.models.OptionInfo object>, modification_pattern: str = <typer.models.OptionInfo object>, overhang: str = <typer.models.OptionInfo object>, skip_off_targets: bool = <typer.models.OptionInfo object>, snp: list[str] = <typer.models.OptionInfo object>, snp_file: Path | None = <typer.models.OptionInfo object>, variant_mode: VariantMode = <typer.models.OptionInfo object>, min_af: float = <typer.models.OptionInfo object>, clinvar_filter_levels: str = <typer.models.OptionInfo object>, variant_assembly: str = <typer.models.OptionInfo object>, verbose: bool = <typer.models.OptionInfo object>, log_file: Path | None = <typer.models.OptionInfo object>, nextflow_docker_image: str | None = <typer.models.OptionInfo object>, max_hits: int | None = <typer.models.OptionInfo object>, max_off_targets: int | None = <typer.models.OptionInfo object>, json_summary: bool = <typer.models.OptionInfo object>) → None

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.

sirnaforge.cli.offtarget(input_candidates_fasta: Path = <typer.models.OptionInfo object>, output_dir: Path = <typer.models.OptionInfo object>, species: str = <typer.models.OptionInfo object>, query_species: str | None = <typer.models.OptionInfo object>, mirna_db: str = <typer.models.OptionInfo object>, mirna_species: str | None = <typer.models.OptionInfo object>, transcriptome_fasta: str | None = <typer.models.OptionInfo object>, transcriptome_filter: str | None = <typer.models.OptionInfo object>, offtarget_indices: str | None = <typer.models.OptionInfo object>, verbose: bool = <typer.models.OptionInfo object>, log_file: Path | None = <typer.models.OptionInfo object>, nextflow_docker_image: str | None = <typer.models.OptionInfo object>) → None

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.

sirnaforge.cli.zfn(output_dir: Path = <typer.models.OptionInfo object>, zfn_subfinger_mutation: list[str] = <typer.models.OptionInfo object>, zfn_max_mismatches_per_subfinger: int | None = <typer.models.OptionInfo object>, zfn_max_substitutions_overall: int | None = <typer.models.OptionInfo object>, zfn_left_half_site: str = <typer.models.OptionInfo object>, zfn_right_half_site: str = <typer.models.OptionInfo object>, zfn_search_space: str | None = <typer.models.OptionInfo object>, zfn_search_space_index: str | None = <typer.models.OptionInfo object>, zfn_search_backend: ZFNSearchBackend = <typer.models.OptionInfo object>, zfn_algorithm: ZFNAlgorithm = <typer.models.OptionInfo object>, zfn_dimer_mode: DimerMode = <typer.models.OptionInfo object>, zfn_spacer_lengths: str = <typer.models.OptionInfo object>, zfn_max_mismatches: int = <typer.models.OptionInfo object>, zfn_window_stride: int | None = <typer.models.OptionInfo object>, zfn_top_n_sites: int | None = <typer.models.OptionInfo object>, zfn_report_n_sites: int | None = <typer.models.OptionInfo object>, cores: int | None = <typer.models.OptionInfo object>, zfn_annotation: str | None = <typer.models.OptionInfo object>, verbose: bool = <typer.models.OptionInfo object>, log_file: Path | None = <typer.models.OptionInfo object>, nextflow_docker_image: str | None = <typer.models.OptionInfo object>, json_summary: bool = <typer.models.OptionInfo object>) → None

Evaluate a ZFN pair and run exhaustive genome-wide off-target search (EXPERIMENTAL).

sirnaforge.cli.design(input_file: Path = <typer.models.ArgumentInfo object>, output: Path = <typer.models.OptionInfo object>, design_mode: str = <typer.models.OptionInfo object>, length: int = <typer.models.OptionInfo object>, top_n: int | None = <typer.models.OptionInfo object>, gc_min: float = <typer.models.OptionInfo object>, gc_max: float = <typer.models.OptionInfo object>, max_poly_runs: int = <typer.models.OptionInfo object>, genome_index: Path | None = <typer.models.OptionInfo object>, snp_file: Path | None = <typer.models.OptionInfo object>, skip_structure: bool = <typer.models.OptionInfo object>, skip_off_targets: bool = <typer.models.OptionInfo object>, modification_pattern: str = <typer.models.OptionInfo object>, overhang: str = <typer.models.OptionInfo object>, verbose: bool = <typer.models.OptionInfo object>) → None

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.

sirnaforge.cli.validate(input_file: Path = <typer.models.ArgumentInfo object>) → None

Validate a FASTA file and report basic statistics.

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

sirnaforge.cli.version() → None

Show CLI version and author information.

sirnaforge.cli.config() → None

Print the default design parameter values.

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

Inspect and clear the unified reference cache.

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

exception sirnaforge.cli.SequencesShowError

Bases: RuntimeError

Raised when sequence display/formatting input is invalid.

sirnaforge.cli.sequences_show(input_file: Path = <typer.models.ArgumentInfo object>, sequence_id: str | None = <typer.models.OptionInfo object>, format: str = <typer.models.OptionInfo object>) → None

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.

sirnaforge.cli.sequences_annotate(input_fasta: Path = <typer.models.ArgumentInfo object>, metadata_json: Path = <typer.models.ArgumentInfo object>, output: Path | None = <typer.models.OptionInfo object>, verbose: bool = <typer.models.OptionInfo object>) → None

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.

sirnaforge.cli.internal_zfn_make_shards(genome_fasta: Path = <typer.models.OptionInfo object>, left_half_site: str = <typer.models.OptionInfo object>, right_half_site: str = <typer.models.OptionInfo object>, spacer_lengths: str = <typer.models.OptionInfo object>, max_mismatches: int = <typer.models.OptionInfo object>, sharding_enabled: str = <typer.models.OptionInfo object>, shard_chunk_mb: float = <typer.models.OptionInfo object>, shard_overlap_bp: int = <typer.models.OptionInfo object>, shard_chromosomes: str = <typer.models.OptionInfo object>, output: Path = <typer.models.OptionInfo object>) → None

Build ZFN shard manifest for Nextflow execution.

sirnaforge.cli.internal_zfn_build_search_index(genome_fasta: Path = <typer.models.OptionInfo object>, search_backend: ZFNSearchBackend = <typer.models.OptionInfo object>, output_dir: Path | None = <typer.models.OptionInfo object>) → None

Build a persisted ZFN search-space index bundle for indexed backends.

sirnaforge.cli.internal_zfn_search_shard(shard_id: str = <typer.models.OptionInfo object>, shard_chrom: str = <typer.models.OptionInfo object>, scan_start_1: int = <typer.models.OptionInfo object>, scan_end_1: int = <typer.models.OptionInfo object>, core_start_1: int | None = <typer.models.OptionInfo object>, core_end_1: int | None = <typer.models.OptionInfo object>, shard_max_mismatches: int = <typer.models.OptionInfo object>, left_half_site: str = <typer.models.OptionInfo object>, right_half_site: str = <typer.models.OptionInfo object>, genome_fasta: Path = <typer.models.OptionInfo object>, search_backend: ZFNSearchBackend = <typer.models.OptionInfo object>, search_space_index: Path | None = <typer.models.OptionInfo object>, algorithm: ZFNAlgorithm = <typer.models.OptionInfo object>, dimer_mode: DimerMode = <typer.models.OptionInfo object>, spacer_lengths: str = <typer.models.OptionInfo object>, annotation_file: Path | None = <typer.models.OptionInfo object>, output_sites_csv: Path = <typer.models.OptionInfo object>, output_summary_json: Path = <typer.models.OptionInfo object>) → None

Run one shard-scoped ZFN search and emit shard artifacts.

sirnaforge.cli.internal_zfn_aggregate_shards(shard_csv_glob: str = <typer.models.OptionInfo object>, output_sites_csv: Path = <typer.models.OptionInfo object>, output_summary_json: Path = <typer.models.OptionInfo object>) → None

Aggregate shard-level ZFN outputs into final ranked outputs.

Core Modules

Design Engine

Core siRNA design algorithms and functionality.

class sirnaforge.core.design.SiRNADesigner(parameters: DesignParameters)

Bases: object

Main siRNA design engine following the algorithm specification.

__init__(parameters: DesignParameters) → None

Initialize designer with given parameters.

design_from_file(input_file: str) → DesignResult

Design siRNAs from input FASTA file.

design_from_sequence(sequence: str, transcript_id: str = 'seq1') → DesignResult

Design siRNAs from a single sequence.

static stamp_repeat_verdict(candidate: SiRNACandidate, observations: dict[str, RepeatObservation]) → None

Stamp repeat metadata and verdict on a single candidate if its guide is flagged.

The REPEAT_ELEMENT verdict is applied only if the candidate is currently passing (passes_filters is True or PASS). A candidate that already failed for another reason (GC, asymmetry, etc.) retains its earlier verdict — precedence is: existing failure > REPEAT_ELEMENT > PASS.

Parameters:

• candidate – Candidate to potentially flag.

• observations – Mapping from normalized guide sequence to RepeatObservation.

sirnaforge.core.design.mirna_max_biogenesis_bonus(scoring_weights: Mapping[str, float] | None = None) → float

Maximum attainable miRNA biogenesis bonus: the divisor that puts a miRNA run on one scale.

Exposed as a function so post-screen rescoring can recover the divisor for a candidate that never passed through MiRNADesigner._score_candidates (and therefore carries no MIRNA_BONUS_MAX_KEY), instead of leaving that row on an undivided scale.

sirnaforge.core.design.apply_mirna_biogenesis_bonus(base_score: float, mirna_bonus: float, max_mirna_bonus: float) → float

Fold the miRNA biogenesis bonus into a 0-100 composite score.

The bonuses widen the attainable range, so rescale by the maximum attainable total instead of clamping: clamping parked every strong candidate at exactly 100.0 and erased the ranking at the top. Order is preserved, since this is monotone in (base score + bonus).

class sirnaforge.core.design.MiRNADesigner(parameters: DesignParameters)

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

__init__(parameters: DesignParameters) → None

Initialize miRNA designer with miRNA-specific config validation.

Thermodynamics Analysis

Thermodynamic calculations for siRNA design using ViennaRNA.

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

Bases: object

Calculate thermodynamic properties for siRNA candidates using ViennaRNA.

__init__(temperature: float = 37.0)

Initialize thermodynamic calculator.

Parameters:

temperature – Temperature in Celsius for calculations

calculate_duplex_stability(guide: str, passenger: str) → float

Calculate duplex stability (deltaG) using ViennaRNA.

Both strands are supplied 5’->3’ and are already complementary, so neither is reverse-complemented here: ViennaRNA’s & cofold notation pairs the two strands antiparallel on its own.

calculate_asymmetry_score(candidate: SiRNACandidate) → tuple[float, float, float]

Calculate thermodynamic asymmetry score using ViennaRNA.

Duplex ends are antiparallel, so the guide 5’ end pairs with the passenger 3’ end and vice versa. Both windows are END_WINDOW_NT long so the two ΔG values are directly comparable.

Returns:

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

calculate_target_accessibility(target_sequence: str, start_pos: int, sirna_length: int) → tuple[float, float]

Calculate target site accessibility using ViennaRNA.

Parameters:

• target_sequence – Full target mRNA sequence

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

• sirna_length – Length of siRNA

Returns:

Tuple of (average_unpaired_probability, mfe)

calculate_melting_temperature(guide: str, passenger: str) → float

Calculate duplex melting temperature in °C.

Uses the RNA nearest-neighbour parameters of Xia et al. (1998) via Biopython. Tm needs ΔH and ΔS separately, which ViennaRNA’s MFE does not provide, so scaling ΔG cannot produce a physical Tm: at the corrected duplex ΔG of a 21mer (about -39 kcal/mol) the previous 37 + 2 * -ΔG approximation returned 101-124 °C.

Parameters:

• guide – Guide strand, 5’->3’

• passenger – Passenger strand, 5’->3’ (expected reverse complement of guide)

static meets_asymmetry_threshold(asymmetry_score: float, threshold: float) → bool

Check an already-computed asymmetry score against a threshold.

Callers that have just computed the score use this instead of is_thermodynamically_favorable() to avoid re-folding both duplex ends.

is_thermodynamically_favorable(candidate: SiRNACandidate, threshold: float = 0.65) → bool

Check if candidate meets thermodynamic asymmetry threshold.

calculate_secondary_structure(sequence: str) → tuple[str, float, float]

Calculate secondary structure for a sequence.

Returns:

Tuple of (structure, mfe, paired_fraction)

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 transcriptome alignments and supports in-process miRNA seed scanning. Optimized for both standalone use and parallelized Nextflow workflows.

class sirnaforge.core.off_target.BwaAnalyzer(index_prefix: str | Path, mode: str = 'transcriptome', seed_length: int = 12, min_score: int = 15, max_hits: int | None = None, seed_start: int = 2, seed_end: int = 8)

Bases: object

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

__init__(index_prefix: str | Path, mode: str = 'transcriptome', seed_length: int = 12, min_score: int = 15, max_hits: int | None = None, seed_start: int = 2, seed_end: int = 8)

Initialize BWA-MEM2 analyzer.

Parameters:

• index_prefix – Path to BWA index

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

• seed_length – BWA seed length parameter

• min_score – Minimum alignment score

• max_hits – Maximum hits to return (None = no limit / exhaustive)

• seed_start – Seed region start (1-based)

• seed_end – Seed region end (1-based)

analyze_sequences(sequences: dict[str, str]) → list[dict[str, Any]]

Run BWA-MEM2 analysis on sequences.

Parameters:

sequences – Dictionary of sequence name -> sequence

Returns:

List of alignment dictionaries

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

Bases: object

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

__init__(species: str, transcriptome_path: str | Path | None = None, mirna_path: str | Path | None = None, transcriptome_index: str | Path | None = None, mirna_index: str | Path | None = None)

Initialize the off-target analysis manager.

analyze_mirna_off_targets(sequences: dict[str, str] | str | Path, output_prefix: str | Path) → tuple[Path, Path]

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

analyze_transcriptome_off_targets(sequences: dict[str, str] | str | Path, output_prefix: str | Path) → tuple[Path, Path]

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

analyze_sirna_candidate(candidate: SiRNACandidate) → dict[str, Any]

Analyze a single siRNA candidate for off-targets.

class sirnaforge.core.off_target.MiRNASeedBackend(*values)

Bases: str, Enum

Internal backend options for miRNA seed scanning.

BWA = 'bwa'

EXHAUSTIVE_PYTHON = 'exhaustive_python'

PYAHOCORASICK = 'pyahocorasick'

sirnaforge.core.off_target.create_temp_fasta(sequences: dict[str, str]) → str

Create temporary FASTA file from sequences.

sirnaforge.core.off_target.validate_and_write_sequences(input_file: str, output_file: str, expected_length: int = 21) → tuple[int, int, list[str]]

Validate siRNA sequences and write valid ones to output file.

sirnaforge.core.off_target.build_bwa_index(fasta_file: str | Path, index_prefix: str | Path) → Path

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

sirnaforge.core.off_target.validate_sirna_sequences(sequences: dict[str, str], expected_length: int = 21) → tuple[dict[str, str], dict[str, str], list[str]]

Validate siRNA sequences using existing FastaUtils.

sirnaforge.core.off_target.parse_fasta_file(fasta_file: str | Path) → dict[str, str]

Parse FASTA file using existing FastaUtils.

sirnaforge.core.off_target.write_fasta_file(sequences: dict[str, str], output_file: str) → None

Write sequences to FASTA file using existing FastaUtils.

sirnaforge.core.off_target.check_tool_availability(tool: str) → bool

Check if external tool is available.

sirnaforge.core.off_target.validate_index_files(index_prefix: str | Path, tool: str = 'bwa') → bool

Validate that index files exist for given tool.

sirnaforge.core.off_target.mirna_seed_hit_identity(hit: dict[str, Any], *, coord_is_one_based: bool = False) → tuple[str, str, int, int, int, float]

Return the semantic identity tuple used for backend comparison.

This normalizes the current in-process seed-scan rows and the existing BWA-derived rows onto the same comparison contract.

sirnaforge.core.off_target.normalize_mirna_seed_hit(hit: dict[str, Any], *, coord_is_one_based: bool = False) → _NormalizedMiRNASeedHit

Normalize backend-specific miRNA seed hit metadata onto one adapter/parity contract.

sirnaforge.core.off_target.run_bwa_alignment_analysis(candidates_file: str | Path, index_prefix: str | Path, species: str, output_dir: str | Path, max_hits: int | None = None, bwa_k: int = 12, bwa_T: int = 15, seed_start: int = 2, seed_end: int = 8) → Path

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 – Path to FASTA file with candidate sequences

• index_prefix – Path to BWA-MEM2 index prefix

• species – Species identifier

• output_dir – Directory to write results

• max_hits – Maximum hits to report per candidate (None = no limit / exhaustive)

• bwa_k – BWA seed length parameter

• bwa_T – BWA minimum score threshold

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

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

Returns:

Path to output directory containing results

sirnaforge.core.off_target.aggregate_offtarget_results(results_dir: str | Path, output_dir: str | Path, genome_species: str) → Path

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 – Directory containing individual analysis results

• output_dir – Directory to write aggregated results

• genome_species – Comma-separated list of genome species analyzed

Returns:

Path to output directory containing aggregated results

sirnaforge.core.off_target.scan_mirna_seed_matches(sequences: dict[str, str], mirna_sequences: dict[str, str], *, backend: MiRNASeedBackend | str = MiRNASeedBackend.PYAHOCORASICK, seed_start: int = 2, seed_end: int = 8, max_mismatches: int = 2, max_hits: int | None = None) → list[dict[str, Any]]

Scan miRNA FASTA records using an in-process seed-scanning backend.

The returned rows preserve the existing internal hit shape used by miRNA analysis, so callers can continue adapting them to DataFrame-based outputs.

sirnaforge.core.off_target.run_mirna_seed_analysis(candidates_file: str | Path, candidate_id: str, mirna_db: str, mirna_species: list[str], output_dir: str | Path, backend: MiRNASeedBackend | str = MiRNASeedBackend.PYAHOCORASICK, seed_start: int = 2, seed_end: int = 8) → Path

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.

The scan produces raw alignments: the guide seed window placed at every position along each miRNA. Only alignments where the guide seed lands on the miRNA’s own seed region (0-based coord == seed_start - 1) are counted as hits in the filtered outputs and summary total_hits; perfect matches in non-seed regions are retained in the *_raw files but are not real miRNA seed off-targets.

Parameters:

• candidates_file – Path to FASTA file with candidate sequences

• candidate_id – Candidate identifier

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

• mirna_species – List of species to analyze against

• output_dir – Directory to write results

• backend – miRNA seed backend to use for analysis (pyahocorasick by default)

• seed_start – Seed region start position (1-based, default 2)

• seed_end – Seed region end position (1-based, default 8)

Returns:

Path to output directory containing results

sirnaforge.core.off_target.aggregate_mirna_results(results_dir: str | Path, output_dir: str | Path, mirna_db: str, mirna_species: str) → Path

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 – Directory containing individual miRNA analysis results

• output_dir – Directory to write aggregated results

• mirna_db – miRNA database used for analysis

• mirna_species – Comma-separated list of species analyzed

Returns:

Path to output directory containing aggregated results

ZFN Design

Warning

EXPERIMENTAL. Every sirnaforge.zfn.* and sirnaforge.models.zfn API below is experimental in 0.6.0 with known unfixed defects, tracked in #82. Do not use ZFN output for any decision without independent validation. Calling sirnaforge.zfn.design.ZFNDesigner.evaluate_pair() logs the full notice at WARNING on first use in a process. Two defects that change nothing visible in the returned objects: ZFNCandidate.worst_site_score / best_offtarget_score are inverted (minimum and maximum site score respectively, whereas the highest-scoring off-target is the most dangerous one), and right_half_site must be supplied as the reverse complement of its published plus-strand text or it matches nothing. See ZFN Module Guide.

Experimental-status notice for the ZFN arm.

This lives in its own module rather than in sirnaforge.zfn.__init__ because the package __init__ imports design, and design has to emit the notice too – reaching back into the package from one of its own submodules would be circular.

The notice is emitted at most once per process. Every ZFN entry point calls emit_zfn_experimental_warning(), so whichever one a caller happens to reach first carries the text, and the ones layered beneath it (CLI -> workflow -> designer) do not repeat it. It is a statement about the module’s status, not about a result, so repeating it per candidate or per run would only train users to skip it.

“Once” means once as the user sees it, not once per log record. When a rich console is supplied, the log record and the panel would otherwise both land on the same terminal stream and print the whole notice twice – sirnaforge.utils.logging_utils attaches a StreamHandler(sys.stdout) to the root logger on first get_logger call, and rich writes to sys.stdout too. emit_zfn_experimental_warning() therefore keeps the log record (so log files, caplog and library callers still get it) but mutes, for that one record only, the handlers writing to the console’s own stream.

Where the notice lands with no console: wherever the host application’s logging sends WARNING. Under sirnaforge’s own get_logger() that is stdout, not stderr – because a handler is always installed, logging’s last-resort stderr handler never fires.

sirnaforge.zfn.experimental.emit_zfn_experimental_warning(console: Console | None = None) → bool

Announce the ZFN arm’s experimental status, at most once per process.

The notice is logged at WARNING on every path, so it reaches log files and any handler the host application configured. It is rendered to the terminal exactly once: with a console the user sees the rich panel, without one they see whatever the log handlers show (stdout, under sirnaforge’s own logging setup).

Parameters:

console – Rich console to render a highlighted notice on. Omit it for library callers – the warning still reaches the log.

Returns:

True if this call emitted the notice, False if an earlier call already did.

sirnaforge.zfn.experimental.reset_zfn_experimental_warning() → None

Re-arm the once-per-process latch. For tests that exercise several ZFN entry points.

High-level ZFN pair evaluation workflow.

class sirnaforge.zfn.design.ZFNDesigner(searcher: ZFNOffTargetSearcher | None = None)

Bases: object

Evaluate one provided ZFN pair with exhaustive off-target search.

__init__(searcher: ZFNOffTargetSearcher | None = None) → None

Initialize with a searcher implementation.

evaluate_pair(params: ZFNDesignParameters, annotation: GenomicAnnotationConfig | None = None) → ZFNDesignResult

Run end-to-end pair evaluation and return typed results.

build_candidate(params: ZFNDesignParameters, sites: list[ZFNOffTargetSite]) → ZFNCandidate

Public wrapper for candidate-summary construction.

score_manufacturability(params: ZFNDesignParameters) → float

Public wrapper for manufacturability scoring logic.

static count_ambiguous_per_triplet(seq: str) → list[int]

Public wrapper for triplet ambiguity counting helper.

tool_versions() → dict[str, str]

Public wrapper for tool version metadata.

ZFN Search

Exhaustive ZFN off-target search implementation for provided half-sites.

sirnaforge.zfn.search.build_zfn_search_index(*, backend: ZFNSearchBackend, genome_fasta: Path, output_dir: Path | None = None) → dict[str, Any]

Build a persisted search-space index bundle for indexed ZFN backends.

sirnaforge.zfn.search.resolve_target_contigs(contig_names: list[str], requested: list[str]) → list[str]

Resolve requested chromosome filters against loaded contig names.

sirnaforge.zfn.search.build_zfn_shard_specs(contig_lengths: dict[str, int], params: ZFNDesignParameters) → list[_ShardSpec]

Build chromosome/chunk shard specs from contig lengths.

This is the authoritative shard planning logic shared by the direct Python searcher and external orchestration layers such as Nextflow.

class sirnaforge.zfn.search.ExhaustiveZFNOffTargetSearcher(annotation_provider: ZFNAnnotationProvider | None = None)

Bases: object

Exhaustive sliding-window off-target search for a provided ZFN pair.

__init__(annotation_provider: ZFNAnnotationProvider | None = None) → None

Initialize searcher with optional annotation provider.

search(params: ZFNDesignParameters, annotation: GenomicAnnotationConfig | None = None) → list[ZFNOffTargetSite]

Search all predicted cut sites with explicit mismatch + spacer constraints.

search_region(params: ZFNDesignParameters, chrom: str, scan_start0: int, scan_end0: int, core_start0: int | None = None, core_end0: int | None = None, annotation: GenomicAnnotationConfig | None = None, top_n_sites: int | None = None) → list[ZFNOffTargetSite]

Search one bounded genomic region using the same core engine as full search.

The region is scanned across scan_start0..scan_end0 and then filtered to the optional core window so overlapping shards can share context without double-reporting the same site.

ZFN Ranking

Ranking utilities for ZFN off-target sites.

This module centralizes deterministic site ordering for reporting/regression use. The tie-break behavior follows the PROGNOS-inspired policy used in tests:

1. Higher score first.

2. Region priority for equal scores: Exon > Promoter > Intron > Intergenic > Unknown.

3. Chromosomal location ordering.

sirnaforge.zfn.rank.rank_sites(sites: Sequence[ZFNOffTargetSite], params: ZFNDesignParameters | None = None) → list[ZFNOffTargetSite]

Rank off-target sites with score-first ordering and deterministic tie-breaks.

Parameters:

• sites – Candidate sites to rank.

• params – Optional design parameters. Currently only used to keep a typed, future-proof API for algorithm-specific ranking extensions.

Returns:

A new sorted list of sites.

ZFN Annotation

GTF/GFF-backed annotation provider for ZFN off-target sites.

class sirnaforge.zfn.annotation.GTFZFNAnnotationProvider

Bases: object

Annotate predicted ZFN sites using local GTF/GFF intervals.

__init__() → None

Initialize provider-level cache of parsed annotation indexes.

annotate(sites: list[ZFNOffTargetSite], config: GenomicAnnotationConfig) → list[ZFNOffTargetSite]

Attach region and nearest gene labels to each site.

ZFN Interfaces

Interfaces for ZFN off-target search and annotation engines.

class sirnaforge.zfn.interfaces.ZFNOffTargetSearcher(*args, **kwargs)

Bases: Protocol

Protocol for ZFN off-target site searchers.

search(params: ZFNDesignParameters, annotation: GenomicAnnotationConfig | None = None) → list[ZFNOffTargetSite]

Return predicted off-target cut sites for a provided ZFN pair.

__init__(*args, **kwargs)

class sirnaforge.zfn.interfaces.ZFNAnnotationProvider(*args, **kwargs)

Bases: Protocol

Optional protocol for site-level region annotation.

annotate(sites: list[ZFNOffTargetSite], config: GenomicAnnotationConfig) → list[ZFNOffTargetSite]

Attach region and nearest-gene annotation to predicted sites.

__init__(*args, **kwargs)

ZFN Nextflow Bridge

Typed Python bridge for ZFN Nextflow modules.

These helpers keep domain logic in Python while Nextflow handles process orchestration.

class sirnaforge.zfn.nextflow_bridge.ZFNShardRow(shard_id: str, chrom: str, core_start_1: int, core_end_1: int, scan_start_1: int, scan_end_1: int, max_mismatches: int)

Bases: object

One row in the ZFN shard manifest.

shard_id: str

chrom: str

core_start_1: int

core_end_1: int

scan_start_1: int

scan_end_1: int

max_mismatches: int

__init__(shard_id: str, chrom: str, core_start_1: int, core_end_1: int, scan_start_1: int, scan_end_1: int, max_mismatches: int) → None

sirnaforge.zfn.nextflow_bridge.make_zfn_shard_manifest(*, genome_fasta: Path, left_half_site: str, right_half_site: str, spacer_lengths: str, max_mismatches: int, sharding_enabled: str | bool, shard_chunk_mb: float, shard_overlap_bp: int, shard_chromosomes: str, output_tsv: Path) → dict[str, int | bool]

Build shard TSV for ZFN search.

Sharding is applied whenever enabled, including for single-contig references. The direct Python searcher remains authoritative for overlap and chunk planning.

sirnaforge.zfn.nextflow_bridge.run_zfn_shard_search(*, shard_id: str, shard_chrom: str, scan_start_1: int, scan_end_1: int, core_start_1: int | None = None, core_end_1: int | None = None, shard_max_mismatches: int, left_half_site: str, right_half_site: str, genome_fasta: Path, search_backend: ZFNSearchBackend, search_space_index: Path | None, algorithm: ZFNAlgorithm, dimer_mode: DimerMode, spacer_lengths: str, annotation_file: Path | None, output_sites_csv: Path, output_summary_json: Path) → dict[str, int | str]

Execute one shard search and persist shard-scoped outputs.

The search scans scan_start_1..scan_end_1 (1-based, inclusive) for sequence context, but only sites whose coordinates fall entirely within the core window (core_start_1..core_end_1) are written to output. This prevents duplicate sites across overlapping shards. When core_start_1 or core_end_1 are not provided they default to the scan boundaries.

sirnaforge.zfn.nextflow_bridge.aggregate_zfn_shard_results(*, shard_csv_glob: str, output_sites_csv: Path, output_summary_json: Path) → dict[str, int]

Merge shard CSVs, deduplicate coordinate/orientation collisions, and rank globally.

ZFN Benchmark Data

Typed ingestion helpers for CCR5 ZFN benchmark tables (PROGNOS supplementary extracts).

class sirnaforge.zfn.benchmark_data.CCR5S10VisibleRow(closest_gene: str, match_type: MatchType, chrom: str, pos_hg19: int, plus_half_site: str, minus_half_site: str, empty_indels: int | None, empty_total: int | None, active_indels: int | None, active_total: int | None, active_mutation_freq_percent: float | None, p_value: float | None, notes: str, sequencing_failure: bool)

Bases: object

One visible row from the extracted S10 CCR5 off-target validation table.

closest_gene: str

match_type: MatchType

chrom: str

pos_hg19: int

plus_half_site: str

minus_half_site: str

empty_indels: int | None

empty_total: int | None

active_indels: int | None

active_total: int | None

active_mutation_freq_percent: float | None

p_value: float | None

notes: str

sequencing_failure: bool

__init__(closest_gene: str, match_type: MatchType, chrom: str, pos_hg19: int, plus_half_site: str, minus_half_site: str, empty_indels: int | None, empty_total: int | None, active_indels: int | None, active_total: int | None, active_mutation_freq_percent: float | None, p_value: float | None, notes: str, sequencing_failure: bool) → None

class sirnaforge.zfn.benchmark_data.CCR5S11HomologyRow(homology_rank: int, t_mismatches: int, plus_mismatches: int, minus_mismatches: int, interrogated_by: str, closest_gene: str, match_type: MatchType, chrom: str, pos_hg19: int, plus_half_site: str, minus_half_site: str)

Bases: object

One visible row from the S11 homology ranking extract.

homology_rank: int

t_mismatches: int

plus_mismatches: int

minus_mismatches: int

interrogated_by: str

closest_gene: str

match_type: MatchType

chrom: str

pos_hg19: int

plus_half_site: str

minus_half_site: str

__init__(homology_rank: int, t_mismatches: int, plus_mismatches: int, minus_mismatches: int, interrogated_by: str, closest_gene: str, match_type: MatchType, chrom: str, pos_hg19: int, plus_half_site: str, minus_half_site: str) → None

class sirnaforge.zfn.benchmark_data.MatchType(left: Literal['L', 'R'], spacer_len: int, right: Literal['L', 'R'])

Bases: object

Structured representation of PROGNOS match type, e.g. L-5-R.

left: Literal['L', 'R']

spacer_len: int

right: Literal['L', 'R']

__init__(left: Literal['L', 'R'], spacer_len: int, right: Literal['L', 'R']) → None

sirnaforge.zfn.benchmark_data.load_ccr5_s10_visible_rows(path: str | Path) → list[CCR5S10VisibleRow]

Load visible S10 rows from CSV into typed records.

sirnaforge.zfn.benchmark_data.load_ccr5_s11_homology_rows(path: str | Path) → list[CCR5S11HomologyRow]

Load visible S11 homology rows from CSV into typed records.

sirnaforge.zfn.benchmark_data.parse_hg19_coordinate(raw: str) → tuple[str, int]

Parse hg19 coordinates in chrN:POS or compact chrNPOS format.

sirnaforge.zfn.benchmark_data.parse_match_type(raw: str) → MatchType

Parse a match-type token such as L-5-R into structured fields.

Data Models

SiRNA Models

Pydantic models for siRNA design data structures.

class sirnaforge.models.sirna.FilterCriteria(*, gc_min: Annotated[float, Ge(ge=0), Le(le=100)] = 35.0, gc_max: Annotated[float, Ge(ge=0), Le(le=100)] = 60.0, max_poly_runs: Annotated[int, Ge(ge=1)] = 3, max_paired_fraction: Annotated[float, Ge(ge=0), Le(le=1)] = 0.6, min_asymmetry_score: Annotated[float, Ge(ge=0.3), Le(le=1)] = 0.65, min_empirical_score: Annotated[float, Ge(ge=0.4), Le(le=0.7)] = 0.5)

Bases: BaseModel

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

gc_min: float

gc_max: float

max_poly_runs: int

max_paired_fraction: float

min_asymmetry_score: float

min_empirical_score: float

classmethod gc_max_greater_than_min(v: float, info: ValidationInfo) → float

Validate that gc_max is greater than or equal to gc_min.

model_config = {}

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

class sirnaforge.models.sirna.OffTargetFilterCriteria(*, max_off_target_count: Annotated[int | None, Ge(ge=0)] = 15, max_transcriptome_hits_0mm: Annotated[int | None, Ge(ge=0)] = 1, max_transcriptome_hits_1mm: Annotated[int | None, Ge(ge=0)] = 10, max_transcriptome_hits_2mm: Annotated[int | None, Ge(ge=0)] = 50, max_transcriptome_seed_perfect: Annotated[int | None, Ge(ge=0)] = None, max_mirna_perfect_seed: Annotated[int | None, Ge(ge=0)] = 0, max_mirna_1mm_seed: Annotated[int | None, Ge(ge=0)] = 10, fail_on_high_risk_mirna: bool = True, max_total_offtarget_hits: Annotated[int | None, Ge(ge=0)] = None)

Bases: BaseModel

Filtering criteria for off-target analysis results.

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

max_off_target_count: int | None

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 = {}

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

class sirnaforge.models.sirna.ScoringWeights(*, asymmetry: Annotated[float, Ge(ge=0), Le(le=1)] = 0.12, gc_content: Annotated[float, Ge(ge=0), Le(le=1)] = 0.1, accessibility: Annotated[float, Ge(ge=0), Le(le=1)] = 0.13, empirical: Annotated[float, Ge(ge=0), Le(le=1)] = 0.15, off_target: Annotated[float, Ge(ge=0), Le(le=1)] = 0.25, isoform_coverage: Annotated[float, Ge(ge=0), Le(le=1)] = 0.15, conservation: Annotated[float, Ge(ge=0), Le(le=1)] = 0.1)

Bases: BaseModel

Relative weights for composite siRNA scoring components.

asymmetry: float

gc_content: float

accessibility: float

empirical: float

off_target: float

isoform_coverage: float

conservation: float

weights_sum_to_one() → ScoringWeights

Validate that scoring weights sum to approximately 1.0.

model_config = {}

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'

ZFN = 'zfn'

class sirnaforge.models.sirna.MiRNADesignConfig(*, gc_min: Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = 30.0, gc_max: Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = 52.0, asymmetry_min: Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = 0.65, max_homopolymer: Annotated[int, ~annotated_types.Ge(ge=1)] = 3, overhang: str = 'UU', modifications: str = 'standard_2ome', off_target_preset: str = 'MIRNA_SEED_7_8', scoring_weights: dict[str, float]=<factory>, enable_pri_hairpin_validation: bool = False)

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).

model_config = {'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(*, design_mode: DesignMode = DesignMode.SIRNA, sirna_length: Annotated[int, ~annotated_types.Ge(ge=19), ~annotated_types.Le(le=23)] = 21, top_n: Annotated[int | None, ~annotated_types.Ge(ge=1)] = None, filters: FilterCriteria = <factory>, offtarget_filters: OffTargetFilterCriteria = <factory>, scoring: ScoringWeights = <factory>, avoid_snps: bool = True, check_off_targets: bool = True, predict_structure: bool = True, apply_modifications: bool = True, modification_pattern: str = 'standard_2ome', default_overhang: str = 'dTdT', snp_file: str | None = None, genome_index: str | None = None)

Bases: BaseModel

Complete configuration parameters for siRNA design workflow.

model_config = {'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 | None

filters: FilterCriteria

offtarget_filters: OffTargetFilterCriteria

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(*, id: str, transcript_id: str, position: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)], guide_sequence: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=19), ~annotated_types.MaxLen(max_length=40)], passenger_sequence: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=19), ~annotated_types.MaxLen(max_length=40)], gc_content: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)], length: ~typing.Annotated[int, ~annotated_types.Ge(ge=19), ~annotated_types.Le(le=40)], asymmetry_score: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)], duplex_stability: float | None = None, structure: str | None = None, mfe: float | None = None, paired_fraction: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = 0.0, off_target_screened: bool = False, off_target_count: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, off_target_penalty: ~typing.Annotated[float, ~annotated_types.Ge(ge=0)] = 0.0, transcriptome_hits_total: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, transcriptome_hits_0mm: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, transcriptome_hits_1mm: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, transcriptome_hits_2mm: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, transcriptome_hits_seed_0mm: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, on_target_confirmed: bool = False, on_target_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, ortholog_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, repeat_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, ortholog_species: str = '', repeat_flagged: bool = False, repeat_transcript_fraction: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = 0.0, mirna_hits_total: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, mirna_hits_0mm_seed: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, mirna_hits_1mm_seed: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, mirna_hits_high_risk: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, guide_pos1_base: str | None = None, pos1_pairing_state: str | None = None, seed_class: str | None = None, supp_13_16_score: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = None, seed_7mer_hits: ~typing.Annotated[int | None, ~annotated_types.Ge(ge=0)] = None, seed_8mer_hits: ~typing.Annotated[int | None, ~annotated_types.Ge(ge=0)] = None, seed_hits_weighted: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0)] = None, off_target_seed_risk_class: str | None = None, transcript_hit_count: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 1, transcript_hit_fraction: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = 1.0, isoform_coverage: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = None, conservation_score: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = None, component_scores: dict[str, float] = <factory>, composite_score: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)], score_asymmetry: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, score_gc_content: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, score_accessibility: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, score_empirical: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, score_off_target: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, score_isoform_coverage: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, score_conservation: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, scored_after_screening: bool = False, weight_set_version: str = '', passes_filters: bool | ~sirnaforge.models.sirna.SiRNACandidate.FilterStatus = True, quality_issues: list[str] = <factory>, overlapped_variants: list[dict[str, ~typing.Any]] = <factory>, allele_specific: bool = False, targeted_alleles: list[str] = <factory>, variant_mode: str | None = None, guide_metadata: ~sirnaforge.models.modifications.StrandMetadata | None = None, passenger_metadata: ~sirnaforge.models.modifications.StrandMetadata | None = None)

Bases: BaseModel

Individual siRNA candidate with computed thermodynamic and efficacy properties.

model_config = {'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_screened: bool

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

on_target_confirmed: bool

on_target_hits: int

ortholog_hits: int

repeat_hits: int

ortholog_species: str

repeat_flagged: bool

repeat_transcript_fraction: float

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

isoform_coverage: float | None

conservation_score: float | None

component_scores: dict[str, float]

composite_score: float

score_asymmetry: float | None

score_gc_content: float | None

score_accessibility: float | None

score_empirical: float | None

score_off_target: float | None

score_isoform_coverage: float | None

score_conservation: float | None

scored_after_screening: bool

weight_set_version: str

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'

LOW_EMPIRICAL_SCORE = 'LOW_EMPIRICAL_SCORE'

DIRTY_CONTROL = 'DIRTY_CONTROL'

REPEAT_ELEMENT = 'REPEAT_ELEMENT'

EXCESS_OFF_TARGETS = 'EXCESS_OFF_TARGETS'

TRANSCRIPTOME_PERFECT_MATCH = 'TRANSCRIPTOME_PERFECT_MATCH'

TRANSCRIPTOME_1MM = 'TRANSCRIPTOME_1MM'

TRANSCRIPTOME_2MM = 'TRANSCRIPTOME_2MM'

TRANSCRIPTOME_SEED_PERFECT = 'TRANSCRIPTOME_SEED_PERFECT'

MIRNA_PERFECT_SEED = 'MIRNA_PERFECT_SEED'

HIGH_RISK_MIRNA = 'HIGH_RISK_MIRNA'

TOTAL_OFFTARGETS = 'TOTAL_OFFTARGETS'

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: str) → str

Validate that sequence contains only valid nucleotides.

classmethod warn_beyond_recommended_length(v: str) → str

Warn (do not fail) when a sequence exceeds the recommended biological max.

19-23 nt is the recommended siRNA range; longer sequences (e.g. Dicer 3’ read-through isoforms) are still analyzed up to ENGINE_MAX_GUIDE_LEN.

classmethod sequences_same_length(v: str, info: ValidationInfo) → str

Validate that passenger sequence is same length as guide sequence.

to_fasta(include_metadata: bool = False) → str

Return FASTA format representation of the guide sequence.

Parameters:

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

Returns:

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

sirnaforge.models.sirna.build_candidate_row(candidate: SiRNACandidate) → dict[str, Any]

Map one SiRNACandidate to its canonical output-row dict.

The single source of truth for candidate CSV columns, shared by DesignResult.save_csv (the sirnaforge design path) and SiRNAWorkflow.step6_generate_reports (the sirnaforge workflow path) so the two writers cannot drift on which columns they emit. Optional attributes use a tolerant getattr since the workflow path feeds candidates from several producers.

class sirnaforge.models.sirna.DesignResult(*, input_file: str, parameters: ~sirnaforge.models.sirna.DesignParameters, candidates: list[~sirnaforge.models.sirna.SiRNACandidate], top_candidates: list[~sirnaforge.models.sirna.SiRNACandidate], total_sequences: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], total_candidates: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], filtered_candidates: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], processing_time: ~typing.Annotated[float, ~annotated_types.Ge(ge=0)], tool_versions: dict[str, str] = <factory>, rejected_candidates: list[~sirnaforge.models.sirna.SiRNACandidate] = <factory>)

Bases: BaseModel

Complete results from siRNA design workflow with metadata and statistics.

model_config = {'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: str) → DataFrame[SiRNACandidateSchema]

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 – Output CSV file path

Returns:

Validated DataFrame conforming to SiRNACandidateSchema

Raises:

pandera.errors.SchemaError – If data validation fails

get_summary() → dict[str, Any]

Generate summary statistics for the design results.

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(*, source_type: SourceType, identifier: str, url: str | None = None)

Bases: BaseModel

Provenance information for siRNA sequences.

Tracks the origin and validation status of siRNA sequences.

source_type: SourceType

identifier: str

url: str | None

to_header_string() → str

Convert provenance to FASTA header format.

Returns:

US10060921B2”

Return type:

Formatted string like “Patent

model_config = {}

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

class sirnaforge.models.modifications.ChemicalModification(*, type: str, positions: list[int] = <factory>)

Bases: BaseModel

Chemical modification annotation for siRNA strands.

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

type: str

positions: list[int]

classmethod validate_type(v: str) → str

Validate modification type is not empty.

classmethod validate_positions(v: list[int]) → list[int]

Validate positions are positive integers.

to_header_string() → str

Convert modification to FASTA header format.

Returns:

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

model_config = {}

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(*, id: str, sequence: str, overhang: str | None = None, chem_mods: list[ChemicalModification] = <factory>, notes: str | None = None, provenance: Provenance | None = None, confirmation_status: ConfirmationStatus = ConfirmationStatus.PENDING)

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.

id: str

sequence: str

overhang: str | None

chem_mods: list[ChemicalModification]

notes: str | None

provenance: Provenance | None

confirmation_status: ConfirmationStatus

classmethod validate_sequence(v: str) → str

Validate sequence contains only valid nucleotides.

validate_modification_positions() → StrandMetadata

Validate that modification positions don’t exceed sequence length.

to_fasta_header(target_gene: str | None = None, strand_role: StrandRole | None = None) → str

Generate FASTA header with embedded metadata.

Parameters:

• target_gene – Target gene name

• strand_role – Role of this strand in the duplex

Returns:

FASTA header string with key-value pairs

__getitem__(item: str) → Any

Get item by key.

get(item: str, default: Any = None) → Any

Get item by key with default.

__contains__(item: object) → bool

Check if item is in fields.

keys() → Iterable[str]

Get field keys.

items() → Iterable[tuple[str, Any]]

Get field items.

model_config = {}

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

class sirnaforge.models.modifications.SequenceRecord(*, target_gene: str, strand_role: StrandRole, metadata: StrandMetadata)

Bases: BaseModel

Complete sequence record with strand metadata.

Associates a strand with its target and role information.

target_gene: str

strand_role: StrandRole

metadata: StrandMetadata

to_fasta() → str

Generate complete FASTA record.

Returns:

Multi-line FASTA string with header and sequence

model_config = {}

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(*, qname: str, qseq: str, coord: Annotated[int, Ge(ge=0)], strand: AlignmentStrand, cigar: str, mapq: Annotated[int, Ge(ge=0), Le(le=255)], as_score: int | None = None, nm: Annotated[int, Ge(ge=0)], seed_mismatches: Annotated[int, Ge(ge=0)], offtarget_score: Annotated[float, Ge(ge=0)])

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.

model_config = {'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: str) → str

Ensure sequence contains only valid nucleotide characters.

classmethod validate_cigar(v: str) → str

Basic CIGAR string validation.

abstractmethod to_dict() → dict[str, Any]

Convert to dictionary for TSV/JSON serialization.

to_tsv_row() → str

Convert to TSV row format.

abstractmethod classmethod tsv_header() → str

Get TSV header line.

class sirnaforge.models.off_target.OffTargetHit(*, qname: str, qseq: str, coord: Annotated[int, Ge(ge=0)], strand: AlignmentStrand, cigar: str, mapq: Annotated[int, Ge(ge=0), Le(le=255)], as_score: int | None = None, nm: Annotated[int, Ge(ge=0)], seed_mismatches: Annotated[int, Ge(ge=0)], offtarget_score: Annotated[float, Ge(ge=0)], species: str, rname: str)

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.

species: str

rname: str

to_dict() → dict[str, Any]

Convert to dictionary for TSV/JSON serialization.

classmethod tsv_header() → str

Get TSV header line.

model_config = {'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(*, qname: str, qseq: str, coord: Annotated[int, Ge(ge=0)], strand: AlignmentStrand, cigar: str, mapq: Annotated[int, Ge(ge=0), Le(le=255)], as_score: int | None = None, nm: Annotated[int, Ge(ge=0)], seed_mismatches: Annotated[int, Ge(ge=0)], offtarget_score: Annotated[float, Ge(ge=0)], species: str, database: MiRNADatabase | str, mirna_id: str)

Bases: BaseAlignmentHit

Single miRNA seed match hit from BWA analysis.

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

species: str

database: MiRNADatabase | str

mirna_id: str

to_dict() → dict[str, Any]

Convert to dictionary for TSV/JSON serialization.

classmethod tsv_header() → str

Get TSV header line.

model_config = {'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(*, candidate_id: str, total_sequences: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], total_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], timestamp: str = <factory>, status: str = 'completed')

Bases: BaseModel

Base class for analysis summary statistics with common metadata fields.

model_config = {'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(*, candidate_id: str, total_sequences: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], total_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], timestamp: str = <factory>, status: str = 'completed', species: str, mode: ~sirnaforge.models.off_target.AnalysisMode, mean_mapq: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=255.0)] = None, mean_mismatches: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0)] = None, mean_seed_mismatches: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0)] = None)

Bases: BaseSummary

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

species: str

mode: AnalysisMode

mean_mapq: float | None

mean_mismatches: float | None

mean_seed_mismatches: float | None

model_config = {'frozen': False}

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

class sirnaforge.models.off_target.MiRNASummary(*, candidate_id: str, total_sequences: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], total_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], timestamp: str = <factory>, status: str = 'completed', mirna_database: ~sirnaforge.models.off_target.MiRNADatabase | str, species_analyzed: list[str], hits_per_species: dict[str, int] = <factory>, total_raw_alignments: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, parameters: dict[str, ~typing.Any] = <factory>)

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).

mirna_database: MiRNADatabase | str

species_analyzed: list[str]

hits_per_species: dict[str, int]

total_raw_alignments: int

parameters: dict[str, Any]

model_config = {'frozen': False}

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

class sirnaforge.models.off_target.BaseAggregatedSummary(*, species_analyzed: list[str], analysis_files_processed: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], combined_tsv: ~pathlib.Path | None = None, combined_json: ~pathlib.Path | None = None, summary_file: ~pathlib.Path | None = None, timestamp: str = <factory>)

Bases: BaseModel

Base class for aggregated analysis summaries with common fields.

model_config = {'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: Path | None) → str | None

Serialize Path objects to strings for JSON compatibility.

class sirnaforge.models.off_target.AggregatedOffTargetSummary(*, species_analyzed: list[str], analysis_files_processed: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], combined_tsv: ~pathlib.Path | None = None, combined_json: ~pathlib.Path | None = None, summary_file: ~pathlib.Path | None = None, timestamp: str = <factory>, total_results: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], hits_per_species: dict[str, int] = <factory>, human_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, other_species_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, species_file_counts: dict[str, int] = <factory>, missing_species: list[str] = <factory>, status: str = 'completed')

Bases: BaseAggregatedSummary

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

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 = {'frozen': False}

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

class sirnaforge.models.off_target.AggregatedMiRNASummary(*, species_analyzed: list[str], analysis_files_processed: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], combined_tsv: ~pathlib.Path | None = None, combined_json: ~pathlib.Path | None = None, summary_file: ~pathlib.Path | None = None, timestamp: str = <factory>, total_mirna_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], mirna_database: ~sirnaforge.models.off_target.MiRNADatabase | str, hits_per_species: dict[str, int] = <factory>, hits_per_candidate: dict[str, int] = <factory>, human_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, other_species_hits: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, total_candidates: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)])

Bases: BaseAggregatedSummary

Summary of aggregated miRNA results across multiple candidates.

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 = {'frozen': False}

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

ZFN Models

Pydantic models for Zinc Finger Nuclease (ZFN) design and off-target search.

class sirnaforge.models.zfn.Strand(*values)

Bases: str, Enum

Genome strand labels for reported sites.

PLUS = '+'

MINUS = '-'

class sirnaforge.models.zfn.DimerMode(*values)

Bases: str, Enum

What dimerization outcomes to consider when enumerating cut sites.

HETERODIMER_ONLY = 'heterodimer_only'

INCLUDE_HOMODIMERS = 'include_homodimers'

class sirnaforge.models.zfn.ZFNAlgorithm(*values)

Bases: str, Enum

Off-target ranking model family for ZFN sites.

HOMOLOGY = 'homology'

CONSERVED_G = 'conserved_g'

ZFN_V2 = 'zfn_v2'

class sirnaforge.models.zfn.MatchOrientation(*values)

Bases: str, Enum

Orientation/ordering of half-sites at a genomic locus.

LR = 'L...R'

RL = 'R...L'

LL = 'L...L'

RR = 'R...R'

class sirnaforge.models.zfn.IUPACMode(*values)

Bases: str, Enum

How degenerate bases are handled in the query.

NONE = 'none'

ALLOW_IUPAC = 'allow_iupac'

EXPAND_IUPAC = 'expand_iupac'

class sirnaforge.models.zfn.ZFNSearchBackend(*values)

Bases: str, Enum

Available scan backends for ZFN half-site search.

EXHAUSTIVE_PYTHON = 'exhaustive_python'

PYAHOCORASICK = 'pyahocorasick'

FM_INDEX = 'fm_index'

class sirnaforge.models.zfn.ZFNHalfSiteConstraints(*, min_len: Annotated[int, Ge(ge=6)] = 9, max_len: Annotated[int, Ge(ge=6)] = 18, max_mismatches: Annotated[int, Ge(ge=0), Le(le=6)] = 2, seed_len_from_fokI: Annotated[int | None, Ge(ge=1)] = 6, seed_max_mismatches: Annotated[int | None, Ge(ge=0)] = 1, iupac_mode: IUPACMode = IUPACMode.ALLOW_IUPAC, window_stride: Annotated[int, Ge(ge=1), Le(le=50)] = 1)

Bases: BaseModel

Constraints on one ZFN half-site.

model_config = {'extra': 'forbid'}

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

min_len: int

max_len: int

max_mismatches: int

seed_len_from_fokI: int | None

seed_max_mismatches: int | None

iupac_mode: IUPACMode

window_stride: int

classmethod max_len_ge_min_len(v: int, info: ValidationInfo) → int

Validate max_len >= min_len.

class sirnaforge.models.zfn.ZFNSpacerConstraints(*, allowed_spacer_lengths: ~typing.Annotated[list[int], ~annotated_types.MinLen(min_length=1)] = <factory>, require_opposite_strands: bool = True)

Bases: BaseModel

Spacer constraints between Left and Right half-sites at a cleavage site.

model_config = {'extra': 'forbid'}

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

allowed_spacer_lengths: list[int]

require_opposite_strands: bool

classmethod normalize_spacer_lengths(v: list[int]) → list[int]

Normalize spacer lengths to a sorted, unique positive list.

class sirnaforge.models.zfn.ZFNOffTargetFilterCriteria(*, max_total_sites: Annotated[int | None, Ge(ge=0)] = None, max_exonic_sites: Annotated[int | None, Ge(ge=0)] = None, max_promoter_sites: Annotated[int | None, Ge(ge=0)] = None, min_site_score_to_count: Annotated[float, Ge(ge=0), Le(le=100)] = 70.0)

Bases: BaseModel

Criteria to fail a candidate ZFN due to predicted off-target burden.

model_config = {'extra': 'forbid'}

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

max_total_sites: int | None

max_exonic_sites: int | None

max_promoter_sites: int | None

min_site_score_to_count: float

class sirnaforge.models.zfn.ZFNScoringWeights(*, on_target_quality: Annotated[float, Ge(ge=0), Le(le=1)] = 0.35, off_target_specificity: Annotated[float, Ge(ge=0), Le(le=1)] = 0.45, manufacturability: Annotated[float, Ge(ge=0), Le(le=1)] = 0.2)

Bases: BaseModel

Weights for composite ZFN candidate scoring.

model_config = {'extra': 'forbid'}

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

on_target_quality: float

off_target_specificity: float

manufacturability: float

classmethod weights_sum_to_one(v: float, info: ValidationInfo) → float

Validate scoring weights sum to ~1.0.

class sirnaforge.models.zfn.ZFNMutationType(*values)

Bases: str, Enum

Allowed mutation categories for ZFN sub-finger constraints.

SUBSTITUTION = 'substitution'

TRANSITION = 'transition'

TRANSVERSION = 'transversion'

INSERTION = 'insertion'

DELETION = 'deletion'

class sirnaforge.models.zfn.ZFNSubfingerMutationConstraint(*, subfinger_index: Annotated[int, Ge(ge=1)], max_mutations: Annotated[int, Ge(ge=0)], mutation_types: Annotated[list[ZFNMutationType], MinLen(min_length=1)])

Bases: BaseModel

Mutation allowance definition for one ZFN sub-finger.

subfinger_index: int

max_mutations: int

mutation_types: list[ZFNMutationType]

model_config = {}

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

class sirnaforge.models.zfn.ZFNDefaultSubfingerMutationConstraint(*, max_mutations: Annotated[int, Ge(ge=0)], mutation_types: Annotated[list[ZFNMutationType], MinLen(min_length=1)])

Bases: BaseModel

Default mutation allowance applied to each sub-finger.

max_mutations: int

mutation_types: list[ZFNMutationType]

model_config = {}

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

class sirnaforge.models.zfn.ZFNOverallMutationConstraint(*, max_mutations: Annotated[int, Ge(ge=0)], mutation_types: Annotated[list[ZFNMutationType], MinLen(min_length=1)])

Bases: BaseModel

Global mutation allowance applied across all sub-fingers.

max_mutations: int

mutation_types: list[ZFNMutationType]

model_config = {}

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

class sirnaforge.models.zfn.ZFNMutationConstraints(*, subfinger_mutations: list[ZFNSubfingerMutationConstraint] = <factory>, default_subfinger_mutation: ZFNDefaultSubfingerMutationConstraint | None = None, overall_mutations: list[ZFNOverallMutationConstraint] = <factory>)

Bases: BaseModel

Composite container for all ZFN sub-finger and global mutation budgets.

model_config = {'extra': 'forbid'}

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

subfinger_mutations: list[ZFNSubfingerMutationConstraint]

default_subfinger_mutation: ZFNDefaultSubfingerMutationConstraint | None

overall_mutations: list[ZFNOverallMutationConstraint]

class sirnaforge.models.zfn.ZFNShardingConfig(*, enabled: bool = True, chunk_size_bp: Annotated[int, ~annotated_types.Ge(ge=1)] = 12000000, overlap_bp: Annotated[int, ~annotated_types.Ge(ge=0)] = 50, chromosomes: list[str] = <factory>, max_workers: Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=128)] = 2, memory_budget_gb: Annotated[float | None, ~annotated_types.Gt(gt=0)] = None, memory_reserve_gb: Annotated[float, ~annotated_types.Ge(ge=0)] = 2.0, target_cpu_utilization: Annotated[float | None, ~annotated_types.Gt(gt=0), ~annotated_types.Le(le=1)] = None, max_cpu_workers: Annotated[int | None, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=128)] = None)

Bases: BaseModel

Optional chromosome/chunk sharding controls for scalable ZFN search.

model_config = {'extra': 'forbid'}

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

enabled: bool

chunk_size_bp: int

overlap_bp: int

chromosomes: list[str]

max_workers: int

memory_budget_gb: float | None

memory_reserve_gb: float

target_cpu_utilization: float | None

max_cpu_workers: int | None

classmethod validate_chromosomes(v: list[str]) → list[str]

Normalize chromosome names while preserving input order.

class sirnaforge.models.zfn.ZFNDesignParameters(*, search_space_reference: str | None = 'ensembl_human_hg38_primary', search_space_fasta: str | None = None, search_space_index: str | None = None, search_backend: ZFNSearchBackend = ZFNSearchBackend.PYAHOCORASICK, left_half_site: str, right_half_site: str, half_site_constraints: ZFNHalfSiteConstraints = <factory>, spacer_constraints: ZFNSpacerConstraints = <factory>, dimer_mode: DimerMode = DimerMode.HETERODIMER_ONLY, algorithm: ZFNAlgorithm = ZFNAlgorithm.ZFN_V2, top_n_sites: Annotated[int, ~annotated_types.Ge(ge=1)] = 5000, report_n_sites: Annotated[int, ~annotated_types.Ge(ge=1)] = 200, off_target_filters: ZFNOffTargetFilterCriteria = <factory>, scoring: ZFNScoringWeights = <factory>, mutation_constraints: ZFNMutationConstraints | None = None, sharding: ZFNShardingConfig = <factory>)

Bases: BaseModel

Top-level configuration for ZFN pair evaluation and off-target search.

model_config = {'extra': 'forbid'}

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

search_space_reference: str | None

search_space_fasta: str | None

search_space_index: str | None

search_backend: ZFNSearchBackend

left_half_site: str

right_half_site: str

half_site_constraints: ZFNHalfSiteConstraints

spacer_constraints: ZFNSpacerConstraints

dimer_mode: DimerMode

algorithm: ZFNAlgorithm

top_n_sites: int

report_n_sites: int

off_target_filters: ZFNOffTargetFilterCriteria

scoring: ZFNScoringWeights

mutation_constraints: ZFNMutationConstraints | None

sharding: ZFNShardingConfig

classmethod validate_dna_bases_and_iupac(v: str) → str

Validate half-sites as DNA/IUPAC strings.

classmethod validate_half_site_lengths(v: str, info: ValidationInfo) → str

Validate half-site lengths against configured range.

classmethod validate_search_space_source(v: str | None) → str | None

Normalize search-space reference keys.

classmethod validate_search_space_fasta(v: str | None) → str | None

Normalize explicit FASTA path/URL strings.

classmethod validate_search_space_index(v: str | None) → str | None

Normalize explicit search index path strings.

canonical_search_contract() → ZFNSearchContract

Return canonical, JSON-serializable ZFN search contract for reports/workflows.

class sirnaforge.models.zfn.ZFNSearchContract(*, left_half_site: str, right_half_site: str, allowed_spacer_lengths: list[int], max_mismatches_per_half_site: Annotated[int, Ge(ge=0)], seed_len_from_foki: Annotated[int | None, Ge(ge=1)] = None, seed_max_mismatches: Annotated[int | None, Ge(ge=0)] = None, dimer_mode: DimerMode, algorithm: ZFNAlgorithm, search_backend: ZFNSearchBackend, require_opposite_strands: bool, orientation_convention: str)

Bases: BaseModel

Canonical internal contract for ZFN search/ranking behavior.

model_config = {'extra': 'forbid'}

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

left_half_site: str

right_half_site: str

allowed_spacer_lengths: list[int]

max_mismatches_per_half_site: int

seed_len_from_foki: int | None

seed_max_mismatches: int | None

dimer_mode: DimerMode

algorithm: ZFNAlgorithm

search_backend: ZFNSearchBackend

require_opposite_strands: bool

orientation_convention: str

class sirnaforge.models.zfn.GenomicAnnotationConfig(*, annotation_path: str | None = None, annotation_reference: str | None = None, cache_dir: str | None = None, promoter_upstream_bp: Annotated[int, Ge(ge=0)] = 2000, promoter_downstream_bp: Annotated[int, Ge(ge=0)] = 200)

Bases: BaseModel

Optional annotation configuration for off-target site classification.

model_config = {'extra': 'forbid'}

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

annotation_path: str | None

annotation_reference: str | None

cache_dir: str | None

promoter_upstream_bp: int

promoter_downstream_bp: int

classmethod validate_optional_paths(v: str | None) → str | None

Normalize optional string paths/keys.

resolved_annotation_path() → Path | None

Return resolved local annotation path when available.

class sirnaforge.models.zfn.ZFNOffTargetSite(*, site_id: str, chrom: str, start_1based: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)], end_1based: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)], strand: ~sirnaforge.models.zfn.Strand, orientation: ~sirnaforge.models.zfn.MatchOrientation, spacer_len: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], sequence: str, left_mismatches: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], right_mismatches: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], left_seed_mismatches: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, right_seed_mismatches: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, left_mismatch_positions: list[int] = <factory>, right_mismatch_positions: list[int] = <factory>, total_mismatches: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], score: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)], score_components: dict[str, float] = <factory>, dimer_compatible: bool = True, region: ~typing.Literal['exon', 'promoter', 'intron', 'intergenic', 'unknown'] = 'unknown', nearest_gene: str | None = None, left_aligned: str, right_aligned: str)

Bases: BaseModel

One predicted cleavage site for a ZFN pair at a genomic locus.

model_config = {'extra': 'forbid'}

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

site_id: str

chrom: str

start_1based: int

end_1based: int

strand: Strand

orientation: MatchOrientation

spacer_len: int

sequence: str

left_mismatches: int

right_mismatches: int

left_seed_mismatches: int

right_seed_mismatches: int

left_mismatch_positions: list[int]

right_mismatch_positions: list[int]

total_mismatches: int

score: float

score_components: dict[str, float]

dimer_compatible: bool

region: Literal['exon', 'promoter', 'intron', 'intergenic', 'unknown']

nearest_gene: str | None

left_aligned: str

right_aligned: str

class sirnaforge.models.zfn.ZFNCandidate(*, id: str, left_half_site: str, right_half_site: str, allowed_spacers: list[int], predicted_sites_total: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, predicted_sites_exonic: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, predicted_sites_promoter: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, worst_site_score: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, best_offtarget_score: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)] = None, passes_offtarget_filters: bool | str = True, component_scores: dict[str, float] = <factory>, composite_score: ~typing.Annotated[float, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=100)], top_offtargets: list[~sirnaforge.models.zfn.ZFNOffTargetSite] = <factory>)

Bases: BaseModel

A ZFN pair candidate with summary metrics and composite score.

model_config = {'extra': 'forbid'}

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

id: str

left_half_site: str

right_half_site: str

allowed_spacers: list[int]

predicted_sites_total: int

predicted_sites_exonic: int

predicted_sites_promoter: int

worst_site_score: float | None

best_offtarget_score: float | None

passes_offtarget_filters: bool | str

component_scores: dict[str, float]

composite_score: float

top_offtargets: list[ZFNOffTargetSite]

class sirnaforge.models.zfn.ZFNDesignResult(*, parameters: ~sirnaforge.models.zfn.ZFNDesignParameters, annotation: ~sirnaforge.models.zfn.GenomicAnnotationConfig | None = None, candidates: list[~sirnaforge.models.zfn.ZFNCandidate] = <factory>, off_target_sites: list[~sirnaforge.models.zfn.ZFNOffTargetSite] = <factory>, processing_time_s: ~typing.Annotated[float, ~annotated_types.Ge(ge=0)], tool_versions: dict[str, str] = <factory>)

Bases: BaseModel

Complete results for ZFN pair evaluation and off-target search.

model_config = {'extra': 'forbid'}

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

parameters: ZFNDesignParameters

annotation: GenomicAnnotationConfig | None

candidates: list[ZFNCandidate]

off_target_sites: list[ZFNOffTargetSite]

processing_time_s: float

tool_versions: dict[str, str]

save_offtargets_csv(filepath: str) → DataFrame

Save off-target site table to CSV.

get_summary() → dict[str, Any]

Generate a compact run summary.

Transcript Annotation Models

Pydantic models for transcript annotation data structures.

class sirnaforge.models.transcript_annotation.Interval(*, seq_region_name: str, start: Annotated[int, Ge(ge=0)], end: Annotated[int, Ge(ge=0)], strand: int | None = None)

Bases: BaseModel

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

seq_region_name: str

start: int

end: int

strand: int | None

model_config = {'frozen': True}

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

__hash__() → int

Make Interval hashable for use in sets/dicts.

__eq__(other: object) → bool

Compare Interval instances for equality.

class sirnaforge.models.transcript_annotation.TranscriptAnnotation(*, transcript_id: str, gene_id: str, symbol: str | None = None, biotype: str | None = None, seq_region_name: str, start: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)], end: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)], strand: int, gene_interval: ~sirnaforge.models.transcript_annotation.Interval | None = None, exons: list[~sirnaforge.models.transcript_annotation.Interval] = <factory>, cds: list[~sirnaforge.models.transcript_annotation.Interval] = <factory>, provider: str, endpoint: str | None = None, reference_choice: str | None = None)

Bases: BaseModel

Comprehensive transcript annotation from genomic databases.

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

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 = {'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(*, transcripts: dict[str, ~sirnaforge.models.transcript_annotation.TranscriptAnnotation]=<factory>, unresolved: list[str] = <factory>, reference_choice: ReferenceChoice)

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.

transcripts: dict[str, TranscriptAnnotation]

unresolved: list[str]

reference_choice: ReferenceChoice

model_config = {'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).

Variant Models

Pydantic models for genomic variant data structures.

class sirnaforge.models.variant.VariantMode(*values)

Bases: str, Enum

Mode for how variants should be handled in siRNA design.

TARGET = 'target'

AVOID = 'avoid'

BOTH = 'both'

class sirnaforge.models.variant.VariantSource(*values)

Bases: str, Enum

Trusted sources for variant data, ordered by priority.

CLINVAR = 'clinvar'

ENSEMBL = 'ensembl'

DBSNP = 'dbsnp'

LOCAL_VCF = 'local-vcf'

class sirnaforge.models.variant.ClinVarSignificance(*values)

Bases: str, Enum

ClinVar clinical significance classifications.

PATHOGENIC = 'Pathogenic'

LIKELY_PATHOGENIC = 'Likely pathogenic'

UNCERTAIN_SIGNIFICANCE = 'Uncertain significance'

LIKELY_BENIGN = 'Likely benign'

BENIGN = 'Benign'

CONFLICTING = 'Conflicting interpretations of pathogenicity'

OTHER = 'Other'

class sirnaforge.models.variant.EnsemblMapping(*, location: str, allele_string: str, assembly_name: str, seq_region_name: str, strand: int, start: int, end: int, coord_system: str, **extra_data: Any)

Bases: BaseModel

Ensembl variation mapping information.

Represents genomic mapping data for a variant from the Ensembl Variation API. Contains coordinate and allele information for a specific genomic location.

model_config = {'extra': 'allow'}

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

location: str

allele_string: str

assembly_name: str

seq_region_name: str

strand: int

start: int

end: int

coord_system: str

class sirnaforge.models.variant.EnsemblPopulationFrequency(*, population: str, frequency: Annotated[float | None, Ge(ge=0), Le(le=1)], allele_count: Annotated[int | None, Ge(ge=0)] = None, allele_number: Annotated[int | None, Ge(ge=0)] = None, allele: str | None = None, **extra_data: Any)

Bases: BaseModel

Population frequency data from Ensembl.

Contains allele frequency information for a specific population from sources like 1000 Genomes, gnomAD, ExAC, etc.

model_config = {'extra': 'allow'}

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

population: str

frequency: float | None

allele_count: int | None

allele_number: int | None

allele: str | None

class sirnaforge.models.variant.EnsemblVariationResponse(*, name: str, var_class: str, source: str, most_severe_consequence: str | None, MAF: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)], minor_allele: str | None, ambiguity: str, mappings: list[~sirnaforge.models.variant.EnsemblMapping], clinical_significance: list[str] | None, synonyms: list[str] | None, evidence: list[str] | None, populations: list[~sirnaforge.models.variant.EnsemblPopulationFrequency | None] = <factory>, **extra_data: ~typing.Any)

Bases: BaseModel

Response model for Ensembl Variation API.

Represents the complete response from the Ensembl REST API variation endpoint. Contains variant metadata, mappings, frequencies, and clinical information.

model_config = {'extra': 'allow'}

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

name: str

var_class: str

source: str

most_severe_consequence: str | None

MAF: float | None

minor_allele: str | None

ambiguity: str

mappings: list[EnsemblMapping]

clinical_significance: list[str] | None

synonyms: list[str] | None

evidence: list[str] | None

populations: list[EnsemblPopulationFrequency | None]

class sirnaforge.models.variant.ClinVarVariationResponse(*, uid: str, obj_type: str | None, accession: str | None, title: str | None, germline_classification: dict | None, clinical_impact_classification: dict | None, variation_set: list[dict] | None, genes: list[dict] | None, molecular_consequence_list: list[str] | None, **extra_data: Any)

Bases: BaseModel

Response model for ClinVar variation summary.

Represents the response from NCBI ClinVar E-utilities esummary endpoint. Contains clinical significance, variation details, and associated conditions.

model_config = {'extra': 'allow'}

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

uid: str

obj_type: str | None

accession: str | None

title: str | None

germline_classification: dict | None

clinical_impact_classification: dict | None

variation_set: list[dict] | None

genes: list[dict] | None

molecular_consequence_list: list[str] | None

class sirnaforge.models.variant.VariantRecord(*, id: str | None = None, chr: str, pos: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)], ref: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], alt: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], assembly: str = 'GRCh38', sources: list[~sirnaforge.models.variant.VariantSource] = <factory>, clinvar_significance: ~sirnaforge.models.variant.ClinVarSignificance | None = None, af: ~typing.Annotated[float | None, ~annotated_types.Ge(ge=0), ~annotated_types.Le(le=1)] = None, population_afs: dict[str, float] = <factory>, annotations: dict[str, ~typing.Any] = <factory>, provenance: dict[str, ~typing.Any] = <factory>)

Bases: BaseModel

Complete variant record with annotations from multiple sources.

model_config = {'extra': 'forbid'}

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

id: str | None

chr: str

pos: int

ref: str

alt: str

assembly: str

sources: list[VariantSource]

clinvar_significance: ClinVarSignificance | None

af: float | None

population_afs: dict[str, float]

annotations: dict[str, Any]

provenance: dict[str, Any]

to_vcf_style() → str

Return variant in VCF-style coordinate format: chr:pos:ref:alt.

get_max_population_af() → float | None

Get the maximum allele frequency across all populations.

Returns:

Maximum population-specific AF, or None if no population data available

get_effective_af_for_mode(mode: VariantMode) → float | None

Get the effective allele frequency based on variant mode.

For ‘avoid’ mode: Use max population AF if available (to avoid SNPs prevalent in any geographic group), otherwise use global AF.

For ‘target’ or ‘both’ modes: Use global AF (targets most common alleles).

Parameters:

mode – Variant mode (target/avoid/both)

Returns:

Effective allele frequency for filtering, or None if no AF data

get_primary_source() → VariantSource | None

Get the highest priority source for this variant.

class sirnaforge.models.variant.VariantQueryType(*values)

Bases: str, Enum

Types of variant query identifiers.

RSID = 'rsid'

COORDINATE = 'coordinate'

HGVS = 'hgvs'

class sirnaforge.models.variant.VariantQuery(*, raw_input: str, query_type: VariantQueryType, rsid: str | None = None, chr: str | None = None, pos: Annotated[int | None, Ge(ge=1)] = None, ref: str | None = None, alt: str | None = None, hgvs: str | None = None, assembly: str = 'GRCh38')

Bases: BaseModel

Parsed variant query with normalized components.

model_config = {'extra': 'forbid'}

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

raw_input: str

query_type: VariantQueryType

rsid: str | None

chr: str | None

pos: int | None

ref: str | None

alt: str | None

hgvs: str | None

assembly: str

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_screened: Series[bool] = 'off_target_screened'

off_target_count: Series[int] = 'off_target_count'

on_target_hits: Series[Int64Dtype] = 'on_target_hits'

ortholog_hits: Series[Int64Dtype] = 'ortholog_hits'

repeat_hits: Series[Int64Dtype] = 'repeat_hits'

ortholog_species: Series[str] = 'ortholog_species'

repeat_flagged: Series[BooleanDtype] = 'repeat_flagged'

repeat_transcript_fraction: Series[float] = 'repeat_transcript_fraction'

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'

isoform_coverage: Series[float] = 'isoform_coverage'

conservation_score: Series[float] = 'conservation_score'

composite_score: Series[float] = 'composite_score'

score_asymmetry: Series[float] = 'score_asymmetry'

score_gc_content: Series[float] = 'score_gc_content'

score_accessibility: Series[float] = 'score_accessibility'

score_empirical: Series[float] = 'score_empirical'

score_off_target: Series[float] = 'score_off_target'

score_isoform_coverage: Series[float] = 'score_isoform_coverage'

score_conservation: Series[float] = 'score_conservation'

scored_after_screening: Series[BooleanDtype] = 'scored_after_screening'

weight_set_version: Series[str] = 'weight_set_version'

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: DataFrame) → bool

Ensure passes_filters contains allowed filter status values.

Derives the allow-list from SiRNACandidate.FilterStatus enum to prevent drift.

classmethod check_sequence_lengths(df: DataFrame) → bool

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

classmethod check_nucleotide_sequences(df: DataFrame) → bool

Validate sequences contain only valid RNA/DNA bases.

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: DataFrame) → bool

Ensure seed_mismatches <= nm (total mismatches).

classmethod validate_perfect_match_score(df: DataFrame) → bool

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

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: DataFrame) → bool

Ensure seed_mismatches <= nm (total mismatches).

classmethod validate_score_consistency(df: DataFrame) → bool

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

Data Access

Base Data Classes

Shared base classes and utilities for genomic data analysis.

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

Bases: Exception

Base exception for database-related errors.

__init__(message: str, database: str | None = None)

Initialize database error.

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

Bases: DatabaseError

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

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

Bases: DatabaseError

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

__init__(query: str, database: str | None = None)

Initialize gene not found error.

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(*, gene_id: str, gene_name: str | None = None, gene_type: str | None = None, chromosome: str | None = None, start: int | None = None, end: int | None = None, strand: int | None = None, description: str | None = None, database: DatabaseType)

Bases: BaseModel

Gene information model.

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 = {'use_enum_values': True}

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

class sirnaforge.data.base.TranscriptInfo(*, transcript_id: str, transcript_name: str | None = None, transcript_type: str | None = None, gene_id: str, gene_name: str | None = None, sequence: str | None = None, length: int | None = None, database: DatabaseType, is_canonical: bool = False)

Bases: BaseModel

Transcript information model.

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 = {'use_enum_values': True}

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

classmethod validate_sequence(v: str | None) → str | None

Validate RNA sequence.

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

Bases: ABC

Abstract base class for database clients.

__init__(timeout: int = 30)

Initialize database client.

abstractmethod async search_gene(query: str, include_sequence: bool = True) → tuple[GeneInfo | None, list[TranscriptInfo]]

Search for a gene and return gene info and transcripts.

Parameters:

• query – Gene ID, gene name, or transcript ID

• include_sequence – Whether to fetch transcript sequences

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: str, sequence_type: SequenceType = SequenceType.CDNA) → str

Get sequence for a specific identifier.

Parameters:

• identifier – Gene ID, transcript ID, etc.

• sequence_type – Type of sequence to retrieve

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.

property query_species: str

Canonical organism whose transcripts a gene query against this database returns.

Every concrete client here is wired to exactly one organism – EnsemblClient pins species=homo_sapiens on every lookup, RefSeqClient appends AND Homo sapiens[Organism] to every esearch term, and GencodeClient targets the human GENCODE release – so the organism of a retrieved transcript set is a property of where the transcripts came from. Callers that need the “query species” must read it from here rather than infer it from the off-target species list, which is an unordered set of genomes to screen against and says nothing about the target.

class sirnaforge.data.base.AbstractTranscriptAnnotationClient(timeout: int = 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

__init__(timeout: int = 30)

Initialize transcript annotation client.

Parameters:

timeout – Request timeout in seconds

abstractmethod async fetch_by_ids(ids: list[str], *, species: str, reference: ReferenceChoice) → TranscriptAnnotationBundle

Fetch transcript annotations by stable IDs.

Parameters:

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

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

• reference – Reference assembly/release choice

Returns:

TranscriptAnnotationBundle containing resolved annotations

Raises:

DatabaseAccessError – For network/server access issues

abstractmethod async fetch_by_regions(regions: list[str], *, species: str, reference: ReferenceChoice) → TranscriptAnnotationBundle

Fetch transcript annotations by genomic regions.

Parameters:

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

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

• reference – Reference assembly/release choice

Returns:

TranscriptAnnotationBundle containing all transcripts overlapping regions

Raises:

DatabaseAccessError – For network/server access issues

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

Bases: AbstractDatabaseClient

Client for Ensembl REST API interactions.

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

Initialize Ensembl client.

property database_type: DatabaseType

Return the database type this client handles.

property query_species: str

Canonical form of the single Ensembl species this client queries.

async search_gene(query: str, include_sequence: bool = True) → tuple[GeneInfo | None, list[TranscriptInfo]]

Search for a gene and return gene info and transcripts.

async get_sequence(identifier: str, sequence_type: SequenceType = SequenceType.CDNA, headers: dict | None = None) → str

Get sequence from Ensembl REST API.

Parameters:

• identifier – Gene ID, transcript ID, etc.

• sequence_type – Type of sequence to retrieve

• headers – Optional HTTP headers

Returns:

Sequence string

Raises:

• DatabaseAccessError – For network/server access issues

• GeneNotFoundError – When identifier is not found in database

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

Bases: AbstractDatabaseClient

Client for RefSeq database via NCBI E-utilities API.

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

Initialize RefSeq client.

property database_type: DatabaseType

Return the database type this client handles.

async search_gene(query: str, include_sequence: bool = True) → tuple[GeneInfo | None, list[TranscriptInfo]]

Search for a gene and return gene info and transcripts.

async get_sequence(identifier: str, _sequence_type: SequenceType = SequenceType.CDNA) → str

Get sequence for a specific identifier from NCBI.

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

Bases: AbstractDatabaseClient

Client for GENCODE database.

__init__(timeout: int = 30)

Initialize GENCODE client.

property database_type: DatabaseType

Return the database type this client handles.

async search_gene(query: str, include_sequence: bool = True) → tuple[GeneInfo | None, list[TranscriptInfo]]

Search for a gene and return gene info and transcripts.

async get_sequence(_identifier: str, _sequence_type: SequenceType = SequenceType.CDNA) → str

Get sequence for a specific identifier from GENCODE.

class sirnaforge.data.base.SequenceUtils

Bases: object

Utility functions for sequence analysis.

static calculate_gc_content(sequence: str) → float

Calculate GC content of a sequence.

static reverse_complement(sequence: str) → str

Get reverse complement of DNA sequence.

static transcribe_dna_to_rna(sequence: str) → str

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

static reverse_transcribe_rna_to_dna(sequence: str) → str

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

class sirnaforge.data.base.FastaUtils

Bases: object

Utility functions for FASTA file operations.

static save_sequences_fasta(sequences: list[tuple[str, str]], output_path: str | Path, line_length: int = 80) → None

Save sequences to FASTA format.

Parameters:

• sequences – List of (header, sequence) tuples

• output_path – Output file path

• line_length – Maximum line length for sequence

static read_fasta(file_path: str | Path) → list[tuple[str, str]]

Read sequences from FASTA file.

Parameters:

file_path – Path to FASTA file

Returns:

List of (header, sequence) tuples

static parse_fasta_to_dict(file_path: str | Path) → dict[str, str]

Parse FASTA file into a dictionary.

Parameters:

file_path – Path to FASTA file

Returns:

Dictionary mapping sequence names to sequences

static write_dict_to_fasta(sequences: dict[str, str], output_path: str | Path) → None

Write sequences dictionary to FASTA format.

Parameters:

• sequences – Dictionary of sequence name -> sequence

• output_path – Output file path

static validate_sirna_sequences(sequences: dict[str, str], expected_length: int = 21) → dict[str, str]

Validate siRNA sequences for correct length and nucleotide content.

Parameters:

• sequences – Dictionary of sequence name -> sequence

• expected_length – Expected siRNA length

Returns:

Dictionary of valid sequences

sirnaforge.data.base.get_database_display_name(database: DatabaseType) → str

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

Annotation Manager

Genomic annotation manager built on the transcriptome/reference cache pipeline.

class sirnaforge.data.annotation_manager.AnnotationSource(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '')

Bases: TranscriptomeSource

Annotation-specific source model.

__init__(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '') → None

class sirnaforge.data.annotation_manager.AnnotationManager(cache_dir: str | Path | None = None, cache_ttl_days: int = 30)

Bases: TranscriptomeManager

Annotation manager with caching and no index-building side effects.

SOURCE_LABEL = 'annotation'

__init__(cache_dir: str | Path | None = None, cache_ttl_days: int = 30)

Initialize annotation manager.

Parameters:

• cache_dir – Optional cache directory override

• cache_ttl_days – Cache TTL in days

get_custom_annotation(annotation_path_or_url: str | Path, cache_name: str | None = None) → Path | None

Resolve and cache a user-supplied annotation resource.

Supports local paths and remote URLs through the shared reference pipeline.

Gene Search

Gene search and sequence retrieval from multiple databases.

class sirnaforge.data.gene_search.GeneSearchResult(*, query: str, database: DatabaseType, gene_info: GeneInfo | None = None, transcripts: list[TranscriptInfo] = <factory>, error: str | None = None, is_access_error: bool = False)

Bases: BaseModel

Complete gene search result.

query: str

database: DatabaseType

gene_info: GeneInfo | None

transcripts: list[TranscriptInfo]

error: str | None

is_access_error: bool

model_config = {'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: int = 30, max_retries: int = 3)

Bases: object

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

__init__(timeout: int = 30, max_retries: int = 3)

Initialize gene searcher with database clients.

Parameters:

• timeout – Request timeout in seconds

• max_retries – Maximum retry attempts

clients: dict[DatabaseType, AbstractDatabaseClient]

get_client(database: DatabaseType) → AbstractDatabaseClient

Get the client for a specific database.

query_species(database: DatabaseType | None = None) → str

Canonical organism whose transcripts a gene query returns.

This is the authoritative “query species”: the organism the retrieved target transcripts belong to, read off the database that retrieved them rather than guessed from the off-target species list.

database=None asks the question search_gene_with_fallback asks – any of the clients may serve the query. Every client is single-organism and they all agree today, so the fallback cannot change the answer; if that ever stops being true the disagreement surfaces here (the requested database’s answer wins and a warning is logged) instead of silently mislabelling the target.

Parameters:

database – Database the caller intends to query, or None for the fallback chain.

Returns:

Canonical species name (e.g. "human").

async search_gene_with_fallback(query: str, include_sequence: bool = True) → GeneSearchResult

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 – Gene ID, gene name, or transcript ID

• include_sequence – Whether to fetch transcript sequences

Returns:

GeneSearchResult from the first accessible database

async search_gene(query: str, database: DatabaseType | None = None, include_sequence: bool = True) → GeneSearchResult

Search for a gene and retrieve its isoforms.

Parameters:

• query – Gene ID, gene name, or transcript ID

• database – Database to search (defaults to Ensembl)

• include_sequence – Whether to fetch transcript sequences

Returns:

GeneSearchResult with gene info and transcripts

async search_multiple_databases(query: str, databases: list[DatabaseType] | None = None, include_sequence: bool = True) → list[GeneSearchResult]

Search across multiple databases.

Parameters:

• query – Gene ID, gene name, or transcript ID

• databases – List of databases to search

• include_sequence – Whether to fetch sequences

Returns:

List of search results from each database

save_transcripts_fasta(transcripts: list[TranscriptInfo], output_path: str | Path, include_metadata: bool = True) → None

Save transcripts to FASTA format using shared utility.

Parameters:

• transcripts – List of transcript information

• output_path – Output file path

• include_metadata – Include metadata in FASTA headers

sirnaforge.data.gene_search.search_gene_sync(query: str, database: DatabaseType = DatabaseType.ENSEMBL, include_sequence: bool = True) → GeneSearchResult

Synchronous wrapper for gene search.

sirnaforge.data.gene_search.search_gene_with_fallback_sync(query: str, include_sequence: bool = True) → GeneSearchResult

Synchronous wrapper for gene search with fallback.

sirnaforge.data.gene_search.search_multiple_databases_sync(query: str, databases: list[DatabaseType] | None = None, include_sequence: bool = True) → list[GeneSearchResult]

Synchronous wrapper for multi-database search.

ORF Analysis

ORF analysis and sequence validation for transcript sequences.

class sirnaforge.data.orf_analysis.ORFInfo(*, 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)

Bases: BaseModel

Information about an Open Reading Frame.

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 = {'frozen': True}

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

class sirnaforge.data.orf_analysis.SequenceAnalysis(*, transcript_id: str, sequence_type: SequenceType, sequence_length: int, gc_content: float, orfs: list[ORFInfo], longest_orf: ORFInfo | None = None, has_valid_orf: bool = False, cds_sequence: str | None = None, protein_sequence: str | None = None, cds_start: int | None = None, cds_end: int | None = None, utr5_length: int | None = None, utr3_length: int | None = None, sequence_region: str | None = None)

Bases: BaseModel

Complete sequence analysis including ORF information.

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 = {'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: AbstractDatabaseClient | None = None)

Bases: object

Analyze ORFs in transcript sequences and validate sequence types.

__init__(database_client: AbstractDatabaseClient | None = None)

Initialize ORF analyzer.

Parameters:

database_client – Optional database client for retrieving additional sequence types

calculate_gc_content(sequence: str) → float

Calculate GC content using shared utility.

find_orfs(sequence: str, min_length: int = 150) → list[ORFInfo]

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

translate_sequence(sequence: str) → str

Translate DNA sequence to protein.

async get_additional_sequence(transcript_id: str, sequence_type: SequenceType) → str | None

Retrieve specific sequence type using the database client if available.

Parameters:

• transcript_id – Transcript identifier

• sequence_type – Type of sequence to retrieve

Returns:

Sequence string or None if not available or client not provided

async analyze_transcript(transcript: TranscriptInfo) → SequenceAnalysis

Perform complete ORF analysis of a transcript.

async analyze_transcripts(transcripts: list[TranscriptInfo]) → dict[str, SequenceAnalysis]

Analyze multiple transcripts.

sirnaforge.data.orf_analysis.create_orf_analyzer(database_client: AbstractDatabaseClient | None = None) → ORFAnalyzer

Create an ORF analyzer with optional database client.

Parameters:

database_client – Optional database client for retrieving additional sequence types

Returns:

ORFAnalyzer instance

async sirnaforge.data.orf_analysis.analyze_multiple_transcript_orfs(transcripts: list[TranscriptInfo], database_client: AbstractDatabaseClient | None = None) → dict[str, SequenceAnalysis]

Analyze ORFs in multiple transcripts.

Parameters:

• transcripts – List of transcripts to analyze

• database_client – Optional database client for additional sequence retrieval

Returns:

Dictionary mapping transcript IDs to SequenceAnalysis results

Genome Manager

Genome reference manager built on the composable transcriptome manager core.

class sirnaforge.data.genome_manager.GenomeManager(cache_dir: str | Path | None = None, cache_ttl_days: int = 30, auto_build_indices: bool = True)

Bases: TranscriptomeManager

Genome FASTA manager with caching and optional BWA-MEM2 index building.

SOURCE_LABEL = 'genome'

SOURCES = {'ensembl_human_hg38_primary': TranscriptomeSource(name='ensembl_human_hg38_primary', url='https://ftp.ensembl.org/pub/current_fasta/homo_sapiens/dna/Homo_sapiens.GRCh38.dna.primary_assembly.fa.gz', species='human', format='fasta', compressed=True, description='Ensembl human GRCh38 primary assembly genomic reference'), 'ensembl_macaque_mmul10_toplevel': TranscriptomeSource(name='ensembl_macaque_mmul10_toplevel', url='https://ftp.ensembl.org/pub/current_fasta/macaca_mulatta/dna/Macaca_mulatta.Mmul_10.dna.toplevel.fa.gz', species='macaque', format='fasta', compressed=True, description='Ensembl rhesus macaque Mmul_10 toplevel assembly genomic reference'), 'ensembl_mouse_grcm39_primary': TranscriptomeSource(name='ensembl_mouse_grcm39_primary', url='https://ftp.ensembl.org/pub/current_fasta/mus_musculus/dna/Mus_musculus.GRCm39.dna.primary_assembly.fa.gz', species='mouse', format='fasta', compressed=True, description='Ensembl mouse GRCm39 primary assembly genomic reference'), 'ensembl_rat_grcr8_toplevel': TranscriptomeSource(name='ensembl_rat_grcr8_toplevel', url='https://ftp.ensembl.org/pub/current_fasta/rattus_norvegicus/dna/Rattus_norvegicus.GRCr8.dna.toplevel.fa.gz', species='rat', format='fasta', compressed=True, description='Ensembl rat GRCr8 toplevel assembly genomic reference')}

__init__(cache_dir: str | Path | None = None, cache_ttl_days: int = 30, auto_build_indices: bool = True)

Initialize genome manager.

Parameters:

• cache_dir – Directory for genome cache

• cache_ttl_days – Cache time-to-live in days

• auto_build_indices – Automatically build BWA-MEM2 indices when missing

get_genome(source_name: str, force_refresh: bool = False, build_index: bool = True) → dict[str, Path] | None

Get genome reference, downloading and indexing as needed.

get_custom_genome(fasta_path: str | Path, build_index: bool = True, cache_name: str | None = None) → dict[str, Path] | None

Get custom genome FASTA through shared caching/indexing pipeline.

Reference Manager

Base Reference Database Manager for shared caching functionality.

This module provides the foundation for managing reference databases (miRNA, transcriptome, genome) with automatic caching, download management, and integrity validation.

class sirnaforge.data.reference_manager.ReferenceSource(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '')

Bases: object

Base configuration for a reference database source.

name: str

url: str

species: str

format: str = 'fasta'

compressed: bool = False

description: str = ''

cache_key() → str

Generate a unique cache key for this source.

__init__(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '') → None

class sirnaforge.data.reference_manager.CacheMetadata(source: ReferenceSource, downloaded_at: str, file_size: int, checksum: str, file_path: str, version: str = '1.0', extra: dict[str, Any] | None = None)

Bases: object

Metadata for cached reference files.

source: ReferenceSource

downloaded_at: str

file_size: int

checksum: str

file_path: str

version: str = '1.0'

extra: dict[str, Any] | None = None

classmethod from_dict(data: dict[str, ~typing.Any], source_class: type[~sirnaforge.data.reference_manager.ReferenceSource] = <class 'sirnaforge.data.reference_manager.ReferenceSource'>) → CacheMetadata

Create CacheMetadata from dictionary.

Parameters:

• data – Dictionary containing metadata

• source_class – Class to use for source (allows subclasses)

to_dict() → dict

Convert CacheMetadata to dictionary.

__init__(source: ReferenceSource, downloaded_at: str, file_size: int, checksum: str, file_path: str, version: str = '1.0', extra: dict[str, Any] | None = None) → None

class sirnaforge.data.reference_manager.ReferenceManager(cache_subdir: str, cache_dir: str | Path | None = None, cache_ttl_days: int = 30, artifact_class: str | None = None)

Bases: ABC, Generic[SourceT]

Base class for reference database managers with caching support.

This provides common functionality for: - Cache directory management with multiple fallback locations - File downloading with retry and progress - Checksum validation - TTL-based cache invalidation - Metadata persistence

Subclasses should implement source-specific operations like filtering, indexing, or multi-source aggregation.

__init__(cache_subdir: str, cache_dir: str | Path | None = None, cache_ttl_days: int = 30, artifact_class: str | None = None)

Initialize the reference manager.

Parameters:

• cache_subdir – Subdirectory name under cache root (e.g., ‘mirna’, ‘transcriptomes’)

• cache_dir – Directory for caching databases (default: ~/.cache/sirnaforge/{cache_subdir})

• cache_ttl_days – Cache time-to-live in days

• artifact_class – Producer-version scope for this cache (default: cache_subdir). One subdirectory holds one kind of artifact, so the subdirectory name is also the natural invalidation scope; override only when two managers share a directory but not a producer.

clear_cache(confirm: bool = False) → dict[str, Any]

Clear all cached files for this manager.

By default this only reports what would be removed. Set confirm=True to actually delete files.

is_producer_version_current(meta: CacheMetadata) → bool

Whether meta was written by the producer version this build ships.

Public because it is the one question every cache consumer needs to be able to ask (and test) directly, independently of TTL and checksums.

stale_producer_entries() → list[str]

Cache keys whose recorded producer version is no longer current.

cache_info() → dict[str, Any]

Get information about the current cache state.

Returns:

Dictionary containing cache statistics

clean_cache(older_than_days: int | None = None) → None

Clean old cache files.

Parameters:

older_than_days – Remove files older than this (default: use TTL)

Transcriptome Manager

Transcriptome Database Manager with local caching and automatic index building.

This module provides a clean interface for downloading, caching, and managing transcriptome FASTA files with automatic BWA-MEM2 index building and cache management.

class sirnaforge.data.transcriptome_manager.TranscriptomeSource(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '')

Bases: ReferenceSource

Transcriptome-specific database source configuration.

Inherits from ReferenceSource with transcriptome-specific extensions.

__init__(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '') → None

class sirnaforge.data.transcriptome_manager.TranscriptomeManager(cache_dir: str | Path | None = None, cache_ttl_days: int = 90, auto_build_indices: bool = True, local_content_dedupe: bool = True, cache_subdir: str = 'transcriptomes', sources: dict[str, TranscriptomeSource] | None = None, source_label: str | None = None)

Bases: ReferenceManager[TranscriptomeSource]

Transcriptome database manager with caching and automatic BWA-MEM2 index building.

SOURCE_LABEL = 'transcriptome'

INDEX_SUFFIXES = ('.0123', '.amb', '.ann', '.bwt.2bit.64', '.pac')

FILTER_URI_FRAGMENT = '#filters='

UNREMOVABLE_INDEX_KEY = 'unremovable_stale_index'

SOURCES = {'ensembl_human_cdna': TranscriptomeSource(name='ensembl_cdna', url='https://ftp.ensembl.org/pub/current_fasta/homo_sapiens/cdna/Homo_sapiens.GRCh38.cdna.all.fa.gz', species='human', format='fasta', compressed=True, description='Ensembl human cDNA sequences (GRCh38)'), 'ensembl_macaque_cdna': TranscriptomeSource(name='ensembl_cdna', url='https://ftp.ensembl.org/pub/current_fasta/macaca_mulatta/cdna/Macaca_mulatta.Mmul_10.cdna.all.fa.gz', species='macaque', format='fasta', compressed=True, description='Ensembl rhesus macaque cDNA sequences (Mmul_10)'), 'ensembl_mouse_cdna': TranscriptomeSource(name='ensembl_cdna', url='https://ftp.ensembl.org/pub/current_fasta/mus_musculus/cdna/Mus_musculus.GRCm39.cdna.all.fa.gz', species='mouse', format='fasta', compressed=True, description='Ensembl mouse cDNA sequences (GRCm39)'), 'ensembl_rat_cdna': TranscriptomeSource(name='ensembl_cdna', url='https://ftp.ensembl.org/pub/current_fasta/rattus_norvegicus/cdna/Rattus_norvegicus.GRCr8.cdna.all.fa.gz', species='rat', format='fasta', compressed=True, description='Ensembl rat cDNA sequences (GRCr8)')}

__init__(cache_dir: str | Path | None = None, cache_ttl_days: int = 90, auto_build_indices: bool = True, local_content_dedupe: bool = True, cache_subdir: str = 'transcriptomes', sources: dict[str, TranscriptomeSource] | None = None, source_label: str | None = None)

Initialize the transcriptome database manager.

Parameters:

• cache_dir – Directory for caching transcriptomes (default: ~/.cache/sirnaforge/transcriptomes)

• cache_ttl_days – Cache time-to-live in days (default: 90 days for large files)

• auto_build_indices – Automatically build BWA-MEM2 indices when missing

• local_content_dedupe – Reuse cached local files by content hash across different input paths

• cache_subdir – Cache subdirectory override for composable manager reuse

• sources – Optional source dictionary override

• source_label – Optional source label used in status/error reporting

sources: dict[str, TranscriptomeSource]

local_content_index: dict[str, str]

describe_source_status(source_name: str) → dict[str, Any]

Return cache metadata for a configured source.

describe_sources_status(source_names: list[str] | tuple[str, ...] | None = None) → list[dict[str, Any]]

Return cache statuses for multiple transcriptome sources.

get_transcriptome(source_name: str, force_refresh: bool = False, build_index: bool = True) → dict[str, Path] | None

Get transcriptome database, downloading and building index if needed.

Parameters:

• source_name – Pre-configured source name (e.g., “ensembl_human_cdna”)

• force_refresh – Force re-download even if cached

• build_index – Build BWA-MEM2 index if missing

Returns:

Dictionary with ‘fasta’ and optionally ‘index’ paths, or None if failed

get_custom_transcriptome(fasta_path: str | Path, build_index: bool = True, cache_name: str | None = None) → dict[str, Path] | None

Process a custom transcriptome FASTA with caching and index building.

Parameters:

• fasta_path – Path or URL to transcriptome FASTA file

• build_index – Build BWA-MEM2 index if missing

• cache_name – Custom cache name (default: derived from filename)

Returns:

Dictionary with ‘fasta’ and optionally ‘index’ paths, or None if failed

get_filtered_transcriptome(source_name: str, filters: list[str], force_refresh: bool = False, build_index: bool = True) → dict[str, Path] | None

Get a filtered transcriptome with caching.

Parameters:

• source_name – Pre-configured source name (e.g., “ensembl_human_cdna”)

• filters – List of filter names (e.g., [‘protein_coding’, ‘canonical_only’])

• force_refresh – Force re-download and re-filter

• build_index – Build BWA-MEM2 index if missing

Returns:

Dictionary with ‘fasta’ and optionally ‘index’ paths, or None if failed

list_available_sources() → dict[str, TranscriptomeSource]

List all pre-configured transcriptome sources.

Transcriptome Filter

Transcriptome filtering utilities for reducing reference size.

Provides functions to filter transcriptome FASTA files based on: - Biotype (protein_coding, etc.) - Canonical transcript status - Specific regions (e.g., 3’ UTR)

This helps reduce memory requirements for BWA-MEM2 indexing on low-RAM machines.

class sirnaforge.data.transcriptome_filter.TranscriptFilter

Bases: object

Filter transcripts based on header metadata and biotype.

static parse_ensembl_header(description: str) → dict[str, str]

Parse Ensembl FASTA header to extract metadata.

Parameters:

description – Full FASTA description line (without ‘>’)

Returns:

Dictionary of parsed fields (gene_biotype, transcript_biotype, gene_symbol, etc.)

static is_protein_coding(record: SeqRecord) → bool

Check if transcript is protein coding.

Parameters:

record – SeqRecord from FASTA

Returns:

True if protein_coding, False otherwise

static is_canonical(record: SeqRecord) → bool

Check if transcript is canonical (MANE Select or Ensembl canonical).

Parameters:

record – SeqRecord from FASTA

Returns:

True if canonical, False otherwise

static filter_fasta(input_fasta: Path, output_fasta: Path, filter_func: Callable[[SeqRecord], bool], filter_name: str = 'custom') → int

Filter FASTA file using a custom filter function.

Parameters:

• input_fasta – Path to input FASTA file

• output_fasta – Path to output FASTA file

• filter_func – Function that returns True for records to keep

• filter_name – Name of filter for logging

Returns:

Number of sequences kept

static apply_protein_coding_filter(input_fasta: Path, output_fasta: Path) → int

Filter to protein-coding transcripts only.

Parameters:

• input_fasta – Path to input FASTA file

• output_fasta – Path to output FASTA file

Returns:

Number of sequences kept

static apply_canonical_filter(input_fasta: Path, output_fasta: Path) → int

Filter to canonical transcripts only.

Parameters:

• input_fasta – Path to input FASTA file

• output_fasta – Path to output FASTA file

Returns:

Number of sequences kept

static apply_combined_filter(input_fasta: Path, output_fasta: Path, filters: list[str]) → int

Apply multiple filters in sequence.

Parameters:

• input_fasta – Path to input FASTA file

• output_fasta – Path to output FASTA file

• filters – List of filter names (e.g., [‘protein_coding’, ‘canonical_only’])

Returns:

Number of sequences kept

sirnaforge.data.transcriptome_filter.get_filter_spec(filter_string: str | None) → list[str]

Parse filter specification string into list of filter names.

Parameters:

filter_string – Comma-separated filter names (e.g., “protein_coding,canonical_only”)

Returns:

List of filter names, or empty list if None

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: int = 30, base_url: str = 'https://rest.ensembl.org', cache_ttl: int = 3600, max_cache_entries: int = 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}")

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

Initialize Ensembl transcript annotation client.

Parameters:

• timeout – Request timeout in seconds

• base_url – Ensembl REST API base URL

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

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

async fetch_by_ids(ids: list[str], *, species: str, reference: ReferenceChoice) → TranscriptAnnotationBundle

Fetch transcript annotations by stable IDs using Ensembl lookup endpoint.

Parameters:

• ids – List of transcript or gene IDs

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

• reference – Reference assembly/release choice

Returns:

TranscriptAnnotationBundle with resolved annotations

async fetch_by_regions(regions: list[str], *, species: str, reference: ReferenceChoice) → TranscriptAnnotationBundle

Fetch transcript annotations by genomic regions using Ensembl overlap endpoint.

Parameters:

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

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

• reference – Reference assembly/release choice

Returns:

TranscriptAnnotationBundle with all transcripts overlapping regions

class sirnaforge.data.transcript_annotation.VepConsequenceClient(timeout: int = 30, base_url: str = '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

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

Initialize VEP client.

Parameters:

• timeout – Request timeout in seconds

• base_url – Ensembl REST API base URL

async enrich_annotations(bundle: TranscriptAnnotationBundle, _species: str = 'homo_sapiens') → TranscriptAnnotationBundle

Enrich transcript annotations with VEP consequence data.

Parameters:

• bundle – Existing transcript annotation bundle

• species – Species name for VEP queries

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: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '')

Bases: ReferenceSource

miRNA-specific database source configuration.

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

__init__(name: str, url: str, species: str, format: str = 'fasta', compressed: bool = False, description: str = '') → None

class sirnaforge.data.mirna_manager.MiRNADatabaseManager(cache_dir: str | Path | None = None, cache_ttl_days: int = 30)

Bases: ReferenceManager[MiRNASource]

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

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() → list[str]

Return sorted list of supported database sources.

classmethod get_all_species() → list[str]

Return sorted list of all species across sources.

classmethod get_species_for_source(source_name: str) → list[str]

Return sorted list of species supported by a given source.

classmethod get_species_aliases(source_name: str) → dict[str, list[str]]

Return mapping of canonical species identifiers to their known aliases.

classmethod get_canonical_species() → list[str]

Return sorted list of canonical species keys.

classmethod canonicalize_species_name(species: str) → str | None

Normalize a raw species identifier to a canonical key.

classmethod canonicalize_species_list(species_list: Sequence[str]) → list[str]

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

classmethod get_genome_species_for_canonical(canonical_species: Sequence[str]) → list[str]

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.

classmethod get_mirna_slugs_for_canonical(canonical_species: Sequence[str], source_name: str) → list[str]

Return normalized miRNA identifiers for canonical species.

classmethod get_supported_canonical_species_for_source(source_name: str) → list[str]

Return canonical species supported by a given source.

classmethod resolve_species_selection(requested_species: Sequence[str], source_name: str, mirna_overrides: Sequence[str] | None = None) → dict[str, list[str]]

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

classmethod normalize_species(source_name: str, species: str) → str | None

Normalize user-provided species identifiers to canonical keys.

classmethod get_source_configuration(source_name: str, species: str) → MiRNASource | None

Retrieve the MiRNASource configuration for a given source/species.

classmethod is_supported_source(source_name: str) → bool

Check if a source is supported.

classmethod is_supported_species(source_name: str, species: str) → bool

Check if a species is supported for the given source.

classmethod get_mirgenedb_species_metadata() → dict[str, dict[str, Any]]

Expose the MirGeneDB species metadata table.

__init__(cache_dir: str | Path | None = None, cache_ttl_days: int = 30)

Initialize the miRNA database manager.

Parameters:

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

• cache_ttl_days – Cache time-to-live in days

get_database(source_name: str, species: str, force_refresh: bool = False) → Path | None

Get miRNA database, downloading and filtering if needed.

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

Parameters:

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

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

• force_refresh – Force re-download even if cached

Returns:

Path to cached FASTA file, or None if failed

get_combined_database(sources: list[str], species: str, output_name: str | None = None) → Path | None

Combine multiple databases into a single file.

Parameters:

• sources – List of source names to combine

• species – Target species

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

Returns:

Path to combined FASTA file

list_available_databases() → dict[str, dict[str, MiRNASource]]

List all available database sources and species.

clear_cache(confirm: bool = False) → dict[str, Any]

Clear miRNA cache using the shared reference-manager implementation.

sirnaforge.data.mirna_manager.main() → None

CLI interface for the miRNA database manager.

Species Registry

Canonical species registry and metadata for miRNA and genome mappings.

sirnaforge.data.species_registry.normalize_species_name(species: str) → str

Normalize species name to canonical form.

Parameters:

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

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'

Variant Resolver

Variant resolution from multiple databases (ClinVar, Ensembl, dbSNP) with caching.

class sirnaforge.data.variant_resolver.VariantResolver(min_af: float = 0.01, clinvar_filters: list[ClinVarSignificance] | None = None, assembly: str = 'GRCh38', source_priority: list[VariantSource] | None = None, cache_dir: Path | None = None, timeout: int = 30, variant_mode: str | None = None)

Bases: object

Resolve variant identifiers to VariantRecord using multiple databases with priority ordering.

Priority order: ClinVar > Ensembl > dbSNP Supports caching and local VCF files. Only supports GRCh38 assembly.

__init__(min_af: float = 0.01, clinvar_filters: list[ClinVarSignificance] | None = None, assembly: str = 'GRCh38', source_priority: list[VariantSource] | None = None, cache_dir: Path | None = None, timeout: int = 30, variant_mode: str | None = None)

Initialize variant resolver.

Parameters:

• min_af – Minimum allele frequency threshold (default: 0.01)

• clinvar_filters – Allowed ClinVar significance levels (default: Pathogenic, Likely pathogenic)

• assembly – Reference genome assembly (only GRCh38 supported)

• source_priority – Source priority list (default: ClinVar > Ensembl > dbSNP)

• cache_dir – Cache directory for variant data

• timeout – HTTP request timeout in seconds

• variant_mode – Variant mode for AF filtering (‘avoid’, ‘target’, ‘both’). In ‘avoid’ mode, uses max population AF if available to avoid SNPs prevalent in any geographic group (e.g., >10% in one population).

parse_identifier(input_str: str) → VariantQuery

Parse variant identifier string into VariantQuery.

Accepts: - rsID: rs12345 - VCF-style coordinate: chr17:7577121:G:A or 17:7577121:G:A - HGVS: NM_000546.6:c.215C>G (basic support)

Parameters:

input_str – User-provided variant identifier

Returns:

VariantQuery with parsed components

Raises:

ValueError – If input format is not recognized

async resolve_variant(query: VariantQuery) → VariantRecord | None

Resolve a variant query to a VariantRecord.

Tries sources in priority order: ClinVar -> Ensembl -> dbSNP Applies AF and ClinVar filters. Uses cache when available.

Parameters:

query – Parsed variant query

Returns:

VariantRecord if found and passes filters, None otherwise

read_vcf(vcf_path: Path) → list[VariantRecord]

Read variants from VCF file (supports bgzip+tabix).

Parameters:

vcf_path – Path to VCF file

Returns:

List of VariantRecord objects passing filters

Raises:

FileNotFoundError – If VCF file doesn’t exist

sirnaforge.data.variant_resolver.resolve_variant_sync(variant_id: str, min_af: float = 0.01, clinvar_filters: list[ClinVarSignificance] | None = None, cache_dir: Path | None = None, variant_mode: str | None = None) → VariantRecord | None

Synchronous wrapper for variant resolution.

Parameters:

• variant_id – Variant identifier (rsID, coordinate, or HGVS)

• min_af – Minimum allele frequency

• clinvar_filters – Allowed ClinVar significance levels

• cache_dir – Cache directory

• variant_mode – Variant mode for AF filtering (‘avoid’, ‘target’, ‘both’)

Returns:

VariantRecord if found and passes filters, None otherwise

Variant Cache

Improved variant caching using Parquet for efficient storage and retrieval.

class sirnaforge.data.variant_cache.VariantParquetCache(cache_dir: Path, ttl_days: int = 90)

Bases: object

Efficient variant cache using Parquet files for better performance than JSON.

Benefits over JSON: - Columnar storage format is much more efficient for variant data - Built-in compression reduces disk usage - Fast filtering and querying with pandas - Batch operations instead of individual file I/O

__init__(cache_dir: Path, ttl_days: int = 90)

Initialize the Parquet-based variant cache.

Parameters:

• cache_dir – Directory for cache storage

• ttl_days – Time-to-live for cached entries in days (default: 90)

get(cache_key: str) → VariantRecord | None

Retrieve a variant from cache by key.

Parameters:

cache_key – Cache key for the variant

Returns:

VariantRecord if found and not stale, None otherwise

put(cache_key: str, variant: VariantRecord) → None

Store a variant in the cache.

Parameters:

• cache_key – Cache key for storage

• variant – VariantRecord to cache

cleanup_stale_entries() → int

Remove entries older than TTL.

Returns:

Number of entries removed

get_stats() → dict[str, Any]

Get cache statistics.

Returns:

Dictionary with cache statistics

clear() → None

Clear all cache entries.

Variant Helpers

Helper functions for generating sequence contexts with variant alleles.

sirnaforge.data.variant_helpers.generate_contexts_for_variant(variant: VariantRecord, reference_sequence: str, transcript_start: int, transcript_id: str, flank_size: int = 50) → dict[str, tuple[str, int, int]]

Generate reference and alternate sequence contexts for a variant.

Parameters:

• variant – Variant record with position and alleles

• reference_sequence – Full reference transcript sequence

• transcript_start – Genomic start position of the transcript (1-based)

• transcript_id – Transcript identifier

• flank_size – Number of nucleotides to include on each side of variant (default: 50)

Returns:

• sequence context (str)

• relative position of variant in context (int, 0-based)

• length of the allele (int)

Return type:

Dictionary with ‘ref’ and ‘alt’ keys, each containing

Raises:

ValueError – If variant position is outside transcript boundaries

sirnaforge.data.variant_helpers.check_candidate_overlaps_variant(candidate_pos: int, candidate_length: int, variant: VariantRecord, transcript_start: int) → bool

Check if a siRNA candidate overlaps with a variant position.

Parameters:

• candidate_pos – 1-based start position of candidate in transcript

• candidate_length – Length of the siRNA candidate

• variant – Variant record to check

• transcript_start – 1-based genomic start position of transcript

Returns:

True if candidate overlaps the variant, False otherwise

sirnaforge.data.variant_helpers.annotate_candidate_with_variant(candidate: SiRNACandidate, variant: VariantRecord, allele: str, variant_mode: str) → None

Annotate a siRNA candidate with variant information.

Modifies the candidate in place to add variant-specific metadata.

Parameters:

• candidate – SiRNACandidate to annotate

• variant – Variant record that overlaps the candidate

• allele – Which allele the candidate targets (‘ref’ or ‘alt’)

• variant_mode – Variant handling mode (‘target’, ‘avoid’, ‘both’)

sirnaforge.data.variant_helpers.apply_variant_to_sequence(sequence: str, variant: VariantRecord, transcript_start: int, allele: str = 'alt') → str

Apply a variant to a reference sequence to generate an alternate sequence.

Parameters:

• sequence – Reference sequence

• variant – Variant record with position and alleles

• transcript_start – 1-based genomic start position of transcript

• allele – Which allele to apply (‘ref’ returns unchanged sequence, ‘alt’ applies variant)

Returns:

Modified sequence with variant applied

Raises:

ValueError – If variant position is outside sequence boundaries or allele is invalid

sirnaforge.data.variant_helpers.get_variant_position_in_transcript(variant: VariantRecord, transcript_start: int) → int

Calculate the 0-based position of a variant within a transcript.

Parameters:

• variant – Variant record

• transcript_start – 1-based genomic start position of transcript

Returns:

0-based position of variant within transcript

Pipeline Integration

Nextflow CLI

Command-line entry points used by embedded Nextflow modules.

sirnaforge.pipeline.nextflow_cli.build_bwa_index_cli(fasta_file: str, species: str, output_dir: str = '.') → dict[str, Any]

Build BWA-MEM2 index for genome/transcriptome.

Parameters:

• fasta_file – Path to input FASTA file

• species – Species identifier

• output_dir – Directory to write index files

Returns:

Dictionary with index prefix path

sirnaforge.pipeline.nextflow_cli.aggregate_results_cli(genome_species: str, output_dir: str = '.', mirna_db: str | None = None, mirna_species: str | None = None, analysis_files: list[str] | None = None, summary_files: list[str] | None = None) → dict[str, Any]

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

Parameters:

• genome_species – Comma-separated list of species

• output_dir – Directory to write aggregated results

• mirna_db – The database that provided the reference

• mirna_species – The species code for the matching miRNA

• analysis_files – Optional explicit list of staged analysis files to aggregate

• summary_files – Optional explicit list of staged summary files to aggregate

Returns:

Dictionary with aggregation statistics

sirnaforge.pipeline.nextflow_cli.aggregate_mirna_results_cli(mirna_db: str, mirna_species: str, results_dir: str = '.', output_dir: str = '.') → dict[str, Any]

Aggregate miRNA seed analysis results from multiple candidates.

Parameters:

• mirna_db – miRNA database name used for analysis

• mirna_species – Comma-separated list of species analyzed

• results_dir – Directory containing individual miRNA results

• output_dir – Directory to write aggregated results

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(*, running_in_docker: bool, docker_available: bool, requested_profile: str, recommended_profile: str, docker_image: str | None = None, profile_override_reason: str | None = None)

Bases: BaseModel

Information about the current execution environment.

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() → bool

Check if the recommended profile differs from the requested profile.

get_execution_summary() → str

Get a human-readable summary of the execution environment.

model_config = {}

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

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

Bases: object

Configuration manager for Nextflow workflows.

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

MEMORY_BUFFER_GB = 0.5

MIN_MEMORY_GB = 1

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

Initialize Nextflow configuration.

Parameters:

• docker_image – Docker container image to use

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

• work_dir – Working directory for Nextflow execution

• nxf_home – Nextflow home cache

• max_cpus – Maximum CPU cores

• max_memory – Maximum memory allocation

• max_time – Maximum execution time

• **kwargs – Additional configuration parameters

get_nextflow_args(input_file: Path, output_dir: Path, genome_species: list[str], additional_params: dict[str, Any] | None = None, include_test_profile: bool = False) → list[str]

Generate Nextflow command arguments.

Parameters:

• input_file – Input FASTA file path

• output_dir – Output directory

• genome_species – List of species for miRNA genome lookups (not genomic DNA)

• additional_params – Additional parameters to pass

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

Returns:

List of command arguments for Nextflow

create_config_file(config_path: Path) → Path

Create a custom Nextflow configuration file.

Parameters:

config_path – Path where to create the config file

Returns:

Path to the created configuration file

validate_docker_available() → bool

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.

Returns:

True if Docker is available and accessible for Nextflow

is_running_in_docker() → bool

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

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

Returns:

True if running inside a Docker container

get_execution_profile() → str

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

Returns:

Recommended execution profile

get_environment_info() → EnvironmentInfo

Get information about the current execution environment.

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

Returns:

EnvironmentInfo model with environment details

classmethod for_testing() → NextflowConfig

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.

Returns:

NextflowConfig instance with test-friendly settings

classmethod for_production(**kwargs: Any) → NextflowConfig

Create a configuration optimized for production use.

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

Parameters:

**kwargs – Additional configuration parameters to override defaults

Returns:

NextflowConfig instance with production settings

classmethod auto_configure(**kwargs: Any) → NextflowConfig

Auto-configure Nextflow settings based on environment detection.

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

Parameters:

**kwargs – Additional configuration parameters to override defaults

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: NextflowConfig | None = None)

Bases: object

Execute Nextflow workflows from Python with proper error handling.

__init__(config: NextflowConfig | None = None)

Initialize Nextflow runner.

Parameters:

config – NextflowConfig instance, creates auto-configured if None

get_main_workflow() → Path

Get path to the main Nextflow workflow.

get_pipeline_revision() → str

Expose the detected pipeline revision identifier.

async run(input_file: Path, output_dir: Path, genome_species: list[str] | None = None, **kwargs: Any) → dict[str, Any]

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

Parameters:

• input_file – Path to input FASTA file

• output_dir – Output directory for results

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

• **kwargs – Additional parameters passed to run_offtarget_analysis

Returns:

Dictionary containing execution results and metadata

Raises:

NextflowExecutionError – If workflow execution fails

run_sync(input_file: Path, output_dir: Path, genome_species: list[str] | None = None, **kwargs: Any) → dict[str, Any]

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

Parameters:

• input_file – Path to input FASTA file

• output_dir – Output directory for results

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

• **kwargs – Additional parameters passed to run_offtarget_analysis

Returns:

Dictionary containing execution results and metadata

async run_offtarget_analysis(input_file: Path, output_dir: Path, genome_species: list[str], additional_params: dict[str, Any] | None = None, show_progress: bool = True) → dict[str, Any]

Run the off-target analysis Nextflow workflow.

Parameters:

• input_file – Path to siRNA candidates FASTA file

• output_dir – Output directory for results

• genome_species – List of species for miRNA genome lookups

• additional_params – Additional parameters for the workflow

• show_progress – Whether to show progress indicators

Returns:

Dictionary containing execution results and metadata

Raises:

NextflowExecutionError – If workflow execution fails

validate_installation() → dict[str, bool | str]

Validate that Nextflow and required tools are available.

Returns:

Dictionary of tool availability status

classmethod for_testing() → NextflowRunner

Create a runner configured for testing.

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

Returns:

NextflowRunner configured for testing

classmethod create(**config_kwargs: Any) → NextflowRunner

Create a NextflowRunner with auto-configured settings.

Parameters:

**config_kwargs – Configuration parameters for NextflowConfig

Returns:

NextflowRunner with auto-configured NextflowConfig

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

Bases: Exception

Exception raised when Nextflow execution fails.

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

Initialize the exception with message and optional stdout/stderr.

Parameters:

• message – Error message describing what went wrong

• stdout – Standard output from the failed command

• stderr – Standard error from the failed command

Pipeline Resources

Resource Management for siRNAforge Pipeline.

This module provides utilities for managing test data and pipeline resources.

sirnaforge.pipeline.resources.resources.get_resource_path(resource_name: str) → Path

Get path to a pipeline resource file.

Parameters:

resource_name – Name of the resource file

Returns:

Path to the resource file

Raises:

FileNotFoundError – If resource is not found

sirnaforge.pipeline.resources.resources.get_test_data_path(filename: str) → Path

Get path to test data file.

Parameters:

filename – Name of the test data file

Returns:

Path to the test data file

Raises:

FileNotFoundError – If test data file is not found

sirnaforge.pipeline.resources.resources.validate_test_data() → dict[str, bool]

Validate that required test data files are present.

Returns:

Dictionary mapping filenames to availability status

class sirnaforge.pipeline.resources.resources.ResourceManager

Bases: object

Manage pipeline resources and test data.

__init__() → None

Initialize resource manager.

ensure_test_data() → bool

Ensure test data directory exists and contains required files.

Returns:

True if all required test data is available

get_test_config() → dict[str, str]

Get test configuration with paths to test data.

Returns:

Dictionary containing test data paths

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: Path, gene_query: str, input_fasta: Path | None = None, database: DatabaseType = DatabaseType.ENSEMBL, design_params: DesignParameters | None = None, nextflow_config: Mapping[str, Any] | None = None, genome_indices_override: str | None = None, genome_species: list[str] | None = None, query_species: str | None = None, mirna_database: str = 'mirgenedb', mirna_species: Sequence[str] | None = None, transcriptome_fasta: str | None = None, transcriptome_filter: str | None = None, transcriptome_selection: ReferenceSelection | None = None, validation_config: ValidationConfig | None = None, log_file: str | None = None, write_json_summary: bool = True, num_threads: int | None = None, input_source: InputSource | None = None, keep_nextflow_work: bool = False, variant_config: VariantWorkflowConfig | None = None, zfn_config: ZFNWorkflowConfig | None = None)

Bases: object

Configuration for the complete siRNA design workflow.

__init__(output_dir: Path, gene_query: str, input_fasta: Path | None = None, database: DatabaseType = DatabaseType.ENSEMBL, design_params: DesignParameters | None = None, nextflow_config: Mapping[str, Any] | None = None, genome_indices_override: str | None = None, genome_species: list[str] | None = None, query_species: str | None = None, mirna_database: str = 'mirgenedb', mirna_species: Sequence[str] | None = None, transcriptome_fasta: str | None = None, transcriptome_filter: str | None = None, transcriptome_selection: ReferenceSelection | None = None, validation_config: ValidationConfig | None = None, log_file: str | None = None, write_json_summary: bool = True, num_threads: int | None = None, input_source: InputSource | None = None, keep_nextflow_work: bool = False, variant_config: VariantWorkflowConfig | None = None, zfn_config: ZFNWorkflowConfig | None = None)

Initialize workflow configuration.

class sirnaforge.workflow.ZFNWorkflowConfig(zfn_params: ZFNDesignParameters, annotation: GenomicAnnotationConfig | None = None)

Bases: object

Configuration for ZFN pair evaluation and off-target search workflow.

This carries the scientifically distinct ZFN parameters: - Left/right half-site sequences (IUPAC-validated, 9-18 bp) - Genomic search space (whole-genome FASTA) - Algorithm choice (homology / conserved_g / zfn_v2) - Spacer/dimer/mismatch constraints - Optional genomic annotation for region classification

__init__(zfn_params: ZFNDesignParameters, annotation: GenomicAnnotationConfig | None = None)

Initialize ZFN workflow configuration.

class sirnaforge.workflow.SiRNAWorkflow(config: WorkflowConfig)

Bases: object

Main workflow orchestrator for siRNA/miRNA/ZFN design pipeline.

__init__(config: WorkflowConfig)

Initialize the workflow orchestrator.

async run_complete_workflow() → dict[str, Any]

Run the complete design workflow (siRNA/miRNA or ZFN).

async step1_retrieve_transcripts(progress: Progress) → list[TranscriptInfo]

Step 1: Retrieve and validate transcript sequences.

async resolve_variants_step(progress: Progress) → list[VariantRecord]

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 – Rich progress tracker

Returns:

List of resolved VariantRecords that passed filters

async step2_validate_orfs(transcripts: list[TranscriptInfo], progress: Progress) → dict[str, Any]

Step 2: Validate ORFs and generate validation report.

async step3_design_sirnas(transcripts: list[TranscriptInfo], progress: Progress) → DesignResult

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.

async step6_generate_reports(design_results: DesignResult) → None

Step 6: Generate comprehensive reports.

async step5_offtarget_analysis(design_results: DesignResult) → dict[str, Any]

Step 5: Detect repeat elements, then run off-target analysis via the Nextflow pipeline.

Repeat detection runs here rather than as its own workflow step so it can reuse the transcriptome reference this step already materializes for screening, instead of fetching it a second time just to locate the query species’ FASTA. The corollary is that check_off_targets=False (--skip-off-targets) skips repeat detection as well: both are reference-based scans, and the reference is what the flag exists to avoid paying for.

async run_nextflow_offtarget_analysis(candidates: list[SiRNACandidate], input_fasta: Path) → dict[str, Any]

Public wrapper for Nextflow off-target analysis execution.

sirnaforge.workflow.apply_zfn_runtime_overrides(zfn_design_params: ZFNDesignParameters, nextflow_config_overrides: dict[str, Any]) → ZFNDesignParameters

Apply non-CLI ZFN sharding/runtime overrides and project Nextflow params.

Starts from zfn_design_params.sharding (typed defaults are authoritative), merges optional JSON overrides from SIRNAFORGE_ZFN_SHARDING_JSON, and mirrors the resolved sharding values into nextflow_config_overrides so the Nextflow route and direct Python route share the same effective config.

The runtime search implementation remains generic and contig-aware, including chunk sharding on single-contig inputs when sharding is enabled.

async sirnaforge.workflow.run_sirna_workflow(gene_query: str, output_dir: str, input_fasta: str | None = None, database: str = 'ensembl', design_mode: str = 'sirna', top_n_candidates: int | None = None, genome_species: list[str] | None = None, query_species: str | None = None, genome_indices_override: str | None = None, mirna_database: str = 'mirgenedb', mirna_species: Sequence[str] | None = None, transcriptome_fasta: str | None = None, transcriptome_filter: str | None = None, transcriptome_selection: ReferenceSelection | None = None, gc_min: float = 30.0, gc_max: float = 52.0, sirna_length: int = 21, modification_pattern: str = 'standard_2ome', overhang: str = 'dTdT', zfn_design_params: ZFNDesignParameters | None = None, zfn_annotation: GenomicAnnotationConfig | None = None, check_off_targets: bool = True, variant_ids: list[str] | None = None, variant_vcf_file: Path | None = None, variant_mode: str = 'avoid', variant_min_af: float = 0.01, variant_clinvar_filters: str = 'Pathogenic,Likely pathogenic', variant_assembly: str = 'GRCh38', log_file: str | None = None, write_json_summary: bool = True, num_threads: int | None = None, allow_transcriptome_with_input_fasta: bool = False, default_transcriptome_sources: Sequence[str] = ('ensembl_human_cdna', 'ensembl_mouse_cdna', 'ensembl_rat_cdna', 'ensembl_macaque_cdna'), keep_nextflow_work: bool = False, nextflow_docker_image: str | None = None, max_hits: int | None = None, max_off_targets: int | None = None) → dict[str, Any]

Run complete siRNA design workflow.

Parameters:

• gene_query – Gene name or ID to search for

• output_dir – Directory for output files

• input_fasta – Local path or remote URI to an input FASTA file

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

• design_mode – Design mode (sirna, mirna, or zfn)

• top_n_candidates – Cap on how many top-ranked candidates are reported (None = no cap, the default). Enumeration and screening always cover every candidate.

• genome_species – Species genomes for off-target analysis

• query_species – Organism the TARGET transcripts belong to. Defaults to the organism the gene-query database serves (human), which is also the species of the default transcriptome; set it when designing against an input FASTA from another organism.

• genome_indices_override – Comma-separated species:/index_prefix overrides for off-target analysis

• mirna_database – miRNA reference database identifier

• mirna_species – miRNA reference species identifiers

• transcriptome_fasta – Path or URL to transcriptome FASTA for off-target analysis

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

• transcriptome_selection – Pre-resolved transcriptome selection metadata

• gc_min – Minimum GC content percentage

• gc_max – Maximum GC content percentage

• sirna_length – siRNA length in nucleotides

• modification_pattern – Chemical modification pattern

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

• zfn_design_params – Optional ZFN design parameters for ZFN mode workflow

• zfn_annotation – Optional genomic annotation config for ZFN off-target classification

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

• variant_ids – List of variant identifiers (rsID, chr:pos:ref:alt, or HGVS) to target or avoid

• variant_vcf_file – Path to VCF file containing variants to target or avoid

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

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

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

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

• log_file – Path to centralized log file

• write_json_summary – Write logs/workflow_summary.json

• num_threads – Optional override for design parallelism

• allow_transcriptome_with_input_fasta – Opt in to resolving default_transcriptome_sources when input_fasta is supplied (default: False). Left False, an input-FASTA run is design-only unless transcriptome_fasta names a reference explicitly: supplying your own sequences should never trigger a multi-gigabyte reference download you did not ask for. Set True to screen an input-FASTA run against the bundled defaults.

• default_transcriptome_sources – Ordered list of transcriptome identifiers evaluated by default

• keep_nextflow_work – Keep Nextflow work directory symlink in output

• nextflow_docker_image – Override Docker image used by the embedded Nextflow pipeline

• max_hits – Override the pipeline’s per-candidate off-target hit cap (None keeps the pipeline’s exhaustive default; set a lower value, e.g. 10000, to speed up large gene-family searches)

• max_off_targets – Override the genuine off-target ceiling that gates PASS vs EXCESS_OFF_TARGETS (None keeps OffTargetFilterCriteria’s default of 15). Unlike max_hits this changes the verdict, not how many hits are recorded.

Returns:

Dictionary with complete workflow results

async sirnaforge.workflow.run_offtarget_only_workflow(input_candidates_fasta: str, output_dir: str, genome_species: list[str] | None = None, query_species: str | None = None, genome_indices_override: str | None = None, mirna_database: str = 'mirgenedb', mirna_species: Sequence[str] | None = None, transcriptome_fasta: str | None = None, transcriptome_filter: str | None = None, transcriptome_selection: ReferenceSelection | None = None, log_file: str | None = None, nextflow_docker_image: str | None = None) → dict[str, Any]

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 – Path to FASTA file with 21-nt siRNA guide sequences

• output_dir – Directory for output files

• genome_species – Species genomes for off-target analysis

• query_species – Organism the input guides were designed against (defaults to human)

• genome_indices_override – Comma-separated species:/index_prefix overrides

• mirna_database – miRNA reference database identifier

• mirna_species – miRNA reference species identifiers

• transcriptome_fasta – Path or URL to transcriptome FASTA for off-target analysis

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

• transcriptome_selection – Pre-resolved transcriptome selection metadata

• log_file – Path to centralized log file

• nextflow_docker_image – Override Docker image used by the embedded Nextflow pipeline

Returns:

Dictionary with off-target analysis results

Variant Workflow Integration

Workflow integration for variant targeting.

class sirnaforge.workflow_variant.VariantWorkflowConfig(variant_ids: list[str] | None = None, vcf_file: Path | None = None, variant_mode: VariantMode = VariantMode.AVOID, min_af: float = 0.01, clinvar_filter_levels: list[ClinVarSignificance] | None = None, assembly: str = 'GRCh38', cache_dir: Path | None = None)

Bases: object

Configuration for variant targeting in the workflow.

__init__(variant_ids: list[str] | None = None, vcf_file: Path | None = None, variant_mode: VariantMode = VariantMode.AVOID, min_af: float = 0.01, clinvar_filter_levels: list[ClinVarSignificance] | None = None, assembly: str = 'GRCh38', cache_dir: Path | None = None)

Initialize variant workflow configuration.

Parameters:

• variant_ids – List of variant identifiers (rsID, coordinate, HGVS)

• vcf_file – Path to VCF file with variants

• variant_mode – How to handle variants (avoid/target/both)

• min_af – Minimum allele frequency threshold

• clinvar_filter_levels – Allowed ClinVar significance levels

• assembly – Reference genome assembly (only GRCh38 supported)

• cache_dir – Cache directory for variant data

property has_variants: bool

Check if any variants are configured.

async sirnaforge.workflow_variant.resolve_workflow_variants(config: VariantWorkflowConfig, gene_name: str, output_dir: Path) → list[VariantRecord]

Resolve all variants for the workflow.

This is an optional workflow step that runs after gene/transcript selection and before siRNA candidate design.

Parameters:

• config – Variant workflow configuration

• gene_name – Gene name for logging/reporting

• output_dir – Output directory for reports

Returns:

List of resolved VariantRecords passing filters

sirnaforge.workflow_variant.normalize_variant_mode(mode: str | VariantMode) → VariantMode

Normalize variant mode to enum, handling string input.

Parameters:

mode – Variant mode as string or enum

Returns:

VariantMode enum

Raises:

ValueError – If mode string is invalid

sirnaforge.workflow_variant.parse_clinvar_filter_string(filter_string: str) → list[ClinVarSignificance]

Parse comma-separated ClinVar filter string to enum list.

Parameters:

filter_string – Comma-separated string of significance levels

Returns:

List of ClinVarSignificance enums

Raises:

ValueError – If any significance level is invalid

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(*, default_level: ValidationLevel = ValidationLevel.STRICT, stage_levels: dict[~sirnaforge.validation.config.ValidationStage, ~sirnaforge.validation.config.ValidationLevel]=<factory>, validate_sequences: bool = True, validate_ranges: bool = True, validate_consistency: bool = True, validate_biology: bool = True, max_validation_errors: Annotated[int, ~annotated_types.Ge(ge=1)] = 100, collect_all_errors: bool = False, batch_size: Annotated[int, ~annotated_types.Ge(ge=1)] = 1000, enable_caching: bool = True)

Bases: BaseModel

Configuration for validation system.

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: ValidationStage) → ValidationLevel

Get validation level for a specific stage.

is_enabled_for_stage(stage: ValidationStage) → bool

Check if validation is enabled for a stage.

should_fail_on_error(stage: ValidationStage) → bool

Check if validation errors should cause failures.

model_config = {}

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() → ValidationConfig

Configuration for development environment.

static production() → ValidationConfig

Configuration for production environment.

static testing() → ValidationConfig

Configuration for testing environment.

static performance() → ValidationConfig

Configuration optimized for performance.

Validation Middleware

Validation middleware for integrating validation into the siRNA design workflow.

class sirnaforge.validation.middleware.ValidationReport(stage: ValidationStage)

Bases: object

Comprehensive validation report for a workflow stage.

__init__(stage: ValidationStage)

Initialize validation report.

add_item_result(result: ValidationResult) → None

Add validation result for an individual item.

finalize() → None

Finalize the report and calculate summary statistics.

to_dict() → dict[str, Any]

Convert report to dictionary.

class sirnaforge.validation.middleware.ValidationMiddleware(config: ValidationConfig)

Bases: object

Middleware for integrating validation throughout the workflow.

__init__(config: ValidationConfig)

Initialize validation middleware.

validate_input_parameters(params: DesignParameters) → ValidationReport

Validate input design parameters.

validate_transcripts(transcripts: list[TranscriptInfo]) → ValidationReport

Validate transcript data after retrieval.

validate_design_results(design_result: DesignResult) → ValidationReport

Validate siRNA design results.

validate_dataframe_output(df: DataFrame, schema_type: str) → ValidationReport

Validate DataFrame output against pandera schemas.

validate_transcript_id_consistency(transcripts: list[TranscriptInfo], candidates: list[SiRNACandidate], orf_data: DataFrame | None = None) → ValidationReport

Validate consistency of transcript IDs across datasets.

save_validation_report(output_path: Path) → None

Save comprehensive validation report.

Validation Utilities

Validation utilities for data consistency and cross-validation.

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

Bases: object

Container for validation results.

__init__(is_valid: bool = True)

Initialize validation result container.

add_error(message: str) → None

Add a validation error.

add_warning(message: str) → None

Add a validation warning.

add_metadata(key: str, value: Any) → None

Add metadata to the result.

merge(other: ValidationResult) → None

Merge another validation result into this one.

summary() → dict[str, Any]

Get a summary of validation results.

class sirnaforge.validation.utils.ValidationUtils

Bases: object

Utility functions for data validation.

static validate_nucleotide_sequence(sequence: str, allow_ambiguous: bool = True) → ValidationResult

Validate nucleotide sequence composition.

static validate_sirna_length(sequence: str) → ValidationResult

Validate siRNA sequence length.

static validate_parameter_consistency(params: DesignParameters) → ValidationResult

Validate design parameter consistency.

static validate_candidate_consistency(candidate: SiRNACandidate) → ValidationResult

Validate siRNA candidate internal consistency.

static validate_dataframe_schema(df: DataFrame, schema_type: str) → ValidationResult

Validate DataFrame against appropriate pandera schema.

static validate_transcript_ids_consistency(candidate_ids: set[str], orf_ids: set[str], transcript_ids: set[str]) → ValidationResult

Validate consistency of transcript IDs across datasets.

static validate_biological_constraints(candidate: SiRNACandidate) → ValidationResult

Validate bioinformatics-specific constraints.

static cross_validate_pydantic_pandera() → ValidationResult

Cross-validate Pydantic model constraints with Pandera schema constraints.

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: str | None = None, log_file: str | None = None) → None

Configure root logger: console + optional rotating file handler.

Parameters:

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

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

sirnaforge.utils.logging_utils.get_logger(name: str, level: str | None = None) → Logger

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.

Cache Utilities

Shared cache helpers for siRNAforge.

Centralizes cache path resolution so every subsystem honors the same SIRNAFORGE_CACHE_DIR / XDG cache layout, and the producer-version stamping that lets us discard artifacts a known-bad writer left on disk.

sirnaforge.utils.cache_utils.resolve_cache_subdir(subdir: str, *, override: str | PathLike[str] | None = None) → Path

Resolve a writable cache directory for the requested subdir.

The lookup order matches ReferenceManager: explicit override, env vars, XDG, $HOME/.cache, workspace-local fallback, then temp dir.

sirnaforge.utils.cache_utils.stable_cache_key(payload: Mapping[str, Any]) → str

Create a deterministic sha256 digest for arbitrary JSON-serializable data.

sirnaforge.utils.cache_utils.LEGACY_PRODUCER_VERSION = '1.0'

Version attributed to artifacts written before producer stamping existed.

CacheMetadata.version has always defaulted to this string, so pre-existing metadata reads back as “produced by the pre-versioning writer” without needing a migration step.

sirnaforge.utils.cache_utils.PRODUCER_VERSION_FIELD = 'producer_version'

Key under which an artifact stamp records the version that produced it.

Manager-backed caches keep the same value in CacheMetadata.version; sidecar stamps have no such document, so they carry this field. Naming it once keeps the writer (write_artifact_stamp) and the reader (is_artifact_stamp_current) from drifting apart. Only artifact classes that are actually validated somewhere are registered below - a constant nobody checks would read as a guarantee we do not give.

sirnaforge.utils.cache_utils.current_producer_version(artifact_class: str) → str

Return the producer version to record for artifact_class.

Unregistered classes report the legacy version: registering a class is what opts it in, so a new cache namespace can never invalidate itself by accident.

sirnaforge.utils.cache_utils.normalize_producer_version(recorded: object) → str

Coerce a recorded producer version into a comparable string.

Missing values (None, NaN from a parquet column that did not exist yet, an empty string) all mean the same thing: the artifact predates stamping.

sirnaforge.utils.cache_utils.is_producer_version_current(artifact_class: str, recorded: object) → bool

Whether an artifact’s recorded producer version is the current one.

sirnaforge.utils.cache_utils.log_discarded_artifact(artifact_class: str, artifact: Path | str, reason: str) → None

Warn that a stale cache artifact is being dropped.

Deliberately WARNING, not DEBUG: this is the one line that explains why a run suddenly spends time regenerating or re-downloading a reference.

sirnaforge.utils.cache_utils.STAMP_FULL_DIGEST_MAX_BYTES = 67108864

Largest output we are willing to read end-to-end on every cache hit.

A combined miRNA FASTA is a few MB, so it is digested whole for a few milliseconds. A BWA-MEM2 index or a transcriptome is gigabytes: reading it in full on every hit would cost more than the cache saves, so those fall back to DIGEST_MODE_SAMPLED.

sirnaforge.utils.cache_utils.STAMP_SAMPLE_BYTES = 4194304

Head/tail window digested for outputs above the full-digest budget.

sirnaforge.utils.cache_utils.artifact_stamp_path(artifact: Path) → Path

Return the sidecar stamp path for a derived artifact or index prefix.

sirnaforge.utils.cache_utils.fingerprint_outputs(outputs: Sequence[Path], *, full_digest_max_bytes: int | None = None) → dict[str, dict[str, Any]]

Fingerprint the bytes an artifact is actually made of.

Keyed by basename so the fingerprint survives a re-homed cache directory, and so a multi-file artifact (a BWA index prefix has four members) records one entry per member and therefore proves its own completeness on read.

Every member records its exact size; the digest is full below STAMP_FULL_DIGEST_MAX_BYTES and sampled above it. Size alone already catches every truncation, which is the failure mode that actually happens.

sirnaforge.utils.cache_utils.fingerprint_inputs(inputs: Sequence[Path]) → dict[str, str]

MD5 each input file, keyed by position and basename.

Position keeps duplicate basenames distinct; basenames rather than full paths keep the fingerprint stable when a cache directory is re-homed (containers mount it elsewhere). Callers that already hold a digest for a large file should pass it directly rather than paying for a second full read.

sirnaforge.utils.cache_utils.write_artifact_stamp(artifact_class: str, artifact: Path, *, inputs: Mapping[str, str] | None = None, outputs: Sequence[Path] | None = None, extra: Mapping[str, Any] | None = None) → Path

Record the producer version, input fingerprint and output bytes of an artifact.

outputs lists the files the artifact consists of, and defaults to the artifact itself. Multi-file artifacts must pass their members explicitly: a BWA index prefix is not a file, and the four files behind it are what has to be verified.

sirnaforge.utils.cache_utils.read_artifact_stamp(artifact: Path) → dict[str, Any] | None

Read a derived artifact’s stamp, or None when it is absent/unreadable.

sirnaforge.utils.cache_utils.discard_artifact_stamp(artifact: Path) → None

Remove a derived artifact’s stamp so it cannot vouch for a rebuilt file.

sirnaforge.utils.cache_utils.is_artifact_stamp_current(artifact_class: str, artifact: Path, *, inputs: Mapping[str, str] | None = None, max_age_days: int | None = None) → bool

Whether a derived artifact may be reused.

Answers “was it produced by the current code, from exactly these inputs, recently enough - and are its own bytes still the bytes we stamped”. The last clause is what makes a stamp unable to vouch for an artifact that was truncated or corrupted after it was written, so callers may reuse on a True answer alone.

CLI Input Helpers

Shared CLI input parsing and validation helpers.

These helpers are intentionally small and side-effect free so that Typer/Rich commands can stay focused on orchestration and UX.

sirnaforge.utils.cli_inputs.parse_csv(value: str) → list[str]

Split a comma-separated string into normalized non-empty tokens.

sirnaforge.utils.cli_inputs.parse_required_csv(value: str, *, error_message: str) → list[str]

Parse a required CSV argument, raising ValueError when empty.

sirnaforge.utils.cli_inputs.parse_optional_csv(value: str | None, *, error_message: str) → list[str] | None

Parse an optional CSV argument.

Returns None when value is None, otherwise returns a list of tokens. Raises ValueError when the provided string contains no usable tokens.

sirnaforge.utils.cli_inputs.extract_override_species_from_offtarget_indices(offtarget_indices: str | None) → list[str] | None

Extract unique species tokens from an offtarget indices override string.

Validates the expected species:/index_prefix format for each entry. Returns None when no override is provided.

class sirnaforge.utils.cli_inputs.SpeciesResolution(source_normalized: str, canonical_species: list[str], genome_species: list[str], mirna_species: list[str])

Bases: object

Resolved species identifiers for genome, canonical, and miRNA scopes.

source_normalized: str

canonical_species: list[str]

genome_species: list[str]

mirna_species: list[str]

__init__(source_normalized: str, canonical_species: list[str], genome_species: list[str], mirna_species: list[str]) → None

sirnaforge.utils.cli_inputs.resolve_species_inputs(*, species: str, mirna_db: str, mirna_species: str | None) → SpeciesResolution

Validate species + miRNA arguments and return resolved identifiers.

Raises ValueError with a user-facing message when inputs are invalid.

Control Candidates

Helpers for injecting control candidates into design results.

sirnaforge.utils.control_candidates.inject_dirty_controls(design_result: DesignResult, count: int = 2) → list[SiRNACandidate]

Append clearly labelled “dirty” control candidates to a design result.

“Dirty” controls are not synthetic sequences. We intentionally reuse the lowest-scoring rejected candidates from the same run so every control is a plausible design that simply failed QC (GC range, off-target score, etc.). Carrying these borderline guides forward guarantees that the downstream off-target pipeline receives at least a couple of sequences that should light up in the reports, making it easier for users to confirm the workflow is wired correctly.

The clones retain their original guide/passenger sequences and metadata but are marked with DIRTY_CONTROL. They are appended to both candidates and top_candidates so that subsequent stages (CSV export, Nextflow off-target analysis, etc.) see the controls exactly where they expect normal guides. IDs are suffixed with __DIRTY_CONTROL_<n> for clarity.

Parameters:

• design_result – Aggregated design result to augment in-place.

• count – Number of controls to add (default: 2).

Returns:

List of control candidates that were appended. Returns an empty list if the design result already contains controls or there are no candidates to copy.

FASTA Utilities

Shared FASTA parsing helpers used across siRNAforge modules.

sirnaforge.utils.fasta.load_fasta_sequences(fasta_path: Path, *, uppercase: bool = True) → dict[str, str]

Load FASTA records into a {record_id: sequence} mapping.

sirnaforge.utils.fasta.load_fasta_contig_lengths(fasta_path: Path) → dict[str, int]

Load FASTA record lengths into a {record_id: length} mapping.

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: str) → list[ChemicalModification]

Apply standard alternating 2’-O-methyl pattern.

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

Parameters:

sequence – RNA sequence to modify

Returns:

List containing one ChemicalModification with alternating positions

sirnaforge.utils.modification_patterns.apply_minimal_terminal_pattern(sequence: str) → list[ChemicalModification]

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 – RNA sequence to modify

Returns:

List containing one ChemicalModification with terminal positions

sirnaforge.utils.modification_patterns.apply_maximal_stability_pattern(sequence: str) → list[ChemicalModification]

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 – RNA sequence to modify

Returns:

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

sirnaforge.utils.modification_patterns.get_modification_pattern(pattern_name: str, sequence: str) → list[ChemicalModification]

Get modification pattern by name.

Parameters:

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

• sequence – RNA sequence to apply pattern to

Returns:

List of ChemicalModification objects

Raises:

ValueError – If pattern_name is not recognized

sirnaforge.utils.modification_patterns.apply_modifications_to_candidate(candidate: SiRNACandidate, pattern_name: str = 'standard_2ome', overhang: str = 'dTdT', target_gene: str | None = None) → SiRNACandidate

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 to annotate

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

• overhang – Overhang sequence (default: dTdT)

• target_gene – Optional target gene name for metadata

Returns:

Updated SiRNACandidate with modification metadata

sirnaforge.utils.modification_patterns.get_modification_summary(candidate: SiRNACandidate) → dict[str, str]

Get a summary of modifications for a candidate.

Parameters:

candidate – SiRNACandidate with modification metadata

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: str, local_path: Path, source_type: str, downloaded: bool, size_bytes: int, sha256: str | None = None)

Bases: object

Normalized representation of a workflow input resource.

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: str, local_path: Path, source_type: str, downloaded: bool, size_bytes: int, sha256: str | None = None) → None

sirnaforge.utils.resource_resolver.resolve_input_source(input_location: str, destination_root: Path, *, timeout: float = 30.0) → InputSource

Resolve a workflow input location into a local path.

Parameters:

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

• destination_root – Directory where downloaded inputs should be stored.

• timeout – Timeout for remote downloads in seconds.

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.

Species Utilities

Shared helpers for reasoning about species labels.

These utilities centralize the normalization and categorization rules we use when aggregating off-target hits across genome/transcriptome and miRNA pipelines. The helpers intentionally accept very loose inputs (common names, scientific labels, assembly identifiers) so downstream callers can rely on a single implementation when deciding whether a hit should be treated as human-specific.

sirnaforge.utils.species.bucket_species(label: str | None) → str

Map an arbitrary label to "human" or "other".

sirnaforge.utils.species.human_vs_other_totals(counts: Mapping[str, int]) → tuple[int, int]

Collapse per-species hit counts into (human, other) totals.

sirnaforge.utils.species.is_human_species(label: str | None) → bool

Return True when the provided label should be bucketed as human.

sirnaforge.utils.species.normalize_species_label(label: str | None) → str

Normalize a species label to a lowercase slug.

The transformation removes punctuation, collapses whitespace, and replaces runs of non-alphanumeric characters with a single underscore so strings like "Homo sapiens (GRCh38)" become "homo_sapiens_grch38".

Typed Decorators

Typed wrappers for third-party decorators used across models.

These wrappers preserve runtime behavior while helping static analyzers (mypy/Pylance) keep function signatures precise when decorator stubs are insufficiently typed.

sirnaforge.utils.typed_decorators.field_validator_typed(*field_names: str, **kwargs: Any) → Callable[[F], F]

Typed wrapper around pydantic field_validator.

sirnaforge.utils.typed_decorators.field_serializer_typed(*field_names: str, **kwargs: Any) → Callable[[F], F]

Typed wrapper around pydantic field_serializer.

sirnaforge.utils.typed_decorators.check_types_typed(func: F) → F

Typed wrapper around pandera check_types.

sirnaforge.utils.typed_decorators.model_validator_typed(*, mode: Literal['wrap', 'before', 'after'] = 'after') → Callable[[F], F]

Typed wrapper around pydantic model_validator.

sirnaforge.utils.typed_decorators.command_decorator_typed(decorator_factory: Callable[[...], Any]) → Callable[[...], Callable[[F], F]]

Typed wrapper for command decorator factories (e.g. Typer.command).

Unified Cache

Unified cache manager for all sirnaforge reference databases.

Provides a single interface to manage miRNA databases, transcriptomes, and indices using composition and protocol-based design.

class sirnaforge.utils.unified_cache.CacheStats

Bases: TypedDict

Statistics for a cache component.

cache_directory: str

total_files: int

total_size_mb: float

cache_ttl_days: int

class sirnaforge.utils.unified_cache.ClearResult

Bases: TypedDict

Result of a cache clear operation.

files_deleted: int

size_freed_mb: float

status: str

class sirnaforge.utils.unified_cache.CacheComponent(*args, **kwargs)

Bases: Protocol

Protocol for cache components.

cache_info() → CacheStats

Get cache statistics.

clear_cache(confirm: bool = False) → ClearResult

Clear the cache.

__init__(*args, **kwargs)

class sirnaforge.utils.unified_cache.UnifiedCacheManager

Bases: object

Unified manager for all sirnaforge caches.

Uses composition to combine miRNA and transcriptome caches into a single, easy-to-use interface.

__init__() → None

Initialize with all cache components.

get_info(include_mirna: bool = True, include_transcriptome: bool = True) → dict[str, CacheStats]

Get cache info for selected components.

Parameters:

• include_mirna – Include miRNA cache stats

• include_transcriptome – Include transcriptome cache stats

Returns:

Dictionary mapping component name to cache stats

clear(clear_mirna: bool = False, clear_transcriptome: bool = False, dry_run: bool = False) → dict[str, ClearResult]

Clear selected cache components.

Parameters:

• clear_mirna – Clear miRNA databases

• clear_transcriptome – Clear transcriptomes and indices

• dry_run – Show what would be deleted without deleting

Returns:

Dictionary mapping component name to clear results

get_total_stats() → dict[str, float | int]

Get combined statistics across all caches.

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: str) → list[ChemicalModification]

Parse ChemMods field from FASTA header.

Parameters:

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

Returns:

List of ChemicalModification objects

sirnaforge.modifications.parse_provenance(prov_str: str, url: str | None = None) → Provenance | None

Parse Provenance field from FASTA header.

Parameters:

• prov_str – String like “Patent:US10060921B2”

• url – Optional URL string

Returns:

Provenance object or None

sirnaforge.modifications.parse_header(record: SeqRecord) → dict[str, Any]

Parse FASTA header to extract metadata.

Parameters:

record – BioPython SeqRecord from FASTA file

Returns:

Dictionary with parsed metadata fields

sirnaforge.modifications.load_metadata(json_path: str | Path) → dict[str, StrandMetadata]

Load and validate metadata from JSON sidecar file using Pydantic.

Parameters:

json_path – Path to JSON file containing metadata

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: str | Path, metadata_path: str | Path, output_path: str | Path) → int

Merge metadata from JSON into FASTA headers.

Uses Pydantic for automatic validation of metadata.

Parameters:

• fasta_path – Input FASTA file

• metadata_path – JSON file with metadata

• output_path – Output FASTA file with updated headers

Returns:

Number of sequences with metadata applied

Raises:

ValidationError – If metadata doesn’t match StrandMetadata schema

sirnaforge.modifications.save_metadata_json(metadata_dict: dict[str, StrandMetadata], output_path: str | Path) → None

Save strand metadata to JSON file using Pydantic serialization.

Parameters:

• metadata_dict – Dictionary mapping strand IDs to StrandMetadata objects

• output_path – Path to output JSON file