Skip to content

mecfs_bio.build_system.task.gwaslab.gwaslab_util

Classes:

Functions:

Attributes:

ChecksumCalculator module-attribute 

ChecksumCalculator = Callable[[Path], str]

RefPathLookup module-attribute 

RefPathLookup = Callable[[str], Path | None]

logger module-attribute 

logger = structlog.get_logger()

RefDownloader

Bases: Protocol

Variant

Represents a genetic variant.

Attributes:

chromosome instance-attribute 

chromosome: int

effect_allele instance-attribute 

effect_allele: str

id property 

id: str

id_normalized property 

id_normalized: str

non_effect_allele instance-attribute 

non_effect_allele: str

position instance-attribute 

position: int

df_to_variants 

df_to_variants(df: DataFrame) -> list[Variant]
Source code in mecfs_bio/build_system/task/gwaslab/gwaslab_util.py 
53
54
55
56
57
58
59
60
61
62
def df_to_variants(df: pd.DataFrame) -> list[Variant]:
    return [
        Variant(
            chromosome=df[GWASLAB_CHROM_COL].iloc[i],
            position=df[GWASLAB_POS_COL].iloc[i],
            effect_allele=df[GWASLAB_EFFECT_ALLELE_COL].iloc[i],
            non_effect_allele=df[GWASLAB_NON_EFFECT_ALLELE_COL].iloc[i],
        )
        for i in range(len(df))
    ]

expected_reference_md5 

expected_reference_md5(ref: str) -> str | None

The checksum gwaslab records for a reference, or None if it records none.

gwaslab leaves this blank for many entries (the genome fastas, the GTFs), so a None here means unverifiable, not invalid.

Source code in mecfs_bio/build_system/task/gwaslab/gwaslab_util.py 
72
73
74
75
76
77
78
79
80
def expected_reference_md5(ref: str) -> str | None:
    """The checksum gwaslab records for a reference, or None if it records none.

    gwaslab leaves this blank for many entries (the genome fastas, the GTFs), so a
    None here means unverifiable, not invalid.
    """
    catalogue = _reference_catalogue()
    assert ref in catalogue, f"unknown gwaslab reference {ref}"
    return catalogue[ref].get(_REFERENCE_MD5_KEY) or None

gwaslab_download_ref_if_missing 

gwaslab_download_ref_if_missing(
    ref: str,
    path_lookup: RefPathLookup = _local_reference_path,
    downloader: RefDownloader = _download_reference,
    checksum: ChecksumCalculator = calc_md5_checksum,
) -> Path

Return the local path to a gwaslab reference, downloading it when absent and re-downloading it when its checksum does not match the one gwaslab records.

Source code in mecfs_bio/build_system/task/gwaslab/gwaslab_util.py 
115
116
117
118
119
120
121
122
123
124
125
126
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
154
155
156
157
158
159
160
161
def gwaslab_download_ref_if_missing(
    ref: str,
    path_lookup: RefPathLookup = _local_reference_path,
    downloader: RefDownloader = _download_reference,
    checksum: ChecksumCalculator = calc_md5_checksum,
) -> Path:
    """Return the local path to a gwaslab reference, downloading it when absent and
    re-downloading it when its checksum does not match the one gwaslab records.
    """
    expected_md5 = expected_reference_md5(ref)

    local_path = _usable_reference(ref, expected_md5, path_lookup, checksum)
    if local_path is not None:
        return local_path

    # NOTE: Gwaslab stores paths of downloaded files in a "config" object stored within the gwaslab package
    # This object can grow out of date, meaning that the file exists but is not referenced in the config
    # Besides actually downloading missing files, calling "download ref" can also sometimes update the config to point to
    # existing local files that are not in the config
    downloader(ref, overwrite=False)
    local_path = _usable_reference(
        ref, expected_md5=expected_md5, path_lookup=path_lookup, checksum=checksum
    )
    if local_path is not None:
        return local_path

    # A superseded copy under the name gwaslab downloads to blocks the fetch entirely:
    # gwaslab skips the download because the name exists, then refuses to register the
    # file because it fails the recorded checksum. Only an overwrite gets past that.
    logger.warning(
        "gwaslab has no reference matching its recorded checksum, re-downloading",
        ref=ref,
    )
    downloader(ref, overwrite=True)

    refreshed_path = path_lookup(ref)
    assert refreshed_path is not None, (
        f"gwaslab failed to download reference {ref}, even with overwrite"
    )
    if expected_md5 is None:
        return refreshed_path
    actual_md5 = checksum(refreshed_path)
    assert actual_md5 == expected_md5, (
        f"gwaslab reference {ref} at {refreshed_path} has md5 {actual_md5}, but "
        f"gwaslab records {expected_md5}, even after a fresh download"
    )
    return refreshed_path