Skip to content

Colocalisation

gentropy.dataset.colocalisation.Colocalisation dataclass

Bases: Dataset

Colocalisation results for pairs of overlapping study-locus.

Source code in src/gentropy/dataset/colocalisation.py 
 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
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass
class Colocalisation(Dataset):
    """Colocalisation results for pairs of overlapping study-locus."""

    @classmethod
    def get_schema(cls: type[Colocalisation]) -> StructType:
        """Provides the schema for the Colocalisation dataset.

        Returns:
            StructType: Schema for the Colocalisation dataset
        """
        return parse_spark_schema("colocalisation.json")

    def append_study_metadata(
        self: Colocalisation,
        study_locus: StudyLocus,
        study_index: StudyIndex,
        *,
        metadata_cols: list[str],
        colocalisation_side: str = "right",
    ) -> DataFrame:
        """Appends metadata from the study to the requested side of the colocalisation dataset.

        Args:
            study_locus (StudyLocus): Dataset containing study loci that links the colocalisation dataset and the study index via the studyId
            study_index (StudyIndex): Dataset containing study index that contains the metadata
            metadata_cols (list[str]): List of study columns to append
            colocalisation_side (str): Which side of the colocalisation dataset to append metadata to. Must be either 'right' or 'left'

        Returns:
            DataFrame: Colocalisation dataset with appended metadata of the study from the requested side

        Raises:
            ValueError: if colocalisation_side is not 'right' or 'left'
        """
        metadata_cols = ["studyId", *metadata_cols]
        if colocalisation_side not in ["right", "left"]:
            raise ValueError(
                f"colocalisation_side must be either 'right' or 'left', got {colocalisation_side}"
            )

        study_loci_w_metadata = (
            study_locus.df.select("studyLocusId", "studyId")
            .join(
                f.broadcast(study_index.df.select("studyId", *metadata_cols)),
                "studyId",
            )
            .distinct()
        )
        coloc_df = (
            # drop `rightStudyType` in case it is requested
            self.df.drop("rightStudyType")
            if "studyType" in metadata_cols and colocalisation_side == "right"
            else self.df
        )
        return (
            # Append that to the respective side of the colocalisation dataset
            study_loci_w_metadata.selectExpr(
                f"studyLocusId as {colocalisation_side}StudyLocusId",
                *[
                    f"{col} as {colocalisation_side}{col[0].upper() + col[1:]}"
                    for col in metadata_cols
                ],
            ).join(coloc_df, f"{colocalisation_side}StudyLocusId", "right")
        )

    def drop_trans_effects(
        self: Colocalisation, study_locus: StudyLocus
    ) -> Colocalisation:
        """Filters the colocalisation dataset to only include cis effects from QTLs (right study locus).

        Args:
            study_locus (StudyLocus): Dataset containing study loci that has metadata about the type of credible set

        Returns:
            Colocalisation: Colocalisation dataset filtered to only include cis effects from QTLs (right study locus)
        """
        cis_study_loci = study_locus.filter(
            (~f.col("isTransQtl")) | (f.col("isTransQtl").isNull())
        ).df.select("studyLocusId")
        filtered_coloc = self.df.join(
            cis_study_loci,
            self.df.rightStudyLocusId == cis_study_loci.studyLocusId,
            "inner",
        ).drop("studyLocusId")
        return Colocalisation(
            _df=filtered_coloc,
            _schema=self.get_schema(),
        )

append_study_metadata(study_locus: StudyLocus, study_index: StudyIndex, *, metadata_cols: list[str], colocalisation_side: str = 'right') -> DataFrame

Appends metadata from the study to the requested side of the colocalisation dataset.

Parameters:

Name Type Description Default
study_locus StudyLocus

Dataset containing study loci that links the colocalisation dataset and the study index via the studyId

 required
study_index StudyIndex

Dataset containing study index that contains the metadata

 required
metadata_cols list[str]

List of study columns to append

 required
colocalisation_side str

Which side of the colocalisation dataset to append metadata to. Must be either 'right' or 'left'

 'right'

Returns:

Name Type Description
DataFrame DataFrame

Colocalisation dataset with appended metadata of the study from the requested side

Raises:

Type Description
ValueError

if colocalisation_side is not 'right' or 'left'
Source code in src/gentropy/dataset/colocalisation.py 
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
85
86
def append_study_metadata(
    self: Colocalisation,
    study_locus: StudyLocus,
    study_index: StudyIndex,
    *,
    metadata_cols: list[str],
    colocalisation_side: str = "right",
) -> DataFrame:
    """Appends metadata from the study to the requested side of the colocalisation dataset.

    Args:
        study_locus (StudyLocus): Dataset containing study loci that links the colocalisation dataset and the study index via the studyId
        study_index (StudyIndex): Dataset containing study index that contains the metadata
        metadata_cols (list[str]): List of study columns to append
        colocalisation_side (str): Which side of the colocalisation dataset to append metadata to. Must be either 'right' or 'left'

    Returns:
        DataFrame: Colocalisation dataset with appended metadata of the study from the requested side

    Raises:
        ValueError: if colocalisation_side is not 'right' or 'left'
    """
    metadata_cols = ["studyId", *metadata_cols]
    if colocalisation_side not in ["right", "left"]:
        raise ValueError(
            f"colocalisation_side must be either 'right' or 'left', got {colocalisation_side}"
        )

    study_loci_w_metadata = (
        study_locus.df.select("studyLocusId", "studyId")
        .join(
            f.broadcast(study_index.df.select("studyId", *metadata_cols)),
            "studyId",
        )
        .distinct()
    )
    coloc_df = (
        # drop `rightStudyType` in case it is requested
        self.df.drop("rightStudyType")
        if "studyType" in metadata_cols and colocalisation_side == "right"
        else self.df
    )
    return (
        # Append that to the respective side of the colocalisation dataset
        study_loci_w_metadata.selectExpr(
            f"studyLocusId as {colocalisation_side}StudyLocusId",
            *[
                f"{col} as {colocalisation_side}{col[0].upper() + col[1:]}"
                for col in metadata_cols
            ],
        ).join(coloc_df, f"{colocalisation_side}StudyLocusId", "right")
    )

drop_trans_effects(study_locus: StudyLocus) -> Colocalisation

Filters the colocalisation dataset to only include cis effects from QTLs (right study locus).

Parameters:

Name Type Description Default
study_locus StudyLocus

Dataset containing study loci that has metadata about the type of credible set

 required

Returns:

Name Type Description
Colocalisation Colocalisation

Colocalisation dataset filtered to only include cis effects from QTLs (right study locus)
Source code in src/gentropy/dataset/colocalisation.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def drop_trans_effects(
    self: Colocalisation, study_locus: StudyLocus
) -> Colocalisation:
    """Filters the colocalisation dataset to only include cis effects from QTLs (right study locus).

    Args:
        study_locus (StudyLocus): Dataset containing study loci that has metadata about the type of credible set

    Returns:
        Colocalisation: Colocalisation dataset filtered to only include cis effects from QTLs (right study locus)
    """
    cis_study_loci = study_locus.filter(
        (~f.col("isTransQtl")) | (f.col("isTransQtl").isNull())
    ).df.select("studyLocusId")
    filtered_coloc = self.df.join(
        cis_study_loci,
        self.df.rightStudyLocusId == cis_study_loci.studyLocusId,
        "inner",
    ).drop("studyLocusId")
    return Colocalisation(
        _df=filtered_coloc,
        _schema=self.get_schema(),
    )

get_schema() -> StructType classmethod

Provides the schema for the Colocalisation dataset.

Returns:

Name Type Description
StructType StructType

Schema for the Colocalisation dataset
Source code in src/gentropy/dataset/colocalisation.py 
26
27
28
29
30
31
32
33
@classmethod
def get_schema(cls: type[Colocalisation]) -> StructType:
    """Provides the schema for the Colocalisation dataset.

    Returns:
        StructType: Schema for the Colocalisation dataset
    """
    return parse_spark_schema("colocalisation.json")

Schema

root
 |-- leftStudyLocusId: string (nullable = false)
 |-- rightStudyLocusId: string (nullable = false)
 |-- rightStudyType: string (nullable = false)
 |-- chromosome: string (nullable = false)
 |-- colocalisationMethod: string (nullable = false)
 |-- numberColocalisingVariants: long (nullable = false)
 |-- h0: double (nullable = true)
 |-- h1: double (nullable = true)
 |-- h2: double (nullable = true)
 |-- h3: double (nullable = true)
 |-- h4: double (nullable = true)
 |-- clpp: double (nullable = true)
 |-- betaRatioSignAverage: double (nullable = true)
  • 2023-01-20
  • 2024-01-17
  • Contributors
    • d0choa
    • web-flow