Skip to content

Annotation parsers

AnnotationMatrixParser

A generic annotation matrix parser class.

Source code in magenpy/parsers/annotation_parsers.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class AnnotationMatrixParser:
    """
    A generic annotation matrix parser class.
    """

    def __init__(self, col_name_converter=None, **read_csv_kwargs):
        """
        :param col_name_converter: A dictionary mapping column names
        in the original table to magenpy's column names for the various
        SNP features in the annotation matrix.
        :param read_csv_kwargs: Keyword arguments to pass to pandas' `read_csv`.
        """

        self.col_name_converter = col_name_converter
        self.read_csv_kwargs = read_csv_kwargs

        # If the delimiter is not specified, assume whitespace by default:
        if (
            "sep" not in self.read_csv_kwargs
            and "delimiter" not in self.read_csv_kwargs
        ):
            self.read_csv_kwargs["sep"] = "\t"

    def _get_read_csv_kwargs(self):
        """Return read options with dtypes for stable variant metadata."""

        kwargs = self.read_csv_kwargs.copy()
        user_dtype = kwargs.get("dtype")
        if user_dtype is not None and not isinstance(user_dtype, Mapping):
            return kwargs

        dtype = {"SNP": str, "POS": np.float64}
        for raw_column, output_column in (self.col_name_converter or {}).items():
            if output_column in dtype:
                dtype[raw_column] = dtype[output_column]
        if isinstance(user_dtype, Mapping):
            dtype.update(user_dtype)
        for column in kwargs.get("converters") or {}:
            dtype.pop(column, None)
        kwargs["dtype"] = dtype
        return kwargs

    def parse(self, annotation_file, drop_na=True):
        """
        Parse the annotation matrix file
        :param annotation_file: The path to the annotation file.
        :param drop_na: Drop any entries with missing values.
        """

        try:
            df = pd.read_csv(annotation_file, **self._get_read_csv_kwargs())
        except Exception as e:
            logger.error(
                "Failed to parse the annotation file(s):\n%s",
                e.stderr.decode(),
            )
            raise e

        if drop_na:
            df = df.dropna()

        if self.col_name_converter is not None:
            df.rename(columns=self.col_name_converter, inplace=True)

        if "POS" in df:
            if df["POS"].isna().any():
                df["POS"] = df["POS"].astype("Int32")
            else:
                df["POS"] = df["POS"].astype(np.int32)

        df.sort_values(["CHR", "POS"], inplace=True)

        annotations = [ann for ann in df.columns if ann not in ("CHR", "SNP", "POS")]

        return df, annotations

__init__(col_name_converter=None, **read_csv_kwargs)

Parameters:

Name Type Description Default
col_name_converter

A dictionary mapping column names in the original table to magenpy's column names for the various SNP features in the annotation matrix.

 None
read_csv_kwargs

Keyword arguments to pass to pandas' read_csv.

 {}
Source code in magenpy/parsers/annotation_parsers.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, col_name_converter=None, **read_csv_kwargs):
    """
    :param col_name_converter: A dictionary mapping column names
    in the original table to magenpy's column names for the various
    SNP features in the annotation matrix.
    :param read_csv_kwargs: Keyword arguments to pass to pandas' `read_csv`.
    """

    self.col_name_converter = col_name_converter
    self.read_csv_kwargs = read_csv_kwargs

    # If the delimiter is not specified, assume whitespace by default:
    if (
        "sep" not in self.read_csv_kwargs
        and "delimiter" not in self.read_csv_kwargs
    ):
        self.read_csv_kwargs["sep"] = "\t"

parse(annotation_file, drop_na=True)

Parse the annotation matrix file

Parameters:

Name Type Description Default
annotation_file

The path to the annotation file.

 required
drop_na

Drop any entries with missing values.

 True
Source code in magenpy/parsers/annotation_parsers.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def parse(self, annotation_file, drop_na=True):
    """
    Parse the annotation matrix file
    :param annotation_file: The path to the annotation file.
    :param drop_na: Drop any entries with missing values.
    """

    try:
        df = pd.read_csv(annotation_file, **self._get_read_csv_kwargs())
    except Exception as e:
        logger.error(
            "Failed to parse the annotation file(s):\n%s",
            e.stderr.decode(),
        )
        raise e

    if drop_na:
        df = df.dropna()

    if self.col_name_converter is not None:
        df.rename(columns=self.col_name_converter, inplace=True)

    if "POS" in df:
        if df["POS"].isna().any():
            df["POS"] = df["POS"].astype("Int32")
        else:
            df["POS"] = df["POS"].astype(np.int32)

    df.sort_values(["CHR", "POS"], inplace=True)

    annotations = [ann for ann in df.columns if ann not in ("CHR", "SNP", "POS")]

    return df, annotations

LDSCAnnotationMatrixParser

Bases: AnnotationMatrixParser

Source code in magenpy/parsers/annotation_parsers.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class LDSCAnnotationMatrixParser(AnnotationMatrixParser):
    def __init__(self, col_name_converter=None, **read_csv_kwargs):
        """
        :param col_name_converter: A dictionary mapping column names
        in the original table to magenpy's column names for the various
        SNP features in the annotation matrix.
        :param read_csv_kwargs: Keyword arguments to pass to pandas' read_csv
        """

        super().__init__(col_name_converter, **read_csv_kwargs)
        self.col_name_converter = self.col_name_converter or {}
        self.col_name_converter.update({"BP": "POS"})

    def parse(self, annotation_file, drop_na=True):
        """
        Parse the annotation matrix file
        :param annotation_file: The path to the annotation file.
        :param drop_na: Drop any entries with missing values.
        """

        df, annotations = super().parse(annotation_file, drop_na=drop_na)

        df = df.drop(["CM", "base"], axis=1)
        annotations = [ann for ann in annotations if ann not in ("CM", "base")]

        return df, annotations

__init__(col_name_converter=None, **read_csv_kwargs)

Parameters:

Name Type Description Default
col_name_converter

A dictionary mapping column names in the original table to magenpy's column names for the various SNP features in the annotation matrix.

 None
read_csv_kwargs

Keyword arguments to pass to pandas' read_csv

 {}
Source code in magenpy/parsers/annotation_parsers.py 
88
89
90
91
92
93
94
95
96
97
98
def __init__(self, col_name_converter=None, **read_csv_kwargs):
    """
    :param col_name_converter: A dictionary mapping column names
    in the original table to magenpy's column names for the various
    SNP features in the annotation matrix.
    :param read_csv_kwargs: Keyword arguments to pass to pandas' read_csv
    """

    super().__init__(col_name_converter, **read_csv_kwargs)
    self.col_name_converter = self.col_name_converter or {}
    self.col_name_converter.update({"BP": "POS"})

parse(annotation_file, drop_na=True)

Parse the annotation matrix file

Parameters:

Name Type Description Default
annotation_file

The path to the annotation file.

 required
drop_na

Drop any entries with missing values.

 True
Source code in magenpy/parsers/annotation_parsers.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def parse(self, annotation_file, drop_na=True):
    """
    Parse the annotation matrix file
    :param annotation_file: The path to the annotation file.
    :param drop_na: Drop any entries with missing values.
    """

    df, annotations = super().parse(annotation_file, drop_na=drop_na)

    df = df.drop(["CM", "base"], axis=1)
    annotations = [ann for ann in annotations if ann not in ("CM", "base")]

    return df, annotations

parse_annotation_bed_file(annot_bed_file)

Parse an annotation bed file in the format specified by Ensemble: https://uswest.ensembl.org/info/website/upload/bed.html

The file contains 3-12 columns, starting with Chromosome, start_coordinate, end_coordinate, etc. After reading the raw file, we let pandas infer whether the file has a header or not and we standardize the names of the first 3 columns and convert the chromosome column into an integer.

Parameters:

Name Type Description Default
annot_bed_file str

The path to the annotation BED file.

 required
Source code in magenpy/parsers/annotation_parsers.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def parse_annotation_bed_file(annot_bed_file):
    """
    Parse an annotation bed file in the format specified by Ensemble:
    https://uswest.ensembl.org/info/website/upload/bed.html

    The file contains 3-12 columns, starting with Chromosome, start_coordinate, end_coordinate, etc.
    After reading the raw file, we let pandas infer whether the file has a header or not and we
    standardize the names of the first 3 columns and convert the chromosome column into an integer.

    :param annot_bed_file: The path to the annotation BED file.
    :type annot_bed_file: str
    """

    try:
        annot_bed = pd.read_csv(
            annot_bed_file,
            usecols=[0, 1, 2],
            sep="\t",
            names=["CHR", "Start", "End"],
            dtype={"CHR": str, "Start": np.int32, "End": np.int32},
        )
    except Exception as e:
        logger.error(
            "Failed to parse the BED file:\n%s",
            e.stderr.decode(),
        )
        raise e

    annot_bed["CHR"] = annot_bed["CHR"].str.replace("chr", "").astype(int)

    return annot_bed