Skip to content

mecfs_bio.build_system.task.magma.magma_annotate_task

Classes:

Attributes:

logger module-attribute 

logger = get_logger()

MagmaAnnotateTask

Bases: Task

Perform the annotate step of the MAGMA pipeline. This step associates SNPs with genes. See page 5 of the MAGMA manual here: https://vu.data.surfsara.nl/s/MUiv3y1SFRePnyG?dir=/&editing=false&openfile=true

The window option is used to optionally associate SNPs outside the transcription region of a gene with the gene.

Methods:

Attributes:

deps property 

deps: list[Task]

gene_loc_file_task instance-attribute 

gene_loc_file_task: Task

gene_loc_id property 

gene_loc_id: AssetId

magma_binary_id property 

magma_binary_id: AssetId

magma_binary_task instance-attribute 

magma_binary_task: Task

meta property 

meta: Meta

snp_loc_file_task instance-attribute 

snp_loc_file_task: Task

snp_loc_id property 

snp_loc_id: AssetId

window class-attribute instance-attribute 

window: tuple[int, int] | None = None

create classmethod 

create(
    asset_id: str,
    magma_binary_task: Task,
    snp_loc_file_task: Task,
    gene_loc_file_task: Task,
    window: tuple[int, int] | None = None,
) -> MagmaAnnotateTask
Source code in mecfs_bio/build_system/task/magma/magma_annotate_task.py 
 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
@classmethod
def create(
    cls,
    asset_id: str,
    magma_binary_task: Task,
    snp_loc_file_task: Task,
    gene_loc_file_task: Task,
    window: tuple[int, int] | None = None,
) -> "MagmaAnnotateTask":
    snp_loc_meta = snp_loc_file_task.meta
    assert isinstance(snp_loc_meta, FilteredGWASDataMeta)
    meta = FilteredGWASDataMeta(
        id=AssetId(asset_id),
        trait=snp_loc_meta.trait,
        project=snp_loc_meta.project,
        sub_dir=PurePath(snp_loc_meta.sub_dir) / "magma",
        extension=".genes.annot",
    )
    return cls(
        meta=meta,
        magma_binary_task=magma_binary_task,
        snp_loc_file_task=snp_loc_file_task,
        gene_loc_file_task=gene_loc_file_task,
        window=window,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/magma/magma_annotate_task.py 
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 execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    binary_asset = fetch(self.magma_binary_id)
    snp_loc_asset = fetch(self.snp_loc_id)
    gene_loc_asset = fetch(self.gene_loc_id)
    assert isinstance(binary_asset, FileAsset)
    assert isinstance(snp_loc_asset, FileAsset)
    assert isinstance(gene_loc_asset, FileAsset)
    binary_path = binary_asset.path
    snp_loc_path = snp_loc_asset.path
    gene_loc_path = gene_loc_asset.path
    out_base_path = scratch_dir / "out_base"
    cmd = [
        str(binary_path),
        "--annotate",
    ]
    if self.window is not None:
        cmd.extend([f"window={self.window[0]},{self.window[1]}"])
    cmd += [
        "--snp-loc",
        str(snp_loc_path),
        "--gene-loc",
        str(gene_loc_path),
        "--out",
        str(out_base_path),
    ]
    logger.debug(f"Running command: {' '.join(cmd)}")
    result = execute_command(cmd)
    logger.debug(f"Command produced result: \n\n{result}\n\n\n")
    out_full_path = Path(str(out_base_path) + ".genes.annot")
    return FileAsset(out_full_path)