Skip to content

mecfs_bio.build_system.task.plot_mr_effect_measure_task

Create an effect measure plot using zepid

Classes:

EffectMeasurePlotConfig

Attributes:

effect_size_col instance-attribute 

effect_size_col: str

effect_size_label instance-attribute 

effect_size_label: str

figsize class-attribute instance-attribute 

figsize: tuple[float, float] = (7, 3)

ref_line_center instance-attribute 

ref_line_center: float

se_col instance-attribute 

se_col: str

t_adjust instance-attribute 

t_adjust: float

title class-attribute instance-attribute 

title: str | None = None

y_label instance-attribute 

y_label: str

y_label_col instance-attribute 

y_label_col: str

PlotMREffectMeasure

Bases: Task

Task to create an effect measure plot Useful for example to show the results of Mendelian Randomization

See: https://zepid.readthedocs.io/en/latest/Reference/Graphics.html?highlight=effectmeasureplot

Methods:

Attributes:

config instance-attribute 

config: EffectMeasurePlotConfig

deps property 

deps: list[Task]

meta property 

meta: Meta

pre_pipe class-attribute instance-attribute 

pre_pipe: DataProcessingPipe = IdentityPipe()

source_df_task instance-attribute 

source_df_task: Task

source_id property 

source_id: AssetId

source_meta property 

source_meta: Meta

create classmethod 

create(
    asset_id: str,
    source_df_task: Task,
    config: EffectMeasurePlotConfig,
    pre_pipe: DataProcessingPipe = IdentityPipe(),
)
Source code in mecfs_bio/build_system/task/plot_mr_effect_measure_task.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@classmethod
def create(
    cls,
    asset_id: str,
    source_df_task: Task,
    config: EffectMeasurePlotConfig,
    pre_pipe: DataProcessingPipe = IdentityPipe(),
):
    source_meta = source_df_task.meta
    assert isinstance(source_meta, ResultTableMeta)
    meta = GWASPlotDirectoryMeta(
        trait=source_meta.trait,
        project=source_meta.project,
        id=AssetId(asset_id),
    )
    return cls(
        source_df_task=source_df_task,
        config=config,
        meta=meta,
        pre_pipe=pre_pipe,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/plot_mr_effect_measure_task.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    source_asset = fetch(self.source_id)
    df = (
        self.pre_pipe.process(
            scan_dataframe_asset(source_asset, meta=self.source_meta)
        )
        .collect()
        .to_pandas()
    )
    df = df.sort_values(by=self.config.effect_size_col)
    mean = df[self.config.effect_size_col]
    se = df[self.config.se_col]
    lower = mean - 1.95 * se
    upper = mean + 1.95 * se
    p = EffectMeasurePlot(
        label=df[self.config.y_label_col].tolist(),
        effect_measure=mean.tolist(),
        lcl=lower.tolist(),
        ucl=upper.tolist(),
    )
    p.labels(
        center=self.config.ref_line_center,
        effectmeasure=self.config.effect_size_label,
    )
    ax = p.plot(
        figsize=self.config.figsize,
        t_adjuster=self.config.t_adjust,
    )
    if self.config.title is not None:
        plt.title(self.config.title)
    ax.spines["top"].set_visible(False)
    ax.spines["right"].set_visible(False)
    ax.spines["bottom"].set_visible(True)
    ax.spines["left"].set_visible(False)
    plt.savefig(scratch_dir / "effect_measure_plot.png", bbox_inches="tight")
    return DirectoryAsset(scratch_dir)