Skip to content

mecfs_bio.build_system.task.ppp_database.hapmap3_membership_task

Normalized HapMap3 membership list for the PPP variant index (HapMap3 mode).

Reads the HapMap3 hg38 snplist bundled inside the gwaslab package -- guaranteed present because gwaslab is a pinned pyproject.toml dependency, so no download or rehosting is needed -- and normalizes it to the membership contract (CHR, POS[hg38], EA, NEA, rsID). Allele orientation here is arbitrary (the index adopts the template protein's orientation); only the allele set is used for matching.

Classes:

  • Hapmap3MembershipTask –

    Produce the normalized HapMap3 membership list (CHR, POS, EA, NEA, rsID) from

Functions:

Attributes:

MEMBERSHIP_COLUMNS module-attribute 

MEMBERSHIP_COLUMNS = [
    GWASLAB_CHROM_COL,
    GWASLAB_POS_COL,
    GWASLAB_EFFECT_ALLELE_COL,
    GWASLAB_NON_EFFECT_ALLELE_COL,
    GWASLAB_RSID_COL,
]

Hapmap3MembershipTask

Bases: GeneratingTask

Produce the normalized HapMap3 membership list (CHR, POS, EA, NEA, rsID) from the gwaslab-bundled hg38 snplist.

Methods:

Attributes:

deps property 

deps: list[Task]

meta instance-attribute 

meta: ReferenceFileMeta

create classmethod 

create(asset_id: str) -> Hapmap3MembershipTask
Source code in mecfs_bio/build_system/task/ppp_database/hapmap3_membership_task.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@classmethod
def create(cls, asset_id: str) -> "Hapmap3MembershipTask":
    return cls(
        meta=ReferenceFileMeta(
            group="ppp_variant_index_membership",
            sub_group="hapmap3",
            sub_folder=PurePath("processed"),
            id=AssetId(asset_id),
            filename="hapmap3_membership",
            extension=".parquet",
            read_spec=DataFrameReadSpec(DataFrameParquetFormat()),
        )
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/ppp_database/hapmap3_membership_task.py 
92
93
94
95
96
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    membership = normalize_hapmap3_snplist(bundled_hapmap3_hg38_snplist_path())
    out_path = scratch_dir / "hapmap3_membership.parquet"
    write_byte_stream_split_parquet(membership, out_path, float_columns=[])
    return FileAsset(out_path)

bundled_hapmap3_hg38_snplist_path 

bundled_hapmap3_hg38_snplist_path() -> Path

Path to the HapMap3 hg38 snplist bundled in the installed gwaslab package.

Source code in mecfs_bio/build_system/task/ppp_database/hapmap3_membership_task.py 
59
60
61
def bundled_hapmap3_hg38_snplist_path() -> Path:
    """Path to the HapMap3 hg38 snplist bundled in the installed gwaslab package."""
    return Path(os.path.dirname(gl.__file__)) / GWASLAB_HAPMAP3_HG38_SNPLIST_RELPATH

normalize_hapmap3_snplist 

normalize_hapmap3_snplist(
    snplist_path: Path,
) -> pl.DataFrame

Read the gwaslab HapMap3 snplist and normalize to the membership contract.

Source code in mecfs_bio/build_system/task/ppp_database/hapmap3_membership_task.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def normalize_hapmap3_snplist(snplist_path: Path) -> pl.DataFrame:
    """Read the gwaslab HapMap3 snplist and normalize to the membership contract."""
    return (
        pl.read_csv(snplist_path, separator="\t")
        .select(
            pl.col(_HM3_CHROM_COL).cast(pl.Int32).alias(GWASLAB_CHROM_COL),
            pl.col(_HM3_POS_COL).cast(pl.Int32).alias(GWASLAB_POS_COL),
            pl.col(_HM3_A1_COL).alias(GWASLAB_EFFECT_ALLELE_COL),
            pl.col(_HM3_A2_COL).alias(GWASLAB_NON_EFFECT_ALLELE_COL),
            pl.col(_HM3_RSID_COL).alias(GWASLAB_RSID_COL),
        )
        .select(MEMBERSHIP_COLUMNS)
    )