Colocalisation step.
This workflow runs colocalisation analyses that assess the degree to which independent signals of the association share the same causal variant in a region of the genome, typically limited by linkage disequilibrium (LD).
Source code in src/gentropy/colocalisation.py
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
85
86
87
88
89
90
91
92 | class ColocalisationStep:
"""Colocalisation step.
This workflow runs colocalisation analyses that assess the degree to which independent signals of the association share the same causal variant in a region of the genome, typically limited by linkage disequilibrium (LD).
"""
def __init__(
self,
session: Session,
credible_set_path: str,
study_index_path: str,
coloc_path: str,
colocalisation_method: str,
) -> None:
"""Run Colocalisation step.
Args:
session (Session): Session object.
credible_set_path (str): Input credible sets path.
study_index_path (str): Input study index path.
coloc_path (str): Output Colocalisation path.
colocalisation_method (str): Colocalisation method.
"""
colocalisation_class = self._get_colocalisation_class(colocalisation_method)
# Extract
credible_set = (
StudyLocus.from_parquet(
session, credible_set_path, recursiveFileLookup=True
).filter(col("finemappingMethod") == "SuSie")
if colocalisation_class is Coloc
else StudyLocus.from_parquet(
session, credible_set_path, recursiveFileLookup=True
)
)
si = StudyIndex.from_parquet(
session, study_index_path, recursiveFileLookup=True
)
# Transform
overlaps = credible_set.filter_credible_set(
CredibleInterval.IS95
).find_overlaps(si)
colocalisation_results = colocalisation_class.colocalise(overlaps) # type: ignore
# Load
colocalisation_results.df.write.mode(session.write_mode).parquet(
f"{coloc_path}/{colocalisation_method.lower()}"
)
@classmethod
def _get_colocalisation_class(cls: type[ColocalisationStep], method: str) -> type:
"""Get colocalisation class.
Args:
method (str): Colocalisation method.
Returns:
type: Colocalisation class.
Raises:
ValueError: if method not available.
Examples:
>>> ColocalisationStep._get_colocalisation_class("ECaviar")
<class 'gentropy.method.colocalisation.ECaviar'>
"""
module_name = "gentropy.method.colocalisation"
module = import_module(module_name)
available_methods = []
for class_name, class_obj in inspect.getmembers(module, inspect.isclass):
if class_obj.__module__ == module_name:
available_methods.append(class_name)
if class_name == method:
return class_obj
raise ValueError(
f"Method {method} is not supported. Available: {(', ').join(available_methods)}"
)
|
__init__(session: Session, credible_set_path: str, study_index_path: str, coloc_path: str, colocalisation_method: str) -> None
Run Colocalisation step.
Parameters:
| Name |
Type |
Description |
Default |
session |
Session
|
|
required
|
credible_set_path |
str
|
Input credible sets path.
|
required
|
study_index_path |
str
|
|
required
|
coloc_path |
str
|
Output Colocalisation path.
|
required
|
colocalisation_method |
str
|
|
required
|
Source code in src/gentropy/colocalisation.py
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 | def __init__(
self,
session: Session,
credible_set_path: str,
study_index_path: str,
coloc_path: str,
colocalisation_method: str,
) -> None:
"""Run Colocalisation step.
Args:
session (Session): Session object.
credible_set_path (str): Input credible sets path.
study_index_path (str): Input study index path.
coloc_path (str): Output Colocalisation path.
colocalisation_method (str): Colocalisation method.
"""
colocalisation_class = self._get_colocalisation_class(colocalisation_method)
# Extract
credible_set = (
StudyLocus.from_parquet(
session, credible_set_path, recursiveFileLookup=True
).filter(col("finemappingMethod") == "SuSie")
if colocalisation_class is Coloc
else StudyLocus.from_parquet(
session, credible_set_path, recursiveFileLookup=True
)
)
si = StudyIndex.from_parquet(
session, study_index_path, recursiveFileLookup=True
)
# Transform
overlaps = credible_set.filter_credible_set(
CredibleInterval.IS95
).find_overlaps(si)
colocalisation_results = colocalisation_class.colocalise(overlaps) # type: ignore
# Load
colocalisation_results.df.write.mode(session.write_mode).parquet(
f"{coloc_path}/{colocalisation_method.lower()}"
)
|