read_in_background_data

Read in background (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. NOTE: unlike the experiment df, there is no option to deduplicate as the background file is expected to be the combination of multiple experiments at this point.

:param background_data_path: Path to the background data qbed file, plain text or gzipped, with or without column headers :type background_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 :return: The background data. :rtype: pd.DataFrame

:raises ValueError: If the background_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

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

read in the data

background_df, background_total_hops = read_in_background_data( … tmp_qbed, … ‘curr_chr_name_convention’, … ‘new_chr_name_convention’, … chrmap_df) list(background_df.columns) == [‘chr’, ‘start’, ‘end’, ‘depth’, … ‘strand’] True background_total_hops 1

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

def read_in_background_data(
    background_data_path: str,
    curr_chr_name_convention: pd.DataFrame,
    new_chr_name_convention: pd.DataFrame,
    chrmap_df: str,
) -> pd.DataFrame:
    """
    Read in background (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`. NOTE: unlike the experiment df, there is no option to deduplicate
    as the background file is expected to be the combination of multiple experiments
    at this point.

    :param background_data_path: Path to the background data qbed file, plain
        text or gzipped, with or without column headers
    :type background_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
    :return: The background data.
    :rtype: pd.DataFrame

    :raises ValueError: If the `background_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
    >>> chrmap_df = pd.DataFrame({'curr_chr_name_convention':
    ...                            ['chr1', 'chr2', 'chr3'],
    ...                           'new_chr_name_convention':
    ...                            ['chrI', 'chrII', 'chrIII']})
    >>> # read in the data
    >>> background_df, background_total_hops = read_in_background_data(
    ...   tmp_qbed,
    ...   'curr_chr_name_convention',
    ...   'new_chr_name_convention',
    ...    chrmap_df)
    >>> list(background_df.columns) == ['chr', 'start', 'end', 'depth',
    ...                                  'strand']
    True
    >>> background_total_hops
    1
    """
    # check input
    if not os.path.exists(background_data_path):
        raise ValueError("background_data_path must exist")
    if not os.path.isfile(background_data_path):
        raise ValueError("background_data_path must be a file")

    # check if data is gzipped
    gzipped = str(background_data_path).endswith(".gz")
    # check if data has column headers
    header = pd.read_csv(background_data_path, sep="\t", nrows=0)
    if header.columns.tolist() != ["chr", "start", "end", "depth", "strand"]:
        header = None
    else:
        header = 0

    # read in data
    try:
        background_df = pd.read_csv(
            background_data_path,
            sep="\t",
            header=header,
            names=["chr", "start", "end", "depth", "strand"],
            dtype={
                "chr": str,
                "start": int,
                "end": int,
                "depth": "int64",
                "strand": str,
            },
            compression="gzip" if gzipped else None,
        )
    except ValueError as e:
        raise ValueError(
            "background_data_path must be a qbed file "
            "with columns `chr`, `start`, `end`, `depth`, "
            "and `strand`"
        ) from e

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

    return qbed_df_to_pyranges(background_df), len(background_df)