mutcleaner.core#
Core functionality for sequence manipulation
- class mutcleaner.core.AminoAcidMutationSet(mutations, name=None, metadata=None)[source]#
Bases:
MutationSet[AminoAcidMutation]Represents a set of amino acid mutations
- Attributes:
mutation_subtypeGet the specific mutation subtype (e.g., ‘amino_acid’, ‘codon_dna’, ‘codon_rna’, ‘codon_both’)
Methods
add_mutation(mutation)Add a mutation to this set
Count mutations by effect type
filter_by_category(category)Filter mutations by category
from_string(string[, sep, is_zero_based, ...])Create a mutation set from a string
Get all missense mutations
get_mutation_at(position)Get mutation at specified position
get_mutation_categories()Get mutation category statistics
get_mutation_count()Get number of mutations
Get all nonsense mutations
get_positions()Get all mutation positions
get_positions_set()Get all mutation positions as a set
get_sorted_by_position()Get mutations sorted by position without modifying the original list
Get all synonymous mutations
has_mutation_at(position)Check if there is a mutation at specified position
Check if any mutations introduce stop codons
is_multiple_mutations()Check if this contains multiple mutations
is_single_mutation()Check if this is a single mutation
remove_mutation(position)Remove mutation at specified position, return True if removed
sort_by_position()Sort mutations by position in ascending order
validate_all()Validate all mutations
- get_missense_mutations()[source]#
Get all missense mutations
- Return type:
List[AminoAcidMutation]
- get_nonsense_mutations()[source]#
Get all nonsense mutations
- Return type:
List[AminoAcidMutation]
- get_synonymous_mutations()[source]#
Get all synonymous mutations
- Return type:
List[AminoAcidMutation]
- class mutcleaner.core.CodonMutationSet(mutations, name=None, metadata=None)[source]#
Bases:
MutationSet[CodonMutation]Represents a set of codon mutations
- Attributes:
mutation_subtypeGet the specific mutation subtype (e.g., ‘amino_acid’, ‘codon_dna’, ‘codon_rna’, ‘codon_both’)
seq_typeGet the sequence type (DNA, RNA, or Both) of the codon mutations
Methods
add_mutation(mutation)Add a mutation to this set
filter_by_category(category)Filter mutations by category
from_string(string[, sep, is_zero_based, ...])Create a mutation set from a string
get_mutation_at(position)Get mutation at specified position
get_mutation_categories()Get mutation category statistics
get_mutation_count()Get number of mutations
get_positions()Get all mutation positions
get_positions_set()Get all mutation positions as a set
get_sorted_by_position()Get mutations sorted by position without modifying the original list
has_mutation_at(position)Check if there is a mutation at specified position
is_multiple_mutations()Check if this contains multiple mutations
is_single_mutation()Check if this is a single mutation
remove_mutation(position)Remove mutation at specified position, return True if removed
sort_by_position()Sort mutations by position in ascending order
to_amino_acid_mutation_set([codon_table])Convert all codon mutations to amino acid mutations
validate_all()Validate all mutations
- property seq_type: Literal['DNA', 'RNA', 'Both']#
Get the sequence type (DNA, RNA, or Both) of the codon mutations
- class mutcleaner.core.CodonTable(name, codon_map, start_codons=None, stop_codons=None)[source]#
Bases:
objectcodon table used to translate codons to amino acids
Methods
get_standard_table([seq_type])get standard codon table (NCBI standard)
get_table_by_name(name[, seq_type])get codon table by name
is_start_codon(codon)check if codon is a start codon
is_stop_codon(codon)check if codon is a stop codon
translate_codon(codon)translate single codon to corresponding amino acid
- classmethod get_standard_table(seq_type='DNA')[source]#
get standard codon table (NCBI standard)
- Return type:
- class mutcleaner.core.DNAAlphabet(include_ambiguous=False)[source]#
Bases:
BaseAlphabetDNA alphabet (A, T, C, G)
Methods
get_invalid_chars(sequence)Get set of invalid characters in sequence
is_valid_char(char)Check if character is valid in this alphabet
is_valid_sequence(sequence)Check if entire sequence is valid
validate_sequence(sequence)Validate sequence and raise error if invalid
- class mutcleaner.core.DNASequence(sequence, alphabet=None, name=None, metadata=None)[source]#
Bases:
BaseSequenceDNA sequence with nucleotide validation
Methods
apply_mutation(mutation)Apply a mutation or set of mutations to the sequence and return a new sequence.
Subclasses may override this method to provide a default alphabet.
get_subsequence(start[, end])get subsequence (0-indexed, inclusive)
infer_mutation(other)Infer a mutation that leads to a specific sequence
Get reverse complement of DNA sequence
Transcribe DNA sequence into RNA sequence
translate([codon_table, start_at_first_met, ...])Translate DNA sequence into amino acid sequence using this codon table.
- classmethod default_alphabet()[source]#
Subclasses may override this method to provide a default alphabet.
By default, it returns None, indicating no default is provided and callers must pass alphabet explicitly.
- Return type:
Optional[BaseAlphabet]
- translate(codon_table=None, start_at_first_met=False, stop_at_stop_codon=False, require_mod3=True, start=None, end=None)[source]#
Translate DNA sequence into amino acid sequence using this codon table.
- Parameters:
codon_table (
Optional[CodonTable]) – Codon table to use for translation. If None, uses standard genetic code.start_at_first_met (
bool) – Start translation at the first start codon if found.stop_at_stop_codon (
bool) – Stop translation when a stop codon is encountered.require_mod3 (
bool) – Whether the sequence must be a multiple of 3 in length.start (
Optional[int]) – Custom 0-based start position. Overrides start_at_first_met.end (
Optional[int]) – Custom 0-based end position. Overrides stop_at_stop_codon.
- Return type:
- Returns:
Translated amino acid sequence.
- class mutcleaner.core.MutationDataset(name=None)[source]#
Bases:
objectDataset container for cleaned mutation data with multiple reference sequences.
All mutation sets must be linked to a reference sequence when added to the dataset. This ensures data integrity and enables proper validation and analysis.
Methods
add_mutation_set(mutation_set, reference_id)Add a mutation set to the dataset, linking to a reference sequence
add_mutation_sets(mutation_sets, reference_ids)Add multiple mutation sets to the dataset
add_reference_sequence(sequence_id, sequence)Add a reference sequence with a unique identifier
Convert all codon mutation sets to amino acid mutation sets
filter_by_effect_type(effect_type)Filter dataset by amino acid mutation effect type (synonymous, missense, nonsense)
filter_by_mutation_type(mutation_type)Filter dataset by mutation type
filter_by_reference(reference_id)Filter dataset to only include mutation sets from a specific reference sequence
from_dataframe(df, reference_sequences[, ...])Create a MutationDataset from a DataFrame containing mutation data.
get_mutation_set_label(mutation_set_index)Get the label for a specific mutation set
get_mutation_set_reference(mutation_set_index)Get the reference sequence ID for a specific mutation set
get_position_coverage([reference_id])Get statistics about position coverage across reference sequences
get_reference_sequence(sequence_id)Get a reference sequence by ID
Get basic statistics about the dataset
Get list of all reference sequence IDs
load(filepath[, load_type])Load a dataset from files.
load_by_reference(base_dir[, dataset_name, ...])Load a dataset from mutcleaner reference-based format.
remove_mutation_set(mutation_set_index)Remove a mutation set from the dataset
remove_reference_sequence(sequence_id)Remove a reference sequence
save(filepath[, save_type])Save the dataset to files.
save_by_reference(base_dir)Save dataset by reference_id, creating separate folders for each reference.
set_mutation_set_label(mutation_set_index, label)Set the label for a specific mutation set
Set the reference sequence for a specific mutation set
Convert dataset to pandas DataFrame
Validate mutations against their reference sequences
- add_mutation_set(mutation_set, reference_id, label=None)[source]#
Add a mutation set to the dataset, linking to a reference sequence
- add_mutation_sets(mutation_sets, reference_ids, labels=None)[source]#
Add multiple mutation sets to the dataset
- add_reference_sequence(sequence_id, sequence)[source]#
Add a reference sequence with a unique identifier
- convert_codon_to_amino_acid_sets(convert_labels=False)[source]#
Convert all codon mutation sets to amino acid mutation sets
- Parameters:
convert_labels (
bool) – Whether to save the labels with the mutation sets (default: False)- Return type:
- filter_by_effect_type(effect_type)[source]#
Filter dataset by amino acid mutation effect type (synonymous, missense, nonsense)
- Return type:
- filter_by_reference(reference_id)[source]#
Filter dataset to only include mutation sets from a specific reference sequence
- Return type:
- classmethod from_dataframe(df, reference_sequences, name=None, specific_mutation_type=None)[source]#
Create a MutationDataset from a DataFrame containing mutation data.
This method reconstructs a MutationDataset from a flattened DataFrame representation, typically used for loading saved mutation datasets from files. The DataFrame should contain mutation information with each row representing a single mutation within mutation sets.
- Parameters:
df – DataFrame containing mutation data with the following required columns: - ‘mutation_set_id’: Identifier for grouping mutations into sets - ‘reference_id’: Identifier for the reference sequence - ‘mutation_string’: String representation of the mutation - ‘position’: Position of the mutation in the sequence - ‘mutation_type’: Type of mutation (‘amino_acid’, ‘codon_dna’, ‘codon_rna’)
- Return type:
- Returns:
A new MutationDataset instance populated with the mutation sets and reference sequences from the DataFrame.
- Raises:
ValueError – If the DataFrame is empty, missing required columns, or references sequences not provided in reference_sequences dict.
Notes
Mutations are grouped by ‘mutation_set_id’ to reconstruct mutation sets
The method automatically determines the appropriate mutation set type
(AminoAcidMutationSet, CodonMutationSet, or generic MutationSet) based on the mutation types within each set - Metadata is extracted from columns with ‘set_’ and ‘mutation_’ prefixes - Only reference sequences that are actually used in the DataFrame are added to the dataset
Examples
>>> import pandas as pd >>> from sequences import ProteinSequence >>> >>> # Create sample DataFrame >>> df = pd.DataFrame({ ... 'mutation_set_id': ['set1', 'set1', 'set2'], ... 'reference_id': ['prot1', 'prot1', 'prot2'], ... 'mutation_string': ['A1V', 'L2P', 'G5R'], ... 'position': [1, 2, 5], ... 'mutation_type': ['amino_acid', 'amino_acid', 'amino_acid'], ... 'wild_amino_acid': ['A', 'L', 'G'], ... 'mutant_amino_acid': ['V', 'P', 'R'], ... 'mutation_set_name': ['variant1', 'variant1', 'variant2'], ... 'label': ['pathogenic', 'pathogenic', 'benign'] ... }) >>> >>> # Define reference sequences >>> ref_seqs = { ... 'prot1': ProteinSequence('ALDEFG', name='protein1'), ... 'prot2': ProteinSequence('MKGLRK', name='protein2') ... } >>> >>> # Create MutationDataset >>> dataset = MutationDataset.from_dataframe(df, ref_seqs, name="my_dataset") >>> print(len(dataset.mutation_sets)) 2
- get_mutation_set_label(mutation_set_index)[source]#
Get the label for a specific mutation set
- Return type:
Any
- get_mutation_set_reference(mutation_set_index)[source]#
Get the reference sequence ID for a specific mutation set
- Return type:
str
- get_position_coverage(reference_id=None)[source]#
Get statistics about position coverage across reference sequences
- Return type:
Dict[str,Any]
- classmethod load(filepath, load_type=None)[source]#
Load a dataset from files.
- Parameters:
filepath (
str) – Base filepath (with or without extension)load_type (
Optional[str]) – Type of load format (“mutcleaner”, “dataframe” or “pickle”). If None, auto-detect from file extension.
- Return type:
- Returns:
Examples
>>> # Auto-detect from extension >>> dataset = MutationDataset.load("my_study.csv") >>> dataset = MutationDataset.load("my_study.pkl")
>>> # Explicit type >>> dataset = MutationDataset.load("my_study", "dataframe")
- classmethod load_by_reference(base_dir, dataset_name=None, is_zero_based=True)[source]#
Load a dataset from mutcleaner reference-based format.
Expected directory structure:
base_dir/ ├── reference_id_1/ │ ├── data.csv │ ├── wt.fasta │ └── metadata.json ├── reference_id_2/ │ ├── data.csv │ ├── wt.fasta │ └── metadata.json └── ...
- Parameters:
base_dir (
Union[str,Path]) – Base directory containing reference foldersdataset_name (
Optional[str]) – Optional name for the loaded datasetis_zero_based (
bool) – Whether origin mutation positions are zero-based
- Return type:
- Returns:
- save(filepath, save_type='mutcleaner')[source]#
Save the dataset to files.
- Parameters:
filepath (
str) – Base filepath (without extension)save_type (
Optional[Literal['mutcleaner','pickle','dataframe']]) – Type of save format (“mutcleaner”, “dataframe” or “pickle”)save_type="dataframe": (For) –
Saves mutations as {filepath}.csv - Saves reference sequences as {filepath}_refs.pkl - Saves metadata as {filepath}_meta.json
save_type="pickle": (For) –
Saves entire dataset as {filepath}.pkl
Examples
>>> dataset.save("my_study", "dataframe") >>> # Creates: my_study.csv, my_study_refs.pkl, my_study_meta.json
- save_by_reference(base_dir)[source]#
Save dataset by reference_id, creating separate folders for each reference.
For each reference_id, creates:
{base_dir}/{reference_id}/data.csv: mutation data with columns [mutation_name, mutated_sequence, label]
{base_dir}/{reference_id}/wt.fasta: wild-type reference sequence
{base_dir}/{reference_id}/metadata.json: statistics and metadata for this reference
- Parameters:
base_dir (
Union[str,Path]) – Base directory to create reference folders in- Return type:
None
- set_mutation_set_label(mutation_set_index, label)[source]#
Set the label for a specific mutation set
- class mutcleaner.core.Pipeline(data=None, name=None, logging_level='INFO')[source]#
Bases:
objectPipeline for processing data with pandas-style method chaining
- Attributes:
artifactsAlways return the artifacts dictionary.
dataAlways return the actual data, never PipelineOutput.
has_pending_stepsCheck if there are delayed steps waiting to be executed
structured_dataReturn PipelineOutput object with both data and artifacts.
Methods
add_delayed_step(func[, index])Add a delayed step before a specific position in the delayed execution queue.
apply(func, *args, **kwargs)Apply function and return new Pipeline (functional style)
assign(**kwargs)Add attributes or computed values to data
copy()Create a deep copy of this pipeline
delayed_then(func, *args, **kwargs)Add a function to the delayed execution queue without running it immediately
execute([steps])Execute delayed steps.
filter(condition)Filter data based on condition
Get all stored artifacts
get_artifact(name)Get a specific artifact by name
get_data()Get current data (same as .data property).
Get information about delayed steps
Get summary of pipeline execution
get_step_result(step_index)Get result from a specific step by index or name
load(filepath[, format, name])Load data from file and create new pipeline
load_structured_data(filepath[, format, name])Load structured data from file and create new pipeline
peek([func, prefix])Inspect data without modifying it (for debugging)
remove_delayed_step(index_or_name)Remove a delayed step at the specified index.
save(filepath[, format])Save current data to file
save_artifacts(filepath[, format])Save all artifacts to file
save_structured_data(filepath[, format])Save structured data (data + artifacts) to file
store(name[, extractor])Store current data or extracted value as artifact
then(func, *args, **kwargs)Apply a function to the current data (pandas.pipe style)
transform(transformer, *args, **kwargs)Alias of then, used to define format transformations.
validate(validator[, error_msg])Validate data and raise error if invalid
Generate a text visualization of the pipeline
- add_delayed_step(func, index=None, *args, **kwargs)[source]#
Add a delayed step before a specific position in the delayed execution queue.
Performs a similar action to the list.insert() method.
- Parameters:
func (
Callable) – Function to add as delayed stepindex (
Optional[int]) – Position to insert the step. If None, appends to the end. Supports negative indexing.**kwargs (*args,) –
Arguments to pass to the function
- Return type:
- Returns:
Self for method chaining
Examples
>>> # Add step at the beginning >>> pipeline.add_delayed_step(func1, 0)
>>> # Add step at the end (same as delayed_then) >>> pipeline.add_delayed_step(func2)
>>> # Insert step at position 2 >>> pipeline.add_delayed_step(func3, 2)
>>> # Insert step before the last one >>> pipeline.add_delayed_step(func4, -1)
- apply(func, *args, **kwargs)[source]#
Apply function and return new Pipeline (functional style)
- Return type:
- property artifacts: Dict[str, Any]#
Always return the artifacts dictionary.
This provides direct access to all stored artifacts from pipeline steps.
- property data: Any#
Always return the actual data, never PipelineOutput.
This ensures consistent user experience - pipeline.data can always be used with methods like .copy(), .append(), etc.
- delayed_then(func, *args, **kwargs)[source]#
Add a function to the delayed execution queue without running it immediately
- Return type:
- execute(steps=None)[source]#
Execute delayed steps.
- Parameters:
steps (
Union[int,List[int],None]) – Which delayed steps to execute: - None: execute all delayed steps - int: execute the first N delayed steps - List[int]: execute specific delayed steps by index- Return type:
- Returns:
Self for method chaining
- get_data()[source]#
Get current data (same as .data property).
Kept for backward compatibility.
- Return type:
Any
- get_delayed_steps_info()[source]#
Get information about delayed steps
- Return type:
List[Dict[str,Any]]
- get_step_result(step_index)[source]#
Get result from a specific step by index or name
- Return type:
Any
- property has_pending_steps: bool#
Check if there are delayed steps waiting to be executed
- classmethod load(filepath, format='pickle', name=None)[source]#
Load data from file and create new pipeline
- Return type:
- classmethod load_structured_data(filepath, format='pickle', name=None)[source]#
Load structured data from file and create new pipeline
- Return type:
- remove_delayed_step(index_or_name)[source]#
Remove a delayed step at the specified index.
- Parameters:
index (int) – Index of the delayed step to remove
- Return type:
- Returns:
Self for method chaining
- Raises:
ValueError – If no delayed step is found with the specified index or name
- save_structured_data(filepath, format='pickle')[source]#
Save structured data (data + artifacts) to file
- Return type:
- property structured_data: PipelineOutput#
Return PipelineOutput object with both data and artifacts.
Use this when you need the complete pipeline state for serialization, passing to other systems, or when working with structured data flows.
- then(func, *args, **kwargs)[source]#
Apply a function to the current data (pandas.pipe style)
- Return type:
- transform(transformer, *args, **kwargs)[source]#
Alias of then, used to define format transformations.
- Return type:
- class mutcleaner.core.ProteinAlphabet(include_stop=True, include_ambiguous=False)[source]#
Bases:
BaseAlphabetProtein alphabet (20 standard amino acids + stop codon)
Methods
get_invalid_chars(sequence)Get set of invalid characters in sequence
get_one_letter_code(three_letter[, strict])Convert three-letter to one-letter amino acid code
get_three_letter_code(one_letter[, strict])Convert one-letter to three-letter amino acid code
is_valid_char(char)Check if character is valid in this alphabet
is_valid_sequence(sequence)Check if entire sequence is valid
validate_sequence(sequence)Validate sequence and raise error if invalid
- class mutcleaner.core.ProteinSequence(sequence, alphabet=None, name=None, metadata=None)[source]#
Bases:
BaseSequenceProtein sequence with amino acid validation
Methods
apply_mutation(mutation)Apply a mutation or set of mutations to the sequence and return a new sequence.
Subclasses may override this method to provide a default alphabet.
find_motif(motif)Find all positions where motif occurs (0-indexed)
get_residue(position)Get amino acid at specific position (0-indexed)
get_subsequence(start[, end])get subsequence (0-indexed, inclusive)
infer_mutation(other)Infer a mutation that leads to a specific sequence
- classmethod default_alphabet()[source]#
Subclasses may override this method to provide a default alphabet.
By default, it returns None, indicating no default is provided and callers must pass alphabet explicitly.
- Return type:
Optional[BaseAlphabet]
- class mutcleaner.core.RNAAlphabet(include_ambiguous=False)[source]#
Bases:
BaseAlphabetRNA alphabet (A, U, C, G)
Methods
get_invalid_chars(sequence)Get set of invalid characters in sequence
is_valid_char(char)Check if character is valid in this alphabet
is_valid_sequence(sequence)Check if entire sequence is valid
validate_sequence(sequence)Validate sequence and raise error if invalid
- class mutcleaner.core.RNASequence(sequence, alphabet=None, name=None, metadata=None)[source]#
Bases:
BaseSequenceRNA sequence with nucleotide validation
Methods
apply_mutation(mutation)Apply a mutation or set of mutations to the sequence and return a new sequence.
Back-transcribe RNA sequence into DNA sequence
Subclasses may override this method to provide a default alphabet.
get_subsequence(start[, end])get subsequence (0-indexed, inclusive)
infer_mutation(other)Infer a mutation that leads to a specific sequence
Get reverse complement of RNA sequence
translate([codon_table, start_at_first_met, ...])Translate RNA sequence into amino acid sequence using this codon table.
- classmethod default_alphabet()[source]#
Subclasses may override this method to provide a default alphabet.
By default, it returns None, indicating no default is provided and callers must pass alphabet explicitly.
- Return type:
Optional[BaseAlphabet]
- translate(codon_table=None, start_at_first_met=False, stop_at_stop_codon=False, require_mod3=True, start=None, end=None)[source]#
Translate RNA sequence into amino acid sequence using this codon table.
- Parameters:
codon_table (
Optional[CodonTable]) – Codon table to use for translation. If None, uses standard genetic code.start_at_first_met (
bool) – Start translation at the first start codon if found.stop_at_stop_codon (
bool) – Stop translation when a stop codon is encountered.require_mod3 (
bool) – Whether the sequence must be a multiple of 3 in length.start (
Optional[int]) – Custom 0-based start position. Overrides start_at_first_met.end (
Optional[int]) – Custom 0-based end position. Overrides stop_at_stop_codon.
- Return type:
- Returns:
Translated amino acid sequence.
- mutcleaner.core.create_pipeline(data, name=None, **kwargs)[source]#
Create a new pipeline with initial data
- Return type:
- mutcleaner.core.multiout_step(**outputs)[source]#
Decorator factory for multi-output pipeline functions.
- Parameters:
**outputs (
str) – Named outputs. Use ‘main’ to specify which output is the main data flow. If ‘main’ is not specified, the first return value is treated as main.- Return type:
Callable[[Callable[[ParamSpec(P, bound=None)],Union[Tuple[Any,...],Any]]],Callable[[ParamSpec(P, bound=None)],MultiOutput]]
Examples
>>> @multiout_step(stats="statistics", plot="visualization") ... def analyze_data(data): ... # return (main, stats, plot) ... return processed, stats_dict, plot_obj
- mutcleaner.core.pipeline_step(_func=None, *, name=None)[source]#
- Overloads:
_func (Callable[P, R]) → Callable[P, R]
name (Optional[str]) → Callable[[Callable[P, R]], Callable[P, R]]
Decorator for single-output pipeline functions.
Use this for functions that return a single value (including tuples as single values). For multiple outputs, use @multiout_step instead.
- Parameters:
name (
Optional[str]) – Custom name for the step. If None, uses function name.
Examples
>>> @pipeline_step ... def process(data): ... return processed_data # Single output
>>> @pipeline_step("process_data") ... def process(data): ... return processed_data # Single output
>>> @pipeline_step() ... def get_coordinates(): ... return (10, 20) # Single tuple output
Modules