Skip to content

mecfs_bio.build_system.task.magma.magma_snp_location_task

Classes:

MagmaSNPFileTask

Bases: Task

Methods:

Attributes:

deps property 

deps: list[Task]

extra_columns_to_output instance-attribute 

extra_columns_to_output: list[str]

gwas_parquet_with_rsid_task instance-attribute 

gwas_parquet_with_rsid_task: Task

meta property 

meta: Meta

pipes instance-attribute 

pipes: Sequence[DataProcessingPipe]

source_id property 

source_id: AssetId

source_meta property 

source_meta: Meta

create_for_magma_snp_p_value_file classmethod 

create_for_magma_snp_p_value_file(
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
    pipes: list[DataProcessingPipe] | None = None,
)
Source code in mecfs_bio/build_system/task/magma/magma_snp_location_task.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@classmethod
def create_for_magma_snp_p_value_file(
    cls,
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
    pipes: list[DataProcessingPipe] | None = None,
):
    if pipes is None:
        pipes = []
    pipes.append(ComputePPipe())
    extra_cols = [GWASLAB_P_COL]
    source_meta = gwas_parquet_with_rsids_task.meta
    meta = create_new_meta(
        source_meta,
        asset_id=asset_id,
        format=DataFrameTextFormat(
            separator=" ",
            has_header=False,
            column_names=[GWASLAB_RSID_COL] + extra_cols,
        ),
        extension=".id.p.txt",
    )
    return cls(
        meta=meta,
        gwas_parquet_with_rsid_task=gwas_parquet_with_rsids_task,
        extra_columns_to_output=extra_cols,
        pipes=pipes,
    )

create_for_magma_snp_p_value_file_compute_if_needed classmethod 

create_for_magma_snp_p_value_file_compute_if_needed(
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
    pipes: list[DataProcessingPipe] | None = None,
)
Source code in mecfs_bio/build_system/task/magma/magma_snp_location_task.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@classmethod
def create_for_magma_snp_p_value_file_compute_if_needed(
    cls,
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
    pipes: list[DataProcessingPipe] | None = None,
):
    if pipes is None:
        pipes = []
    pipes.append(ComputePIfNeededPipe())
    extra_cols = [GWASLAB_P_COL]
    source_meta = gwas_parquet_with_rsids_task.meta
    meta = create_new_meta(
        source_meta,
        asset_id=asset_id,
        format=DataFrameTextFormat(
            separator=" ",
            has_header=False,
            column_names=[GWASLAB_RSID_COL] + extra_cols,
        ),
        extension=".id.p.txt",
    )
    return cls(
        meta=meta,
        gwas_parquet_with_rsid_task=gwas_parquet_with_rsids_task,
        extra_columns_to_output=extra_cols,
        pipes=pipes,
    )

create_for_magma_snp_p_value_file_precomputed_p classmethod 

create_for_magma_snp_p_value_file_precomputed_p(
    gwas_parquet_with_rsids_task: Task, asset_id: str
)

As above, but assume the data already contains p values.

Source code in mecfs_bio/build_system/task/magma/magma_snp_location_task.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@classmethod
def create_for_magma_snp_p_value_file_precomputed_p(
    cls,
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
):
    """
    As above, but assume the data already contains p values.
    """
    extra_cols = [GWASLAB_P_COL]
    source_meta = gwas_parquet_with_rsids_task.meta
    meta = create_new_meta(
        source_meta,
        asset_id=asset_id,
        format=DataFrameTextFormat(
            separator=" ",
            has_header=False,
            column_names=[GWASLAB_RSID_COL] + extra_cols,
        ),
        extension=".id.p.txt",
    )
    return cls(
        meta=meta,
        gwas_parquet_with_rsid_task=gwas_parquet_with_rsids_task,
        extra_columns_to_output=extra_cols,
        pipes=[],
    )

create_for_magma_snp_pos_file classmethod 

create_for_magma_snp_pos_file(
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
    pipes: list[DataProcessingPipe] | None = None,
)
Source code in mecfs_bio/build_system/task/magma/magma_snp_location_task.py 
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
@classmethod
def create_for_magma_snp_pos_file(
    cls,
    gwas_parquet_with_rsids_task: Task,
    asset_id: str,
    pipes: list[DataProcessingPipe] | None = None,
):
    if pipes is None:
        pipes = []
    extra_cols = [GWASLAB_CHROM_COL, GWASLAB_POS_COL]
    source_meta = gwas_parquet_with_rsids_task.meta
    meta = create_new_meta(
        source_meta,
        asset_id=asset_id,
        format=DataFrameTextFormat(
            separator=" ",
            has_header=False,
            column_names=[GWASLAB_RSID_COL] + extra_cols,
        ),
        extension=".id.chr.genpos.txt",
    )
    return cls(
        meta=meta,
        gwas_parquet_with_rsid_task=gwas_parquet_with_rsids_task,
        extra_columns_to_output=extra_cols,
        pipes=pipes,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/magma/magma_snp_location_task.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    source_asset = fetch(self.source_id)
    gwas_data = scan_dataframe_asset(source_asset, meta=self.source_meta)
    for pipe in self.pipes:
        gwas_data = pipe.process(gwas_data)
    gwas_data = gwas_data.unique(
        subset=GWASLAB_RSID_COL, order_by=[GWASLAB_CHROM_COL, GWASLAB_POS_COL]
    )
    out_path = scratch_dir / "output"
    gwas_data.sort([GWASLAB_CHROM_COL, GWASLAB_POS_COL]).select(
        GWASLAB_RSID_COL, *self.extra_columns_to_output
    ).collect().to_polars().write_csv(out_path, include_header=False, separator=" ")
    return FileAsset(
        out_path,
    )