Study index validation step.
This step reads and outputs a study index dataset with flagged studies
when target of disease validation fails.
Source code in src/gentropy/study_validation.py
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101 | class StudyValidationStep:
"""Study index validation step.
This step reads and outputs a study index dataset with flagged studies
when target of disease validation fails.
"""
def __init__(
self,
session: Session,
study_index_path: list[str],
target_index_path: str,
disease_index_path: str,
biosample_index_path: str,
valid_study_index_path: str,
invalid_study_index_path: str,
# NOTE: do not use container as default argument!
# https://github.com/satwikkansal/wtfpython?tab=readme-ov-file#-beware-of-default-mutable-arguments
invalid_qc_reasons: list[str] | None = None,
deprecated_project_ids: list[str] | None = None,
heritability_input_path: str | None = None,
) -> None:
"""Initialize step.
Args:
session (Session): Session object.
study_index_path (list[str]): Path to study index file.
target_index_path (str): Path to target index file.
disease_index_path (str): Path to disease index file.
biosample_index_path (str): Path to biosample index file.
valid_study_index_path (str): Path to write the valid records.
invalid_study_index_path (str): Path to write the output file.
invalid_qc_reasons (list[str] | None): List of invalid quality check reason names from `StudyQualityCheck` (e.g. ['DUPLICATED_STUDY']).
deprecated_project_ids (list[str] | None): List of deprecated projectIds, (e.g. ['GTEx']).
heritability_input_path (str | None): Optional path to LDSC heritability parquet output. When provided, heritability estimates are annotated onto the study index.
"""
invalid_qc_reasons = list(invalid_qc_reasons) if invalid_qc_reasons else []
deprecated_project_ids = (
list(deprecated_project_ids) if deprecated_project_ids else []
)
# Reading datasets:
target_index = TargetIndex.from_parquet(session, target_index_path)
biosample_index = BiosampleIndex.from_parquet(session, biosample_index_path)
# Reading disease index and pre-process.
# This logic does not belong anywhere, but gentropy has no disease dataset yet.
disease_index = (
session.spark.read.parquet(disease_index_path)
.select(
f.col("id").alias("diseaseId"),
f.explode_outer(
f.when(
f.col("obsoleteTerms").isNotNull(),
f.array_union(f.array("id"), f.col("obsoleteTerms")),
)
).alias("efo"),
)
.withColumn("efo", f.coalesce(f.col("efo"), f.col("diseaseId")))
)
study_index = StudyIndex.from_parquet(session, list(study_index_path))
# Running validation:
validated = (
study_index.deconvolute_studies() # Deconvolute studies where the same study is ingested from multiple sources
.validate_study_type() # Flagging non-supported study types
.validate_project_id(deprecated_project_ids) # Flag obsolete projectIds
.validate_target(target_index) # Flagging QTL studies with invalid targets
.validate_disease(disease_index) # Flagging invalid EFOs
.validate_biosample(biosample_index) # Flagging invalid biosample in QTLs
.validate_analysis_flags() # Flagging studies with case case design
)
if heritability_input_path is not None:
heritability_df = session.spark.read.parquet(heritability_input_path)
validated = validated.collect_heritability(heritability_df)
study_index_with_qc = validated.persist() # we will need this for 2 types of outputs
result = study_index_with_qc.valid_rows(invalid_qc_reasons)
(
result.valid.df.coalesce(session.output_partitions)
.write.mode(session.write_mode)
.parquet(valid_study_index_path)
)
(
result.invalid.df.coalesce(session.output_partitions)
.write.mode(session.write_mode)
.parquet(invalid_study_index_path)
)
|