read_in_experiment_data

Read in experiment (hops) data from a qbed file. The qbed file may be plain text or gzipped and may or may not have column headers. If the column headers are present, they must be in the following order: chr, start, end, strand, depth. If the column headers are not present, the columns must be in same order and number. Datatypes are checked but will not be coerced – errors are raised if they do not match the expected datatypes. the chr column is relabeled from the curr_chr_name_convention to the new_chr_name_convention using the chrmap_df.

:param experiment_data_path: Path to the qbed file, plain text or gzipped, with or without column headers :type experiment_data_path: str :param curr_chr_name_convention: The current chromosome name convention. :type curr_chr_name_convention: str :param new_chr_name_convention: The new chromosome name convention. :type new_chr_name_convention: str :param chrmap_df: The chrmap dataframe. :type chrmap_df: pd.DataFrame :param deduplicate: Whether to deduplicate the experiment data based on chr, start, end such that if an insertion occurs at the same coordinate but on opposite strands, only one record is retained. :type deduplicate: bool

:return: The experiment data as a dataframe with the chr column refactored to the new_chr_name_convention :rtype: pd.DataFrame

:raises ValueError: If the experiment_data_path does not exist or is not a file; if the column headers exist but do not match expectation or if the datatypes do not match expectation.

:Example:

import pandas as pd import os import tempfile tmp_qbed = tempfile.NamedTemporaryFile(suffix=’.qbed’).name with open(tmp_qbed, ‘w’) as f: … _ = f.write(‘chr\tstart\tend\tstrand\tdepth\n’) … _ = f.write(‘chr1\t1\t2\t+\t1\n’)

create a temporary chrmap file

tmp_chrmap = tempfile.NamedTemporaryFile(suffix=’.csv’).name chrmap_df = pd.DataFrame({‘curr_chr_name_convention’: … [‘chr1’, ‘chr2’, ‘chr3’], … ‘new_chr_name_convention’: … [‘chrI’, ‘chrII’, ‘chrIII’]})

read in the data

experiment_df, experiment_total_hops = read_in_experiment_data( … tmp_qbed, … ‘curr_chr_name_convention’, … ‘new_chr_name_convention’, … chrmap_df) list(experiment_df.columns) == [‘chr’, ‘start’, ‘end’, ‘depth’, … ‘strand’] True experiment_total_hops 1

Source code in callingcardstools/PeakCalling/yeast/read_in_data.py

def read_in_experiment_data(
    experiment_data_path: str,
    curr_chr_name_convention: pd.DataFrame,
    new_chr_name_convention: pd.DataFrame,
    chrmap_df: str,
    deduplicate: bool = True
) -> pd.DataFrame:
    """
    Read in experiment (hops) data from a qbed file. The qbed file may be
    plain text or gzipped and may or may not have column headers. If the
    column headers are present, they must be in the following order:
    `chr`, `start`, `end`, `strand`, `depth`. If the column headers are
    not present, the columns must be in same order and number. Datatypes
    are checked but will not be coerced -- errors are raised if they do not
    match the expected datatypes. the `chr` column is relabeled from the
    `curr_chr_name_convention` to the `new_chr_name_convention` using the
    `chrmap_df`.

    :param experiment_data_path: Path to the qbed file, plain text or gzipped,
        with or without column headers
    :type experiment_data_path: str
    :param curr_chr_name_convention: The current chromosome name convention.
    :type curr_chr_name_convention: str
    :param new_chr_name_convention: The new chromosome name convention.
    :type new_chr_name_convention: str
    :param chrmap_df: The chrmap dataframe.
    :type chrmap_df: pd.DataFrame
    :param deduplicate: Whether to deduplicate the experiment data based on
        `chr`, `start`, `end` such that if an insertion occurs at the same
        coordinate but on opposite strands, only one record is retained.
    :type deduplicate: bool

    :return: The experiment data as a dataframe with the `chr` column
        refactored to the `new_chr_name_convention`
    :rtype: pd.DataFrame

    :raises ValueError: If the `experiment_data_path` does not exist or
        is not a file; if the column headers exist but do not match expectation
        or if the datatypes do not match expectation.

    :Example:

    >>> import pandas as pd
    >>> import os
    >>> import tempfile
    >>> tmp_qbed = tempfile.NamedTemporaryFile(suffix='.qbed').name
    >>> with open(tmp_qbed, 'w') as f:
    ...    _ = f.write('chr\\tstart\\tend\\tstrand\\tdepth\\n')
    ...    _ = f.write('chr1\\t1\\t2\\t+\\t1\\n')
    >>> # create a temporary chrmap file
    >>> tmp_chrmap = tempfile.NamedTemporaryFile(suffix='.csv').name
    >>> chrmap_df = pd.DataFrame({'curr_chr_name_convention':
    ...                            ['chr1', 'chr2', 'chr3'],
    ...                           'new_chr_name_convention':
    ...                            ['chrI', 'chrII', 'chrIII']})
    >>> # read in the data
    >>> experiment_df, experiment_total_hops = read_in_experiment_data(
    ...   tmp_qbed,
    ...   'curr_chr_name_convention',
    ...   'new_chr_name_convention',
    ...    chrmap_df)
    >>> list(experiment_df.columns) == ['chr', 'start', 'end', 'depth',
    ...                                 'strand']
    True
    >>> experiment_total_hops
    1
    """
    # check input
    if not os.path.exists(experiment_data_path):
        raise ValueError("experiment_data_path must exist")
    if not os.path.isfile(experiment_data_path):
        raise ValueError("experiment_data_path must be a file")

    # check if data is gzipped
    gzipped = str(experiment_data_path).endswith(".gz")
    # check if data has column headers
    header = pd.read_csv(
        experiment_data_path, sep="\t", compression="gzip" if gzipped else None, nrows=0
    )
    if header.columns.tolist() != ["chr", "start", "end", "depth", "strand"]:
        header = None
    else:
        header = 0
    # read in data
    try:
        experiment_df = pd.read_csv(
            experiment_data_path,
            sep="\t",
            header=header,
            names=["chr", "start", "end", "depth", "strand"],
            dtype={"chr": str, "start": int, "end": int, "depth": int, "strand": str},
            compression="gzip" if gzipped else None,
        )
    except ValueError as e:
        raise ValueError(
            "experiment_data_path must be a qbed file "
            "with columns `chr`, `start`, `end`, `strand`, "
            "and `depth`"
        ) from e

    # relabel chr column
    experiment_df = relabel_chr_column(
        experiment_df, chrmap_df, curr_chr_name_convention, new_chr_name_convention
    )

    if deduplicate:
        # deduplicate based on `chr`, `start`, `end` such that if an insertion occurs
        # at the same coordinate but on opposite strands, only one record is retained
        experiment_df.drop_duplicates(subset=["chr", "start", "end"], inplace=True)
        # set strand to '*'
        experiment_df["strand"] = "*"

    return qbed_df_to_pyranges(experiment_df), len(experiment_df)