Skip to content

mecfs_bio.build_system.task.protein_coloc.ppp_window_fetcher

Fetch dense protein summary statistics for a few locus windows, then discard the download.

The DenseWindowFetcher protocol is what the colocalization Task depends on, so any proteomics source (UKB-PPP, CSF pQTL, ...) can be plugged in. Fetched frames are in the common window schema (WINDOW_SCHEMA, positions in GRCh38) but NOT yet oriented to the reference genome: EA/NEA are whatever the source reports, and orientation happens downstream.

PppWindowFetcher downloads a protein's UKB-PPP tar from Synapse once per call, reads only the chromosome files the windows need (one at a time, only the needed columns), filters each window, and deletes the tar and the extracted files before returning. In PPP's regenie files ALLELE1 is the effect allele and A1FREQ its frequency; ALLELE0 is the other allele.

Classes:

DenseWindowFetcher

Bases: Protocol

Methods:

  • fetch –

    Return one WINDOW_SCHEMA frame per window, keyed by locus_id.

fetch 

fetch(
    protein_id: str,
    windows: Sequence[LocusWindow],
    scratch_dir: Path,
    wf: WF,
) -> dict[str, pl.DataFrame]

Return one WINDOW_SCHEMA frame per window, keyed by locus_id.

Source code in mecfs_bio/build_system/task/protein_coloc/ppp_window_fetcher.py 
79
80
81
82
83
84
85
86
87
def fetch(
    self,
    protein_id: str,
    windows: Sequence[LocusWindow],
    scratch_dir: Path,
    wf: WF,
) -> dict[str, pl.DataFrame]:
    """Return one WINDOW_SCHEMA frame per window, keyed by locus_id."""
    ...

LocusWindow

A closed GRCh38 interval [start, end] on chromosome chrom (X coded as 23).

Attributes:

chrom instance-attribute 

chrom: int

end instance-attribute 

end: int

locus_id instance-attribute 

locus_id: str

start instance-attribute 

start: int

PppWindowFetcher

DenseWindowFetcher for UKB-PPP.

proteins: the manifest rows of every protein that may be requested, looked up by OID.

Methods:

Attributes:

proteins instance-attribute 

proteins: tuple[PppProteinFile, ...]

fetch 

fetch(
    protein_id: str,
    windows: Sequence[LocusWindow],
    scratch_dir: Path,
    wf: WF,
) -> dict[str, pl.DataFrame]
Source code in mecfs_bio/build_system/task/protein_coloc/ppp_window_fetcher.py 
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
def fetch(
    self,
    protein_id: str,
    windows: Sequence[LocusWindow],
    scratch_dir: Path,
    wf: WF,
) -> dict[str, pl.DataFrame]:
    protein = self._protein(protein_id)
    download_dir = scratch_dir / f"dense_{protein_id}"
    download_dir.mkdir(parents=True, exist_ok=True)
    try:
        tar_path = wf.download_from_synapse(
            protein.synid, download_dir, protein.tar_filename
        )
        extracted_dir = extract_protein_tar(tar_path, download_dir)
        tar_path.unlink()
        frames: dict[str, pl.DataFrame] = {}
        for chrom in sorted({w.chrom for w in windows}):
            rows = _read_chromosome(_chromosome_file(extracted_dir, chrom))
            for window in windows:
                if window.chrom == chrom:
                    frames[window.locus_id] = _slice_window(rows, window)
        return frames
    finally:
        shutil.rmtree(download_dir, ignore_errors=True)