Classify utils

classify_utils

This module provides functions for classifying biological sequences into variants.

Contents

extract_features: Extract features from sequences and save to a DataFrame.
process_dataframe: Process a DataFrame and optionally save data and model.
sequence_classification: Perform sequence classification based on provided data and options.
classify: Perform sequence classification pipeline.
extract_features_to_predict: Extract features from sequences for prediction and return as a DataFrame.
process_dataframe_predict: Process a DataFrame for prediction using MinMaxScaler.
predict_data: Predict classes using a trained RandomForestClassifier model.
predict: Predict sequence classes using a trained model.

Note

This module is designed to work with biological sequences and their classifications, allowing researchers to quickly classify and analyze sequence variants.

Todo

Implement tests.

`message = Messages()` `module-attribute`

Set the Message class for logging.

`classify(word, step, save_path, dir_path, get_kmers=False, reference_path=None, dictonary='DNA', chunk_size=100)`

Perform sequence classification pipeline.

This is the main function for the classification module. It performs sequence classification using the specified parameters and options. The function includes feature extraction, model training, and optional saving of data, model, and confusion matrix plot.

Parameters:

Name	Type	Description	Default
`word`	`int`	The length of each k-mer.	required
`step`	`int`	The step size for moving the sliding window.	required
`save_path`	`str`	The path to save the processed data and model files.	required
`dir_path`	`str`	The path to the directory containing sequence data.	required
`get_kmers`	`bool`	Whether to extract exclusive k-mers. Default is False.	`False`
`reference_path`	`str`	The path to the reference sequence data file. Default is None.	`None`
`dictonary`	`str`	The DNA dictionary for k-mer analysis. Default is 'DNA'.	`'DNA'`
`chunk_size`	`int`	The chunk size for loading sequences. Default is 100.	`100`

Returns:

Type	Description
	Message class: A message confirming the classification pipeline has completed.

Source code in python/gramep/classify_utils.py

def classify(
    word: int,
    step: int,
    save_path: str,
    dir_path: str,
    get_kmers: bool = False,
    reference_path: str | None = None,
    dictonary: str = 'DNA',
    chunk_size: int = 100,
):
    """
    Perform sequence classification pipeline.

    This is the main function for the classification module. It performs \
    sequence classification
    using the specified parameters and options. The function includes feature \
    extraction, model training, and optional saving of data, model, and \
    confusion matrix plot.

    Args:
        word (int): The length of each k-mer.
        step (int): The step size for moving the sliding window.
        save_path (str): The path to save the processed data and model files.
        dir_path (str): The path to the directory containing sequence data.
        get_kmers (bool, optional): Whether to extract exclusive k-mers. \
        Default is False.
        reference_path (str, optional): The path to the reference sequence data file. \
        Default is None.
        dictonary (str): The DNA dictionary for k-mer analysis. Default is 'DNA'.
        chunk_size (int, optional): The chunk size for loading sequences. \
        Default is 100.

    Returns:
        Message class: A message confirming the classification pipeline has completed.
    """
    exclusive_kmers = None
    if get_kmers:
        file_list = [
            name for name in listdir(dir_path) if fnmatch(name, '*.fasta')
        ]
        files = [dir_path + '/' + name for name in file_list]
        with joblib_progress(
            'Extracting exclusive k-mers ...', total=len(files)
        ):
            exclusive_kmers = Parallel(n_jobs=-2)(
                delayed(get_only_kmers)(
                    reference_path=reference_path,
                    sequence_path=file,
                    word=word,
                    step=step,
                    dictonary=dictonary,
                    save_path=save_path,
                    chunk_size=chunk_size,
                )
                for file in files
            )

        exclusive_kmers = np.unique(np.concatenate(exclusive_kmers))

        message.info_founded_features(len(exclusive_kmers))

    data_frame = extract_features(
        word=word,
        step=step,
        save_path=save_path,
        dir_path=dir_path,
        dictonary=dictonary,
        variants_kmers=exclusive_kmers,
        chunk_size=chunk_size,
    )

    # Process the feature matrix, ie, do MinMax scaler
    df_process, name_class = process_dataframe(
        data_frame=data_frame,
        dir_path=save_path,
    )
    sequence_classification(
        data_frame=df_process,
        name_class=name_class,
        dir_path=save_path,
    )
    return message.info_done()

`extract_features(word, step, save_path, dir_path, dictonary, variants_kmers=None, chunk_size=100)`

Extract features from sequences and save to a DataFrame.

This function extracts features from sequences located in the specified directory, and then saves the extracted features to a pandas DataFrame. The extracted features are based on the exclusive kmers.

Parameters:

Name	Type	Description	Default
`word`	`int`	The length of each k-mer.	required
`step`	`int`	The step size for moving the sliding window.	required
`save_path`	`str`	The path to save the extracted features DataFrame.	required
`dir_path`	`str`	The path to the directory containing sequence data.	required
`dictonary`	`str`	The DNA dictionary for k-mer analysis.	required
`variants_kmers`	`None`	The exclusive k-mers. Default is None.	`None`
`chunk_size`	`int`	The chunk size for loading sequences. Default is 100.	`100`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: A pandas DataFrame containing the extracted features.

Source code in python/gramep/classify_utils.py

def extract_features(
    word: int,
    step: int,
    save_path: str,
    dir_path: str,
    dictonary: str,
    variants_kmers: None = None,
    chunk_size: int = 100,
) -> pd.DataFrame:
    """
    Extract features from sequences and save to a DataFrame.

    This function extracts features from sequences located in the specified \
    directory, and
    then saves the extracted features to a pandas DataFrame. The extracted \
    features are based
    on the exclusive kmers.

    Args:
        word (int): The length of each k-mer.
        step (int): The step size for moving the sliding window.
        save_path (str): The path to save the extracted features DataFrame.
        dir_path (str): The path to the directory containing sequence data.
        dictonary (str): The DNA dictionary for k-mer analysis.
        variants_kmers (None, optional): The exclusive k-mers. Default is None.
        chunk_size (int, optional): The chunk size for loading sequences. \
        Default is 100.

    Returns:
        pd.DataFrame: A pandas DataFrame containing the extracted features.
    """

    message.info_start_objetive('Extracting the features ...')

    progress = Progress(
        SpinnerColumn(),
        TaskProgressColumn(),
        TextColumn('[progress.description]{task.description}'),
        BarColumn(),
        TimeElapsedColumn(),
    )

    if variants_kmers is None:
        variants_kmers = load_variants_kmers(save_path=save_path)
        message.info_founded_features(len(variants_kmers))

    file_list = [
        dir_path + '/' + name
        for name in listdir(dir_path)
        if fnmatch(name, '*.fasta')
    ]

    with progress:
        progress.add_task('[cyan]Loading sequences ...', total=None)
        feat_list = load_sequences_classify(
            file_list, word, step, dictonary, variants_kmers, False, chunk_size
        )

    data_frame = pd.DataFrame(feat_list)

    return data_frame

`extract_features_to_predict(word, step, save_path, predict_seq_path, dictonary='DNA', chunk_size=100)`

Extract features from sequences for prediction and return as a DataFrame.

This function extracts features from sequences located in the specified file for prediction purposes. The extracted features are based on the specified word length, step size, and DNA dictionary. The extracted features are returned as a pandas DataFrame.

Parameters:

Name	Type	Description	Default
`word`	`int`	The length of each k-mer.	required
`step`	`int`	The step size for moving the sliding window.	required
`save_path`	`str`	The path to save the extracted features for prediction.	required
`predict_seq_path`	`str`	The path to the file containing sequences for prediction.	required
`dictonary`	`str`	The DNA dictionary for k-mer analysis. Default is 'DNA'.	`'DNA'`
`chunk_size`	`int`	The chunk size for loading sequences. Default is 100.	`100`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: A pandas DataFrame containing the extracted features for prediction.

Source code in python/gramep/classify_utils.py

def extract_features_to_predict(
    word: int,
    step: int,
    save_path: str,
    predict_seq_path: str,
    dictonary: str = 'DNA',
    chunk_size: int = 100,
) -> pd.DataFrame:
    """
    Extract features from sequences for prediction and return as a DataFrame.

    This function extracts features from sequences located in the specified \
    file for prediction purposes.
    The extracted features are based on the specified word length, step size, \
    and DNA dictionary.
    The extracted features are returned as a pandas DataFrame.

    Args:
        word (int): The length of each k-mer.
        step (int): The step size for moving the sliding window.
        save_path (str): The path to save the extracted features for prediction.
        predict_seq_path (str): The path to the file containing sequences \
        for prediction.
        dictonary (str): The DNA dictionary for k-mer analysis. Default is 'DNA'.
        chunk_size (int, optional): The chunk size for loading sequences. \
        Default is 100.

    Returns:
        pd.DataFrame: A pandas DataFrame containing the extracted features \
        for prediction.
    """

    message.info_start_prediction()
    progress = Progress(
        SpinnerColumn(),
        TaskProgressColumn(),
        TextColumn('[progress.description]{task.description}'),
        BarColumn(),
        TimeElapsedColumn(),
    )

    variants_kmers = load_variants_kmers(save_path=save_path)

    with progress:
        progress.add_task('[cyan]Loading sequences ...', total=None)

        features = load_sequences_classify(
            [predict_seq_path],
            word,
            step,
            dictonary,
            variants_kmers,
            True,
            chunk_size,
        )

    data_frame = pd.DataFrame(features)

    return data_frame

`predict(word, step, save_path, predict_seq_path, dir_path, dictonary, load_ranges_path, load_model_path, chunk_size=100)`

Predict sequence classes using a trained model.

This function performs sequence class prediction using the specified parameters and a trained model. It extracts features from sequences in the specified file for prediction and scales them using the MinMaxScaler object loaded from the given path. The prediction results are saved as a CSV file.

Parameters:

Name	Type	Description	Default
`word`	`int`	The length of each k-mer.	required
`step`	`int`	The step size for moving the sliding window.	required
`save_path`	`str`	The path to save the processed data and predictions.	required
`predict_seq_path`	`str`	The path to the file containing sequences for prediction.	required
`dir_path`	`str`	The path to the directory containing additional files.	required
`dictonary`	`str`	The DNA dictionary for k-mer analysis.	required
`load_ranges_path`	`str`	The path to load the MinMaxScaler object ranges.	required
`load_model_path`	`str`	The path to load the trained model.	required
`chunk_size`	`int`	The chunk size for loading sequences. Default is 100.	`100`

Returns:

Name	Type	Description
`str`		A message confirming the successful prediction and saving of results.

Source code in python/gramep/classify_utils.py

def predict(
    word: int,
    step: int,
    save_path: str,
    predict_seq_path: str,
    dir_path: str,
    dictonary: str,
    load_ranges_path: str,
    load_model_path: str,
    chunk_size: int = 100,
):
    """
    Predict sequence classes using a trained model.

    This function performs sequence class prediction using the specified \
    parameters and a trained model.
    It extracts features from sequences in the specified file for prediction \
    and scales them using the
    MinMaxScaler object loaded from the given path. The prediction results are \
    saved as a CSV file.

    Args:
        word (int): The length of each k-mer.
        step (int): The step size for moving the sliding window.
        save_path (str): The path to save the processed data and predictions.
        predict_seq_path (str): The path to the file containing sequences \
        for prediction.
        dir_path (str): The path to the directory containing additional files.
        dictonary (str): The DNA dictionary for k-mer analysis.
        load_ranges_path (str): The path to load the MinMaxScaler object ranges.
        load_model_path (str): The path to load the trained model.
        chunk_size (int, optional): The chunk size for loading sequences. \
        Default is 100.

    Returns:
        str: A message confirming the successful prediction and saving of results.
    """

    data_frame = extract_features_to_predict(
        word,
        step,
        save_path,
        predict_seq_path,
        dictonary,
        chunk_size,
    )

    data_frame = process_dataframe_predict(
        data_frame, load_ranges_path=load_ranges_path
    )

    predicted_data, id_values = predict_data(
        data_frame, load_model_path=load_model_path
    )

    return save_predict_data(id_values, predicted_data, dir_path)

`predict_data(data_frame, load_model_path)`

Predict classes using a trained RandomForestClassifier model.

This function predicts classes for the provided data frame using the trained RandomForestClassifier model loaded from the specified path. It returns two pandas Series: one containing the predicted classes and the other containing the 'ID' values.

Parameters:

Name	Type	Description	Default
`data_frame`	`DataFrame`	The data frame containing features for prediction.	required
`load_model_path`	`str`	The path to load the trained RandomForestClassifier model.	required

Returns:

Type	Description
`tuple[Series, Series]`	tuple[pd.Series, pd.Series]: A tuple containing a Series of predicted classes and a Series of 'ID' values.

Source code in python/gramep/classify_utils.py

def predict_data(
    data_frame: pd.DataFrame, load_model_path: str
) -> tuple[pd.Series, pd.Series]:
    """
    Predict classes using a trained RandomForestClassifier model.

    This function predicts classes for the provided data frame using the trained
    RandomForestClassifier model loaded from the specified path. It returns two
    pandas Series: one containing the predicted classes and the other containing
    the 'ID' values.

    Args:
        data_frame (pd.DataFrame): The data frame containing features for prediction.
        load_model_path (str): The path to load the trained \
        RandomForestClassifier model.

    Returns:
        tuple[pd.Series, pd.Series]: A tuple containing a Series of predicted \
        classes and a Series of 'ID' values.
    """

    rf_classifier = load_model(load_model_path=load_model_path)

    id_values = data_frame['ID']
    x_axis = data_frame.drop(columns='ID', axis=1)

    message.info_prediction()
    return rf_classifier.predict(x_axis), id_values

`process_dataframe(data_frame, dir_path=None)`

Process a DataFrame and optionally save data and model.

This function takes a DataFrame and performs processing on it. It optionally saves the processed data and a model based on the specified flags. It returns a tuple containing the processed DataFrame and a numpy array.

Parameters:

Name	Type	Description	Default
`data_frame`	`DataFrame`	The DataFrame to be processed.	required
`dir_path`	`str`	The directory path for saving data and model. Default is None.	`None`

Returns:

Type	Description
`tuple[DataFrame, ndarray]`	tuple[pd.DataFrame, np.ndarray]: A tuple containing the processed DataFrame and a numpy array.

Source code in python/gramep/classify_utils.py

def process_dataframe(
    data_frame: pd.DataFrame,
    dir_path: str = None,
) -> tuple[pd.DataFrame, np.ndarray]:
    """
    Process a DataFrame and optionally save data and model.

    This function takes a DataFrame and performs processing on it. It optionally saves
    the processed data and a model based on the specified flags. It returns a tuple
    containing the processed DataFrame and a numpy array.

    Args:
        data_frame (pd.DataFrame): The DataFrame to be processed.
        dir_path (str, optional): The directory path for saving data and model. \
        Default is None.

    Returns:
        tuple[pd.DataFrame, np.ndarray]: A tuple containing the processed \
        DataFrame and a numpy array.
    """

    message.info_processing_dataframe()

    class_values = data_frame['CLASS']
    class_names_to_save = data_frame['CLASS']
    name_class = np.unique(class_values).tolist()

    data_frame.drop(columns=['CLASS'], axis=1, inplace=True)
    data_frame.replace([np.inf, -np.inf], 0, inplace=True)
    data_frame.replace(np.nan, 0, inplace=True)

    df = data_frame.sort_index(axis=1)
    del data_frame

    df_col_names = df.columns

    # MinMax Scaler
    minMax_scaler = MinMaxScaler()
    minMax_scaler.fit(df)
    df_minmax = minMax_scaler.transform(df)
    df = pd.DataFrame(df_minmax)
    del df_minmax
    label_encdr = LabelEncoder()
    class_values = label_encdr.fit_transform(class_values)

    df.columns = df_col_names
    df['CLASS'] = class_values

    save_data(
        data_frame=df,
        class_names_to_save=class_names_to_save,
        dir_path=dir_path,
    )

    save_ranges(ranges=minMax_scaler, dir_path=dir_path)
    return df, name_class

`process_dataframe_predict(data_frame, load_ranges_path)`

Process a DataFrame for prediction using MinMaxScaler.

This function processes the provided DataFrame for prediction using the MinMaxScaler object loaded from the specified path. It scales the data and restores column names and 'ID' values.

Parameters:

Name	Type	Description	Default
`data_frame`	`DataFrame`	The DataFrame to be processed for prediction.	required
`load_ranges_path`	`str`	The path to load the MinMaxScaler object ranges.	required

Returns:

Type	Description
`DataFrame`	pd.DataFrame: A processed DataFrame with scaled values and restored 'ID' column.

Source code in python/gramep/classify_utils.py

def process_dataframe_predict(
    data_frame: pd.DataFrame, load_ranges_path: str
) -> pd.DataFrame:
    """
    Process a DataFrame for prediction using MinMaxScaler.

    This function processes the provided DataFrame for prediction using the \
    MinMaxScaler object
    loaded from the specified path. It scales the data and restores column names \
    and 'ID' values.

    Args:
        data_frame (pd.DataFrame): The DataFrame to be processed for prediction.
        load_ranges_path (str): The path to load the MinMaxScaler object ranges.

    Returns:
        pd.DataFrame: A processed DataFrame with scaled values and restored 'ID' column.
    """

    minMax_scaler = load_ranges(load_ranges_path=load_ranges_path)

    id_values = data_frame['ID']

    data_frame.drop(columns=['ID'], axis=1, inplace=True)
    data_frame.replace([np.inf, -np.inf], 0, inplace=True)
    data_frame.replace(np.nan, 0, inplace=True)

    df = data_frame.sort_index(axis=1)
    del data_frame

    df_col_names = df.columns

    df_minmax = minMax_scaler.transform(df)
    df = pd.DataFrame(df_minmax)
    del df_minmax

    df.columns = df_col_names
    df['ID'] = id_values

    return df

`sequence_classification(data_frame, name_class, dir_path)`

Perform sequence classification based on provided data and options.

This function performs sequence classification using the provided data frame and class names. It allows for optional saving of a trained model and confusion matrix plot based on specified flags.

Parameters:

Name	Type	Description	Default
`data_frame`	`DataFrame`	The data frame containing sequence data and features.	required
`name_class`	`ndarray`	The array of class names corresponding to the data.	required
`dir_path`	`str`	The path to the directory for saving model and plot files.	required

Returns:

Type	Description
`None`	None

Source code in python/gramep/classify_utils.py

def sequence_classification(
    data_frame: pd.DataFrame,
    name_class: np.ndarray,
    dir_path: str,
) -> None:
    """
    Perform sequence classification based on provided data and options.

    This function performs sequence classification using the provided data frame \
    and class names.
    It allows for optional saving of a trained model and confusion matrix plot \
    based on specified flags.

    Args:
        data_frame (pd.DataFrame): The data frame containing sequence data and features.
        name_class (np.ndarray): The array of class names corresponding to the data.
        dir_path (str): The path to the directory for saving model and plot files.

    Returns:
        None
    """

    message.info_classifying()

    # Split the data into training and test sets
    x_axis = data_frame.drop(columns='CLASS', axis=1)
    y_axis = data_frame['CLASS']
    x_train, x_test, y_train, y_test = train_test_split(
        x_axis, y_axis, test_size=0.2, stratify=y_axis
    )

    # Create a random forest classifier with default parameters
    rf_classifier = RandomForestClassifier(n_estimators=100)
    rf_classifier.fit(x_train, y_train)

    save_model(model=rf_classifier, dir_path=dir_path)

    # Make predictions on the test set
    y_pred = rf_classifier.predict(x_test)

    # Evaluate the model - 10-fold
    cross_val = RepeatedStratifiedKFold(
        n_splits=10, n_repeats=10, random_state=7
    )
    n_scores = cross_validate(
        rf_classifier,
        x_axis,
        y_axis,
        scoring='accuracy',
        cv=cross_val,
        n_jobs=-1,
        error_score='raise',
    )
    message.result_mean_kfold(str(np.mean(n_scores['test_score'])))

    # Print the classification report
    acc = str(accuracy_score(y_test, y_pred))
    metrics = str(
        classification_report(y_test, y_pred, target_names=name_class)
    )
    message.result_accuracy(acc)
    message.result_metrics(metrics)
    save_metrics(acc=acc, metrics=metrics, dir_path=dir_path)
    del acc, metrics

    conf_mtx = confusion_matrix(y_true=y_test, y_pred=y_pred)
    vmax = max(np.unique(y_test, return_counts=True)[1])
    save_confusion_matrix(
        conf_mtx=conf_mtx,
        name_class=name_class,
        vmax=vmax,
        dir_path=dir_path,
    )
    return

Classify utils

classify_utils

message = Messages() module-attribute

classify(word, step, save_path, dir_path, get_kmers=False, reference_path=None, dictonary='DNA', chunk_size=100)

extract_features(word, step, save_path, dir_path, dictonary, variants_kmers=None, chunk_size=100)

extract_features_to_predict(word, step, save_path, predict_seq_path, dictonary='DNA', chunk_size=100)

predict(word, step, save_path, predict_seq_path, dir_path, dictonary, load_ranges_path, load_model_path, chunk_size=100)

predict_data(data_frame, load_model_path)

process_dataframe(data_frame, dir_path=None)

process_dataframe_predict(data_frame, load_ranges_path)

sequence_classification(data_frame, name_class, dir_path)

`message = Messages()` `module-attribute`

`classify(word, step, save_path, dir_path, get_kmers=False, reference_path=None, dictonary='DNA', chunk_size=100)`

`extract_features(word, step, save_path, dir_path, dictonary, variants_kmers=None, chunk_size=100)`

`extract_features_to_predict(word, step, save_path, predict_seq_path, dictonary='DNA', chunk_size=100)`

`predict(word, step, save_path, predict_seq_path, dir_path, dictonary, load_ranges_path, load_model_path, chunk_size=100)`

`predict_data(data_frame, load_model_path)`

`process_dataframe(data_frame, dir_path=None)`

`process_dataframe_predict(data_frame, load_ranges_path)`

`sequence_classification(data_frame, name_class, dir_path)`