Skip to content

mecfs_bio.build_system.task.sldsc.prepare_sldsc_baseline_task

Adapter Task: convert ldsc baseline LD scores (chromosome-split, @ pattern) and regression-weight LD scores into a prepared S-LDSC baseline (see prepared_sldsc_reference). Zero-variance baseline columns are dropped here, as ldsc's _check_variance does on the full reference, and the baseline is restricted to SNPs that have a weight LD score (ldsc inner-joins the weights; the HapMap3 weights omit the MHC region).

Classes:

PrepareSLDSCBaselineTask

Bases: Task

baseline_pattern: path inside the baseline directory with @ standing for the chromosome, e.g. baseline_v1.2/baseline.@ weights_pattern: same, for the regression-weight LD scores (exactly one LD column).

Methods:

Attributes:

baseline_directory_task instance-attribute 

baseline_directory_task: Task

baseline_pattern instance-attribute 

baseline_pattern: str

deps property 

deps: list[Task]

meta instance-attribute 

meta: Meta

weights_directory_task instance-attribute 

weights_directory_task: Task

weights_pattern instance-attribute 

weights_pattern: str

create classmethod 

create(
    asset_id: str,
    baseline_directory_task: Task,
    baseline_pattern: str,
    weights_directory_task: Task,
    weights_pattern: str,
) -> PrepareSLDSCBaselineTask
Source code in mecfs_bio/build_system/task/sldsc/prepare_sldsc_baseline_task.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@classmethod
def create(
    cls,
    asset_id: str,
    baseline_directory_task: Task,
    baseline_pattern: str,
    weights_directory_task: Task,
    weights_pattern: str,
) -> "PrepareSLDSCBaselineTask":
    source_meta = baseline_directory_task.meta
    assert isinstance(source_meta, ReferenceDataDirectoryMeta)
    return cls(
        meta=ReferenceDataDirectoryMeta(
            group=source_meta.group,
            sub_group=source_meta.sub_group,
            sub_folder=PREPARED_SLDSC_SUB_FOLDER,
            id=AssetId(asset_id),
        ),
        baseline_directory_task=baseline_directory_task,
        baseline_pattern=baseline_pattern,
        weights_directory_task=weights_directory_task,
        weights_pattern=weights_pattern,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/sldsc/prepare_sldsc_baseline_task.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    baseline_dir = _fetch_directory(fetch, self.baseline_directory_task.asset_id)
    weights_dir = _fetch_directory(fetch, self.weights_directory_task.asset_id)
    baseline = _restrict_to_weighted_snps(
        ld=read_ld_score_sources(
            root=baseline_dir,
            prefixes=[self.baseline_pattern],
            drop_zero_variance=True,
        ),
        w_ld=_read_weight_ld(root=weights_dir, pattern=self.weights_pattern),
    )
    out_dir = scratch_dir / "prepared_baseline"
    write_prepared_baseline(out_dir, baseline)
    return DirectoryAsset(out_dir)