mutcleaner.cleaners.basic_cleaners#

Functions

`add_columns`(dataset, columns_to_add[, overwrite])	Add multiple constant-valued columns to a DataFrame using a dictionary.
`add_sequences_to_dataset`(dataset, ...[, ...])	Add full wild-type sequences to the dataset from sequence dictionary
`aggregate_labels_by_name`(dataset, ...[, ...])	Resolve/aggregate label columns for rows sharing the same (composite) name.
`apply_mutations_to_sequences`(dataset[, ...])	Apply mutations to sequences to generate mutated sequences.
`average_labels_by_name`(dataset, ...[, ...])	Average label columns for rows sharing the same name.
`convert_data_types`(dataset, type_conversions)	Convert data types for specified columns.
`convert_to_mutation_dataset_format`(df[, ...])	Convert a mutation DataFrame to the format required by MutationDataset.from_dataframe().
`extract_and_rename_columns`(dataset, ...[, ...])	Extract useful columns and rename them to standard format.
`filter_and_clean_data`(dataset[, filters, ...])	Filter and clean data based on specified conditions.
`infer_mutations_from_sequences`(dataset[, ...])	Infer mutations by comparing wild-type sequences with mutated sequences.
`infer_wildtype_sequences`(dataset[, ...])	Infer wild-type sequences from mutated sequences and add WT rows.
`merge_columns`(dataset, columns_to_merge, ...)	Merge multiple columns into a single column using a separator
`read_dataset`(file_path[, file_format])	Read dataset from specified file format and return as a pandas DataFrame.
`remap_mutation_positions_by_name`(dataset[, ...])	Remap mutation positions using name-specific offsets or position maps.
`replace_in_column`(df, name_column, old[, new])	Replace or remove a literal substring in a specified DataFrame column.
`split_columns`(dataset, column_to_split, ...)	Split a single column into multiple columns using a separator
`subtract_labels_by_wt`(dataset, name_column, ...)	For each group defined by `name_column`, subtract the group's WT label values (where `mutation_column == wt_identifier`) from all rows' label values.
`validate_mutations`(dataset[, ...])	Validate and format mutation information.

mutcleaner.cleaners.basic_cleaners.add_columns(dataset, columns_to_add, overwrite=True)[source]#

Add multiple constant-valued columns to a DataFrame using a dictionary.

Parameters:

dataset (DataFrame) – The input DataFrame to which the ‘dataset_name’ will be added.
columns_to_add (Dict[str, Any]) – A dictionary where keys are the new column names and values are the constant values to assign to all rows in those columns.
overwrite (bool) – Whether to overwrite existing columns with the same name. If False, a warning will be printed and existing columns will be skipped.

Return type:

DataFrame

Returns:

The DataFrame with the additional columns.

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'mut_info' : ['A0G', 'V1D', 'K2A'],
...     'mut_seq' : ['GVKDF', 'ADKDF', 'AVADF']})
>>> new_cols = {
...     "name": "protein1",
...     "wt_seq": "AGKDF"
... }
>>> df = add_columns(df, columns_to_add=new_cols)
>>> print(list(df.columns))
['mut_info', 'mut_seq', 'name', 'wt_seq']

mutcleaner.cleaners.basic_cleaners.add_sequences_to_dataset(dataset, sequence_source, name_column='name', sequence_column='sequence', header_parser=None)[source]#

Add full wild-type sequences to the dataset from sequence dictionary

This function maps sequences from a dictionary to the dataset. Records without matching sequences are separated into the failed dataset.

Parameters:

dataset (DataFrame) – Dataset containing protein names
sequence_source (Union[Dict[str, str], str, Path]) – Mapping from protein name to full wild-type sequence
name_column (str) – Column name containing protein identifiers

Return type:

Tuple[DataFrame, DataFrame]

Returns:

(successful_dataset, failed_dataset) - datasets with and without sequences

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'name': ['prot1', 'prot2', 'prot3'],
...     'score': [1.0, 2.0, 3.0]
... })
>>> from typing import Dict
>>> from mutcleaner.cleaners.basic_cleaners import add_sequences_to_dataset
>>> from mutcleaner.loaders import load_sequences

>>> seq_dict = {'prot1': 'AKCD', 'prot2': 'EFGH'}
>>> successful, failed = add_sequences_to_dataset(df, seq_dict)
>>> print(len(successful))  # Should be 2
2
>>> print(len(failed))  # Should be 1
1

mutcleaner.cleaners.basic_cleaners.aggregate_labels_by_name(dataset, name_columns, label_columns, remove_origin_columns=True, strategy='mean', *, nearest_by=None, nearest_weights=None, output_suffix=None)[source]#

Resolve/aggregate label columns for rows sharing the same (composite) name. Supports built-in strategies (‘mean’, ‘first’, ‘nearest’) or a custom callable.

Parameters:

dataset (DataFrame) – Input dataframe.
name_columns (Union[str, Sequence[str]]) – Column name(s) used for grouping. Supports one or multiple columns.
label_columns (Union[str, Sequence[str]]) – Label column(s) to resolve/aggregate.
remove_origin_columns (bool) – If True, return one row per name with resolved labels using the original label column names. If False, keep original rows and merge resolved values back as extra columns (see output_suffix).
strategy (Union[str, Callable[[DataFrame, List[str]], Series]]) –
- “mean”: per-group numeric mean for each label (NaNs ignored). - “first”: take the first row in each group for the labels. - “nearest”: pick the row minimizing a multi-criteria distance; requires nearest_by. - Callable(group_df, label_cols) -> pd.Series: custom resolver that returns a single-row Series whose index includes the label columns.
nearest_by (Union[Sequence[Tuple[str, float]], Dict[str, float], None]) – Criteria for “nearest” strategy. Provide either: * a list of (column, target) pairs in priority order, or * a dict {column: target}. When dict is used, priority order is the insertion order (Python 3.7+ preserves dict insertion order). The distance is computed per criterion as abs(col - target), and rows are compared lexicographically by the (possibly weighted) distance tuple.
nearest_weights (Union[Sequence[Tuple[str, float]], Dict[str, float], None]) – Optional weights for “nearest” strategy (same columns as in nearest_by). If provided, each distance is multiplied by its weight before comparison. Defaults to all weights = 1.0.
output_suffix (Optional[str]) – Suffix used when remove_origin_columns=False to name merged columns. Default depends on strategy: - “_mean_by_name”, “_first_by_name”, “_nearest_by_name”, or “_custom_by_name”.

Return type:

DataFrame

Returns:

If remove_origin_columns is True: one row per group with resolved labels. Otherwise: original rows + resolved label columns (with suffix).

Raises:

KeyError – If required columns are missing.
ValueError – If label columns are non-numeric for “mean”, or if “nearest” is selected without nearest_by, or if custom strategy returns invalid shape/index.

Examples

Mean aggregation, one row per name:

>>> import pandas as pd
>>> df = pd.DataFrame({
...     "gene": ["A","A","B"],
...     "specie": ["hs","hs","hs"],
...     "ddG": [1.0, 3.0, 2.0],
...     "dTm": [10.0, 30.0, 20.0]
... })
>>> aggregate_labels_by_name(
...     df,
...     name_columns=["gene","specie"],
...     label_columns=["ddG","dTm"],
...     strategy="mean",
...     remove_origin_columns=True
... )
  gene specie  ddG   dTm
0    A     hs  2.0  20.0
1    B     hs  2.0  20.0

Nearest row by multiple criteria (temperature, then pH) with weights, keeping original rows and adding resolved columns:

>>> df2 = pd.DataFrame({
...     "gene": ["A","A","A","B"],
...     "temperature": [20.0, 24.5, 26.0, 25.0],
...     "pH": [7.2, 7.4, 7.3, 7.5],
...     "ddG": [1.1, 1.9, 2.2, 2.0],
...     "dTm": [9.0, 20.0, 21.0, 20.0],
... })
>>> aggregate_labels_by_name(
...     df2,
...     name_columns="gene",
...     label_columns=["ddG","dTm"],
...     strategy="nearest",
...     nearest_by=[("temperature", 25.0), ("pH", 7.4)],
...     nearest_weights={"temperature": 2.0, "pH": 1.0},
...     remove_origin_columns=False
... ).filter(regex="^gene$|_nearest_by_name$|^ddG$|^dTm$")
  gene  ddG   dTm  ddG_nearest_by_name  dTm_nearest_by_name
0    A  1.1   9.0                  1.9                 20.0
1    A  1.9  20.0                  1.9                 20.0
2    A  2.2  21.0                  1.9                 20.0
3    B  2.0  20.0                  2.0                 20.0

mutcleaner.cleaners.basic_cleaners.apply_mutations_to_sequences(dataset, sequence_column='sequence', name_column='name', mutation_column='mut_info', position_columns=None, mutation_sep=',', is_zero_based=True, sequence_type='protein', mutation_type=None, alphabet=None, num_workers=4)[source]#

Apply mutations to sequences to generate mutated sequences.

This function takes mutation information and applies it to wild-type sequences to generate the corresponding mutated sequences. It supports parallel processing and can handle position-based sequence extraction.

Parameters:

dataset (DataFrame) – Input dataset containing mutation information and sequence data
sequence_column (str) – Column name containing wild-type sequences
name_column (str) – Column name containing protein identifiers
mutation_column (str) – Column name containing mutation information
position_columns (Optional[Dict[str, str]]) – Position column mapping {“start”: “start_col”, “end”: “end_col”} Used for extracting sequence regions
mutation_sep (str) – Separator used to split multiple mutations in a single string
is_zero_based (bool) – Whether origin mutation positions are zero-based
sequence_type (str) – Type of sequence (‘protein’, ‘dna’, ‘rna’)
num_workers (int) – Number of parallel workers for processing, set to -1 for all available CPUs

Return type:

Tuple[DataFrame, DataFrame]

Returns:

(successful_dataset, failed_dataset) - datasets with and without errors

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'name': ['prot1', 'prot1', 'prot2'],
...     'sequence': ['AKCDEF', 'AKCDEF', 'FEGHIS'],
...     'mut_info': ['A0K', 'C2D', 'E1F'],
...     'score': [1.0, 2.0, 3.0]
... })
>>> successful, failed = apply_mutations_to_sequences(df)
>>> print(successful['mut_seq'].tolist())
['KKCDEF', 'AKDDEF', 'FFGHIS']
>>> print(len(failed))  # Should be 0 if all mutations are valid
0

mutcleaner.cleaners.basic_cleaners.average_labels_by_name(dataset, name_columns, label_columns, remove_origin_columns=True)[source]#

Average label columns for rows sharing the same name.

Parameters:

df (pd.DataFrame) – Input dataframe.
name_columns (Union[str, Sequence[str]]) – Column name(s) used for grouping (the “name” keys). Supports a single column or multiple columns for composite keys.
label_cols (Union[str, Sequence[str]]) – Label column(s) to average.
remove_origin_columns (bool) – If True, return one row per name with averaged labels using the ORIGINAL label column names (deduplicated by design). If False, keep original rows and add per-name mean columns named <label>_mean_by_name.

Return type:

DataFrame

Returns:

If remove_origin_columns is True: columns = [name_column] + label_columns (averaged), one row per unique name. - If False: same rows as input plus <label>_mean_by_name columns.

Raises:

KeyError – If name_column or any of label_columns is missing.
ValueError – If any label column is non-numeric.v

Examples

>>> df = pd.DataFrame({
...     "gene": ["A","A","B"],
...     "specie": ["hs","hs","hs"],
...     "y1": [1.0, 3.0, 2.0],
...     "y2": [10.0, 30.0, 20.0]
... })
>>> # multiple name keys
>>> average_labels_by_name(df, ["gene","specie"], ["y1","y2"], remove_origin_columns=True)
  gene specie   y1    y2
0    A     hs  2.0  20.0
1    B     hs  2.0  20.0

>>> # keep original rows and add means
>>> average_labels_by_name(df, "gene", ["y1","y2"], remove_origin_columns=False)
  gene specie   y1    y2  y1_mean_by_name  y2_mean_by_name
0    A     hs  1.0  10.0              2.0             20.0
1    A     hs  3.0  30.0              2.0             20.0
2    B     hs  2.0  20.0              2.0             20.0

mutcleaner.cleaners.basic_cleaners.convert_data_types(dataset, type_conversions, handle_errors='coerce', optimize_memory=True, use_batch_processing=False, chunk_size=10000)[source]#

Convert data types for specified columns.

This function provides unified data type conversion with error handling options. Supports pandas, numpy, and Python built-in types with memory optimization.

Parameters:

dataset (DataFrame) – Input dataset with columns to be converted
type_conversions (Dict[str, Union[str, Type, dtype]]) – Type conversion mapping in format {column_name: target_type} Supported formats: - String types: ‘float’, ‘int’, ‘str’, ‘category’, ‘bool’, ‘datetime’ - Numpy types: np.float32, np.float64, np.int32, np.int64, etc. - Pandas types: ‘Int64’, ‘Float64’, ‘string’, ‘boolean’ - Python types: float, int, str, bool
handle_errors (str) – Error handling strategy: ‘raise’, ‘coerce’, or ‘ignore’
optimize_memory (bool) – Whether to automatically optimize memory usage by choosing smaller dtypes
use_batch_processing (bool) – Whether to use batch processing for large datasets
chunk_size (int) – Chunk size when using batch processing

Return type:

DataFrame

Returns:

Dataset with converted data types

Examples

>>> import pandas as pd
>>> import numpy as np
>>> df = pd.DataFrame({
...     'score': ['1.5', '2.3', '3.7'],
...     'count': ['10', '20', '30'],
...     'name': [123, 456, 789],
...     'flag': ['True', 'False', 'True']
... })
>>> conversions = {
...     'score': np.float32,
...     'count': 'Int64',
...     'name': 'string',
...     'flag': 'boolean'
... }
>>> result = convert_data_types(df, conversions)

mutcleaner.cleaners.basic_cleaners.convert_to_mutation_dataset_format(df, name_column='name', mutation_column='mut_info', sequence_column=None, mutated_sequence_column='mut_seq', sequence_type='protein', label_column='score', include_wild_type=False, mutation_set_prefix='set', is_zero_based=False, additional_metadata=None)[source]#

Convert a mutation DataFrame to the format required by MutationDataset.from_dataframe().

This function supports two input formats: 1. Format with WT rows: Contains explicit ‘WT’ entries with wild-type sequences 2. Format with sequence column: Each row contains the wild-type sequence

Parameters:

df – Input DataFrame. Supports two formats:

Return type:

Tuple[pd.DataFrame, Dict[str, BaseSequence]]

Returns:

(converted_dataframe, reference_sequences_dict)

converted_dataframe: DataFrame in MutationDataset.from_dataframe() format reference_sequences_dict: Dictionary mapping reference_id to wild-type sequences (extracted from WT rows in Format 1 or sequence column in Format 2)

Raises:

ValueError – If required columns are missing or mutation strings cannot be parsed.

Examples

>>> import pandas as pd

Format 1: With WT rows and multi-mutations

>>> df1 = pd.DataFrame({
...     'name': ['prot1', 'prot1', 'prot1', 'prot2', 'prot2'],
...     'mut_info': ['A0S,Q1D', 'C2D', 'WT', 'E0F', 'WT'],
...     'mut_seq': ['SDCDEF', 'AQDDEF', 'AQCDEF', 'FGHIGHK', 'EGHIGHK'],
...     'score': [1.5, 2.0, 0.0, 3.0, 0.0]
... })
>>> result_df1, ref_seqs1 = convert_to_mutation_dataset_format(df1)
>>> # Input has 5 rows but output has 6 rows (A0S,Q1D -> 2 rows)

Format 2: With sequence column and multi-mutations

>>> df2 = pd.DataFrame({
...     'name': ['prot1', 'prot1', 'prot2'],
...     'sequence': ['AKCDEF', 'AKCDEF', 'FEGHIS'],
...     'mut_info': ['A0K,C2D', 'Q1P', 'E1F'],
...     'score': [1.5, 2.0, 3.0],
...     'mut_seq': ['KKDDEF', 'APCDEF', 'FFGHIS']
... })
>>> result_df2, ref_seqs2 = convert_to_mutation_dataset_format(
...     df2, sequence_column='sequence'
... )
>>> print(ref_seqs2['prot1'])
AKCDEF
>>> # First row generates 2 output rows for A0K and C2D mutations

mutcleaner.cleaners.basic_cleaners.extract_and_rename_columns(dataset, column_mapping, required_columns=None)[source]#

Extract useful columns and rename them to standard format.

This function extracts specified columns from the input dataset and renames them according to the provided mapping. It helps standardize column names across different datasets.

Parameters:

dataset (DataFrame) – Input dataset containing the data to be processed
column_mapping (Dict[str, str]) – Column name mapping from original names to new names Format: {original_column_name: new_column_name}
required_columns (Optional[Sequence[str]]) – Required column names. If None, extracts all mapped columns

Return type:

DataFrame

Returns:

Dataset with extracted and renamed columns

Raises:

ValueError – If required columns are missing from the input dataset

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'uniprot_ID': ['P12345', 'Q67890'],
...     'mutation_type': ['A123B', 'C456D'],
...     'score_value': [1.5, -2.3],
...     'extra_col': ['x', 'y']
... })
>>> mapping = {
...     'uniprot_ID': 'name',
...     'mutation_type': 'mut_info',
...     'score_value': 'label'
... }
>>> result = extract_and_rename_columns(df, mapping)
>>> print(result.columns.tolist())
['name', 'mut_info', 'label']

mutcleaner.cleaners.basic_cleaners.filter_and_clean_data(dataset, filters=None, exclude_patterns=None, drop_na_columns=None)[source]#

Filter and clean data based on specified conditions.

This function provides flexible data filtering and cleaning capabilities, including value-based filtering, pattern exclusion, and null value removal.

Parameters:

dataset (DataFrame) – Input dataset to be filtered and cleaned
filters (Optional[Dict[str, Union[Any, Callable[[Series], Series]]]]) – Filter conditions in format {column_name: condition_value_or_function} If value is callable, it will be applied to the column
exclude_patterns (Optional[Dict[str, Union[str, List[str]]]]) – Exclusion patterns in format {column_name: regex_pattern_or_list} Rows matching these patterns will be excluded
drop_na_columns (Optional[List[str]]) – List of column names where null values should be dropped

Return type:

Tuple[DataFrame, DataFrame]

Returns:

A tuple containing two datasets:

successful_datasetpd.DataFrame
Rows that pass all filtering conditions.
failed_datasetpd.DataFrame
Rows that fail at least one filtering condition.

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'mut_type': ['A123B', 'wt', 'C456D', 'insert', 'E789F'],
...     'score': [1.5, 2.0, '-', 3.2, 4.1],
...     'quality': ['good', 'bad', 'good', 'good', None]
... })
>>> filters = {'score': lambda x: x != '-'}
>>> exclude_patterns = {'mut_type': ['wt', 'insert']}
>>> drop_na_columns = ['quality']
>>> successful, failed = filter_and_clean_data(df, filters, exclude_patterns, drop_na_columns)
>>> print(len(successful))  # Should be 1 (A123B row)
1

mutcleaner.cleaners.basic_cleaners.infer_mutations_from_sequences(dataset, wt_sequence_column='wt_seq', mut_sequence_column='mut_seq', mutation_sep=',', sequence_type='protein', num_workers=4)[source]#

Infer mutations by comparing wild-type sequences with mutated sequences.

This function takes wild-type and mutated sequences and identifies the mutations that occurred by comparing them position by position. It only handles sequences of equal length. Uses parallel processing for large datasets.

Parameters:

dataset (DataFrame) – Input dataset containing wild-type and mutated sequence data. Each dict should contain the specified column keys.
wt_sequence_column (str) – Column key containing wild-type sequences
mut_sequence_column (str) – Column key containing mutated sequences
mutation_sep (str) – Separator used to join multiple mutations in output string
sequence_type (str) – Type of sequence (‘protein’, ‘dna’, ‘rna’)
num_workers (int) – Number of parallel workers for processing, set to -1 for all available CPUs

Return type:

Tuple[DataFrame, DataFrame]

Returns:

(successful_results, failed_results) - datasets with and without errors

Examples

>>> dataset = pd.DataFrame([
...     {'name': 'prot1', 'wt_seq': 'AKCDEF', 'mut_seq': 'KKCDEF'},
...     {'name': 'prot1', 'wt_seq': 'AKCDEF', 'mut_seq': 'AKDDEF'},
...     {'name': 'prot2', 'wt_seq': 'FEGHIS', 'mut_seq': 'FFGHIS'},
...     {'name': 'prot3', 'wt_seq': 'ABC', 'mut_seq': 'ASC'}
... ])
>>> successful, failed = infer_mutations_from_sequences(dataset)
>>> successful
    name  wt_seq mut_seq inferred_mutations
0  prot1  AKCDEF  KKCDEF                A0K
1  prot1  AKCDEF  AKDDEF                C2D
2  prot2  FEGHIS  FFGHIS                E1F
>>> failed
    name wt_seq mut_seq                                  error_message
3  prot3    ABC    ABCD  Invalid characters in Protein sequence: {'B'}

mutcleaner.cleaners.basic_cleaners.infer_wildtype_sequences(dataset, name_column='name', mutation_column='mut_info', sequence_column='mut_seq', label_columns=None, wt_label=0.0, mutation_sep=',', is_zero_based=False, sequence_type='protein', handle_multiple_wt='error', num_workers=4)[source]#

Infer wild-type sequences from mutated sequences and add WT rows.

This function takes mutated sequences and their corresponding mutations to infer the original wild-type sequences. For each protein, it adds WT row(s) to the dataset with the inferred wild-type sequence.

Parameters:

dataset (DataFrame) – Input dataset containing mutated sequences and mutation information
name_column (str) – Column name containing protein identifiers
mutation_column (str) – Column name containing mutation information
sequence_column (str) – Column name containing mutated sequences
label_columns (Optional[List[str]]) – List of label column names to preserve
wt_label (float) – Wild type score for WT rows
mutation_sep (str) – Separator used to split multiple mutations in a single string
is_zero_based (bool) – Whether origin mutation positions are zero-based
sequence_type (Literal['protein', 'dna', 'rna']) – Type of sequence (‘protein’, ‘dna’, ‘rna’)
handle_multiple_wt (Literal['error', 'separate', 'first']) – How to handle multiple wild-type sequences: ‘separate’, ‘first’, or ‘error’
num_workers (int) – Number of parallel workers for processing, set to -1 for all available CPUs

Return type:

Tuple[DataFrame, DataFrame]

Returns:

(successful_dataset, problematic_dataset) - datasets with added WT rows

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'name': ['prot1', 'prot1', 'prot2'],
...     'mut_info': ['A0S', 'C2D', 'E0F'],
...     'mut_seq': ['SQCDEF', 'AQDDEF', 'FGHIGHK'],
...     'score': [1.0, 2.0, 3.0]
... })
>>> success, failed = infer_wildtype_sequences(
...     df, label_columns=['score']
... )
>>> print(len(success))  # Should have original rows + WT rows

mutcleaner.cleaners.basic_cleaners.merge_columns(dataset, columns_to_merge, new_column_name, separator='_', drop_original=False, na_rep=None, prefix=None, suffix=None, custom_formatter=None)[source]#

Merge multiple columns into a single column using a separator

This function combines values from multiple columns into a new column, with flexible formatting options.

Parameters:

dataset (DataFrame) – Input dataset
columns_to_merge (List[str]) – List of column names to merge
new_column_name (str) – Name for the new merged column
separator (str) – Separator to use between values
drop_original (bool) – Whether to drop the original columns after merging
na_rep (Optional[str]) – String representation of NaN values. If None, NaN values are skipped.
prefix (Optional[str]) – Prefix to add to the merged value
suffix (Optional[str]) – Suffix to add to the merged value
custom_formatter (Optional[Callable[[Series], str]]) – Custom function to format each row. Takes a pd.Series and returns a string. If provided, ignores separator, prefix, suffix parameters.

Return type:

DataFrame

Returns:

Dataset with the new merged column

Examples

Basic usage:

>>> df = pd.DataFrame({
...     'gene': ['BRCA1', 'TP53', 'EGFR'],
...     'position': [100, 200, 300],
...     'mutation': ['A', 'T', 'G']
... })
>>> result = merge_columns(df, ['gene', 'position', 'mutation'], 'mutation_id', separator='_')
>>> print(result['mutation_id'])
0    BRCA1_100_A
1     TP53_200_T
2     EGFR_300_G

With prefix and suffix:

>>> result = merge_columns(
...     df, ['gene', 'position'], 'gene_pos',
...     separator=':', prefix='[', suffix=']'
... )
>>> print(result['gene_pos'])
0    [BRCA1:100]
1     [TP53:200]
2     [EGFR:300]

Handling NaN values:

>>> df_with_nan = pd.DataFrame({
...     'col1': ['A', 'B', None],
...     'col2': ['X', None, 'Z'],
...     'col3': [1, 2, 3]
... })
>>> result = merge_columns(
...     df_with_nan, ['col1', 'col2', 'col3'], 'merged',
...     separator='-', na_rep='NA'
... )
>>> print(result['merged'])
0    A-X-1
1    B-NA-2
2    NA-Z-3

Custom formatter:

>>> def format_mutation(row):
...     return f"{row['gene']}:{row['position']}{row['mutation']}"
>>> result = merge_columns(
...     df, ['gene', 'position', 'mutation'], 'hgvs',
...     custom_formatter=format_mutation
... )
>>> print(result['hgvs'])
0    BRCA1:100A
1     TP53:200T
2     EGFR:300G

mutcleaner.cleaners.basic_cleaners.read_dataset(file_path, file_format=None, **kwargs)[source]#

Read dataset from specified file format and return as a pandas DataFrame.

Parameters:

file_path (Union[str, Path]) – Path to the dataset file
file_format (Optional[str]) – Format of the dataset file (“csv”, “tsv”, “xlsx”, etc.)
kwargs (Dict[str, Any]) – Additional keyword arguments for file reading

Return type:

DataFrame

Returns:

Dataset loaded from the specified file

Examples

>>> # Specify file_format parameter
>>> df = read_dataset("data.csv", "csv")
>>>
>>> # Detect file_format automatically
>>> df = read_dataset("data.csv")

mutcleaner.cleaners.basic_cleaners.remap_mutation_positions_by_name(dataset, position_offsets=None, position_maps=None, name_column='name', mutation_column='mut_info', mutation_separator=',', strict=True)[source]#

Remap mutation positions using name-specific offsets or position maps.

Fixed offsets are suitable when source and target numbering systems differ by a constant value. Explicit position maps are suitable when residue correspondence is non-linear because of insertions, deletions, or other numbering differences.

When both rules are provided for the same name, an explicitly mapped position takes precedence over the fixed offset. Positions absent from the explicit map fall back to the fixed offset when one is available.

Parameters:

dataset (DataFrame) – Input dataset.
position_offsets (Optional[Dict[Any, int]]) – Mapping from protein names to fixed residue-position offsets.
position_maps (Optional[Dict[Any, Dict[int, int]]]) – Mapping from protein names to explicit source-to-target position maps.
name_column (str) – Column containing protein or dataset names.
mutation_column (str) – Column containing mutation descriptions.
mutation_separator (str) – Separator between individual mutations.
strict (bool) – Whether to raise an error for invalid mutation strings or positions missing from an explicit map when no fixed offset is available. If False, unsupported mutation tokens or unmapped positions are left unchanged.

Return type:

DataFrame

Returns:

Dataset with remapped mutation positions.

Raises:

ValueError – If required columns are missing, neither remapping rule is provided, or mutation positions cannot be remapped when strict=True.

Examples

Create an example dataset:

>>> df = pd.DataFrame({
...     "name": ["proteinA", "proteinB"],
...     "mut_info": ["S10T,F14V", "A5G"],
... })

Apply a fixed three-residue offset:

>>> result = remap_mutation_positions_by_name(
...     df,
...     position_offsets={"proteinA": 3},
... )
>>> result["mut_info"].tolist()
['S13T,F17V', 'A5G']

Apply a non-linear explicit position map:

>>> result = remap_mutation_positions_by_name(
...     df,
...     position_maps={"proteinA": {10: 12, 14: 19}},
... )
>>> result["mut_info"].tolist()
['S12T,F19V', 'A5G']

Combine a fixed offset with an explicit position map:

>>> result = remap_mutation_positions_by_name(
...     df,
...     position_offsets={"proteinA": 3},
...     position_maps={"proteinA": {14: 20}},
... )
>>> result["mut_info"].tolist()
['S13T,F20V', 'A5G']

mutcleaner.cleaners.basic_cleaners.replace_in_column(df, name_column, old, new='')[source]#

Replace or remove a literal substring in a specified DataFrame column.

This function performs a literal (non-regex) string replacement on the specified column. It replaces occurrences of a target substring (‘old’) with a new substring (‘new’). If ‘new’ is not provided, the target substring is removed.

Parameters:

df (DataFrame) – Input DataFrame containing at least one column with string identifiers.
name_column (str) – Name of the column on which to perform replacements.
old (str) – Non-empty substring to be replaced or removed.
new (str) – Replacement substring. If omitted or None, occurrences of ‘old’ will be removed. Defaults to “” (deletion).

Return type:

DataFrame

Returns:

The DataFrame with the specified replacements applied.

Raises:

ValueError – If ‘old’ is an empty string.
TypeError – If ‘df’ is not a pd.DataFrame or ‘old’ is not a str
KeyError – If ‘name_column’ does not exist in the DataFrame.

Examples

Remove the substring “.pdb”:

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'name': ['prot1.pdb', 'prot1.pdb', 'prot1.pdb', 'prot2.pdb', 'prot2.pdb'],
...     'mut_info': ['A0S,Q1D', 'C2D', 'WT', 'E0F', 'WT'],
...     'mut_seq': ['SDCDEF', 'AQDDEF', 'AQCDEF', 'FGHIGHK', 'EGHIGHK'],
...     'score': [1.5, 2.0, 0.0, 3.0, 0.0]
... )
>>> replace_in_column(df, "name", ".pdb")
    name mut_info  mut_seq  score
0  prot1  A0S,Q1D   SDCDEF    1.5
1  prot1      C2D   AQDDEF    2.0
2  prot1       WT   AQCDEF    0.0
3  prot2      E0F  FGHIGHK    3.0
4  prot2       WT  EGHIGHK    0.0

Replace “.pdb” with “.pross”:

>>> replace_in_column(df, "name", ".pdb", ".pross")
        name mut_info  mut_seq  score
prot1.pross  A0S,Q1D   SDCDEF    1.5
prot1.pross      C2D   AQDDEF    2.0
prot1.pross       WT   AQCDEF    0.0
prot2.pross      E0F  FGHIGHK    3.0
prot2.pross       WT  EGHIGHK    0.0

mutcleaner.cleaners.basic_cleaners.split_columns(dataset, column_to_split, new_column_names, separator='_', max_splits=None, drop_original=False, fill_value='NaN', strip_whitespace=True, regex=False, custom_splitter=None)[source]#

Split a single column into multiple columns using a separator

This function splits values from one column into multiple new columns, with flexible splitting options.

Parameters:

dataset (DataFrame) – Input dataset
column_to_split (str) – Name of the column to split
new_column_names (List[str]) – List of names for the new columns
separator (str) – Separator to use for splitting. Can be regex pattern if regex=True.
max_splits (Optional[int]) – Maximum number of splits to perform. If None, splits on all occurrences.
drop_original (bool) – Whether to drop the original column after splitting
fill_value (Optional[str]) – Value to use when split results have fewer parts than new_column_names. If None, uses NaN for missing parts.
strip_whitespace (bool) – Whether to strip whitespace from split results
regex (bool) – Whether to treat separator as a regex pattern
custom_splitter (Optional[Callable[[str], List[str]]]) – Custom function to split each value. Takes a string and returns a list of strings. If provided, ignores separator, max_splits, regex parameters.

Return type:

DataFrame

Returns:

Dataset with the new split columns

Examples

Basic usage:

>>> df = pd.DataFrame({
...     'mutation_id': ['BRCA1_100_A', 'TP53_200_T', 'EGFR_300_G'],
...     'score': [0.95, 0.87, 0.92]
... })
>>> result = split_columns(
...     df, 'mutation_id', ['gene', 'position', 'mutation'], separator='_'
... )
Splitting column 'mutation_id' into ['gene', 'position', 'mutation']...
Splitting using separator: '_'
Successfully created split columns: ['gene', 'position', 'mutation']
>>> print(result[['gene', 'position', 'mutation']])
    gene position mutation
0  BRCA1      100        A
1   TP53      200        T
2   EGFR      300        G

With max_splits:

>>> df = pd.DataFrame({
...     'path': ['home/user/documents/file.txt', 'data/analysis/results.csv']
... })
>>> result = split_columns(
...     df, 'path', ['root', 'rest'], separator='/', max_splits=1
... )
Splitting column 'path' into ['root', 'rest']...
Splitting using separator: '/'
Successfully created split columns: ['root', 'rest']
>>> print(result[['root', 'rest']])
   root                     rest
0  home  user/documents/file.txt
1  data     analysis/results.csv

Handling insufficient splits with fill_value:

>>> df = pd.DataFrame({
...     'incomplete': ['A_B', 'X_Y_Z', 'M']
... })
>>> result = split_columns(
...     df, 'incomplete', ['col1', 'col2', 'col3'],
...     separator='_', fill_value='MISSING'
... )
Splitting column 'incomplete' into ['col1', 'col2', 'col3']...
Splitting using separator: '_'
Successfully created split columns: ['col1', 'col2', 'col3']
>>> print(result[['col1', 'col2', 'col3']])
  col1     col2     col3
0    A        B  MISSING
1    X        Y        Z
2    M  MISSING  MISSING

Using regex separator:

>>> df = pd.DataFrame({
...     'text': ['word1-word2_word3', 'itemA|itemB-itemC']
... })
>>> result = split_columns(
...     df, 'text', ['part1', 'part2', 'part3'],
...     separator=r'[-_|]', regex=True
... )
Splitting column 'text' into ['part1', 'part2', 'part3']...
Splitting using separator: '[-_|]' (regex)
Successfully created split columns: ['part1', 'part2', 'part3']
>>> print(result[['part1', 'part2', 'part3']])
   part1  part2  part3
0  word1  word2  word3
1  itemA  itemB  itemC

Custom splitter:

>>> def parse_coordinates(coord_str):
...     # Parse "chr1:12345-67890" format
...     parts = coord_str.replace(':', '_').replace('-', '_').split('_')
...     return parts
>>> df = pd.DataFrame({
...     'coordinates': ['chr1:12345-67890', 'chr2:98765-43210']
... })
>>> result = split_columns(
...     df, 'coordinates', ['chromosome', 'start', 'end'],
...     custom_splitter=parse_coordinates
... )
Splitting column 'coordinates' into ['chromosome', 'start', 'end']...
Using custom splitter...
100%|█████████████████████████████| 2/2 [00:00<00:00, xx it/s]
Successfully created split columns: ['chromosome', 'start', 'end']
>>> print(result[['chromosome', 'start', 'end']])
  chromosome  start    end
0       chr1  12345  67890
1       chr2  98765  43210

Handling NaN values:

>>> df = pd.DataFrame({
...     'data': ['A_B_C', None, 'X_Y']
... })
>>> result = split_columns(
...     df, 'data', ['col1', 'col2', 'col3'], separator='_'
... )
Splitting column 'data' into ['col1', 'col2', 'col3']...
Splitting using separator: '_'
Successfully created split columns: ['col1', 'col2', 'col3']
>>> print(result[['col1', 'col2', 'col3']])
   col1 col2 col3
0     A    B    C
1  None  NaN  NaN
2     X    Y  NaN

mutcleaner.cleaners.basic_cleaners.subtract_labels_by_wt(dataset, name_column, label_columns, mutation_column, wt_identifier='wt', suffix='_minus_wt', in_place=False, drop_wt_row=False)[source]#

For each group defined by name_column, subtract the group’s WT label values (where mutation_column == wt_identifier) from all rows’ label values.

Returns two dataframes via the multi-output step: - successful: rows of groups where subtraction succeeded

(columns overwritten if in_place=True; otherwise new columns <label><suffix> are added). If drop_wt_row=True, WT rows are removed in this output.

failed: one (or more) diagnostic rows per failed group, including an error_message column describing the reason.

Parameters:

dataset (DataFrame)
name_column (str) – Grouping key column.
label_columns (Union[str, Sequence[str]]) – Numeric label column(s) to subtract by WT.
mutation_column (str) – Column that identifies the WT row with wt_identifier.
wt_identifier (str) – Token used to mark the WT row.
suffix (str) – Suffix for new columns (ignored if in_place=True).
in_place (bool) – If True, overwrite original label columns with the deltas. If False, add new columns named <label><suffix>.
drop_wt_row (bool) – If True, drop WT rows from the successful output.

Return type:

Tuple[DataFrame, DataFrame]

Returns:

Examples

>>> df = pd.DataFrame({
...     "name": ["A","A","B","B","B","C","C"],
...     "mut" : ["wt","A0G","wt","K2R","L3F","S1N","T4A"],
...     "y1"  : [10.0,  12.0, 20.0,  22.0,  19.0,  8.0,   6.0],
...     "y2"  : [ 5.0,   4.0, 15.0,  10.0,  18.0,  3.0,   2.0],
... })
>>> successful, failed = subtract_labels_by_wt(
...     dataset=df,
...     name_column="name",
...     label_columns=["y1", "y2"],
...     mutation_column="mut",
...     wt_identifier="wt",
...     in_place=True,
...     drop_wt_row=True,
... )
>>> successful
  name  mut   y1   y2
0    A  A0G  2.0 -1.0
1    B  K2R  2.0 -5.0
2    B  L3F -1.0  3.0
>>> failed["failed"]
  name  mut   y1   y2                        error_message
0    C  S1N  8.0  3.0  No WT row (mutation_column == 'wt')

mutcleaner.cleaners.basic_cleaners.validate_mutations(dataset, mutation_column='mut_info', format_mutations=True, mutation_sep=',', is_zero_based=False, mutation_type=None, alphabet=None, exclude_patterns=None, cache_results=True, num_workers=4)[source]#

Validate and format mutation information.

This function validates mutation strings, optionally formats them to a standard representation, and separates valid and invalid mutations into different datasets. It supports caching for improved performance on datasets with repeated mutations.

Parameters:

dataset (DataFrame) – Input dataset containing mutation information
mutation_column (str) – Name of the column containing mutation information
format_mutations (bool) – Whether to format mutations to standard representation
mutation_sep (str) – Separator used to split multiple mutations in a single string (e.g., ‘A123B,C456D’)
is_zero_based (bool) – Whether origin mutation positions are zero-based
mutation_type (str, optional) – Type of mutation (e.g., ‘CodonMutation’, ‘AminoAcidMutation’)
alphabet (BaseAlphabet, optional) – Alphabet to use for mutation validation
exclude (Union[str, List[str], callable, None], default=None) – Patterns to exclude from validation. Can be: - A regex pattern string (e.g., r’^WT$|^wildtype$’) - A list of exact strings to match (e.g., [‘WT’, ‘wildtype’, ‘UNKNOWN’]) - A list containing both regex patterns (starting with ‘regex:’) and exact strings - A callable that takes a string value and returns True if it should be excluded - None to validate all values
cache_results (bool) – Whether to cache formatting results for performance
num_workers (int) – Number of parallel workers for processing, set to -1 for all available CPUs

Return type:

Tuple[DataFrame, DataFrame]

Returns:

(successful_dataset, failed_dataset) - datasets with valid and invalid mutations

Examples

>>> import pandas as pd
>>> df = pd.DataFrame({
...     'name': ['protein1', 'protein1', 'protein2'],
...     'mut_info': ['A123S', 'C456D,E789F', 'InvalidMut'],
...     'score': [1.5, 2.3, 3.7]
... })
>>> successful, failed = validate_mutations(df, mutation_column='mut_info', mutation_sep=',')
>>> print(len(successful))
2
>>> print(successful['mut_info'].tolist())  # Formatted mutations
['A123S', 'C456D,E789F']
>>> print(len(failed))
1
>>> print(failed['failed']['error_message'].iloc[0])  # Error message for failed mutation
'ValueError: No valid mutations could be parsed...'
>>> # Exclude patterns
>>> df = pd.DataFrame({
...     'name': ['protein1', 'protein1', 'protein2', 'protein3', 'protein4'],
...     'mut_info': ['A123S', 'C456D,E789F', 'WT', 'wildtype', 'UNKNOWN'],
...     'score': [1.5, 2.3, 3.7, 4.1, 5.2]
... })
>>>
>>> # Exclude exact string matches
>>> successful, failed = validate_mutations(
...     df,
...     mutation_column='mut_info',
...     exclude_patterns=['WT', 'wildtype', 'UNKNOWN']
... )
>>>
>>> # Exclude using regex pattern
>>> successful, failed = validate_mutations(
...     df,
...     mutation_column='mut_info',
...     exclude_patterns=r'^(WT|wildtype|UNKNOWN)$'
... )
>>>
>>> # Mixed approach: regex patterns (prefixed with 'regex:') and exact strings
>>> successful, failed = validate_mutations(
...     df,
...     mutation_column='mut_info',
...     exclude_patterns=['regex:^WT$', 'regex:.*type.*', 'UNKNOWN']
... )
>>>
>>> # Using a custom function
>>> exclude_func = lambda x: str(x).upper() in ['WT', 'WILDTYPE'] or 'UNKNOWN' in str(x)
>>> successful, failed = validate_mutations(
...     df,
...     mutation_column='mut_info',
...     exclude_patterns=exclude_func
... )
>>> successful
       name     mut_info  score
0  protein2           WT    3.7
1  protein3     wildtype    4.1
2  protein4      UNKNOWN    5.2
3  protein1        A122S    1.5
4  protein1  C455D,E788F    2.3