Skip to content

mecfs_bio.build_system.task.pipe_dataframe_task

Task to transform a dataframe asset using a DataProcessingPipes.

Classes:

PipeDataFrameTask

Bases: Task

Methods:

Attributes:

backend class-attribute instance-attribute 

backend: ValidBackend = 'ibis'

deps property 

deps: list[Task]

meta instance-attribute 

meta: Meta

out_format instance-attribute 

out_format: OutFormat

pipes instance-attribute 

pipes: Sequence[DataProcessingPipe]

source_data_task instance-attribute 

source_data_task: Task

create classmethod 

create(
    source_task: Task,
    asset_id: str,
    out_format: OutFormat,
    pipes: Sequence[DataProcessingPipe],
    backend: ValidBackend = "ibis",
) -> PipeDataFrameTask
Source code in mecfs_bio/build_system/task/pipe_dataframe_task.py 
 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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@classmethod
def create(
    cls,
    source_task: Task,
    asset_id: str,
    out_format: OutFormat,
    pipes: Sequence[DataProcessingPipe],
    backend: ValidBackend = "ibis",
) -> "PipeDataFrameTask":
    source_meta = source_task.meta
    extension, read_spec = get_extension_and_read_spec_from_format(
        out_format=out_format
    )
    meta: Meta
    if isinstance(source_meta, ReferenceFileMeta):
        meta = ReferenceFileMeta(
            group=source_meta.group,
            sub_group=source_meta.sub_group,
            sub_folder=PurePath("processed"),
            id=AssetId(asset_id),
            extension=extension,
            read_spec=read_spec,
        )
    elif isinstance(source_meta, GWASSummaryDataFileMeta):
        meta = FilteredGWASDataMeta(
            id=AssetId(asset_id),
            trait=source_meta.trait,
            project=source_meta.project,
            sub_dir=PurePath("processed"),
            read_spec=read_spec,
        )
    elif isinstance(source_meta, ResultTableMeta):
        meta = ResultTableMeta(
            id=AssetId(asset_id),
            trait=source_meta.trait,
            project=source_meta.project,
            extension=extension,
            read_spec=read_spec,
        )
    elif isinstance(source_meta, FilteredGWASDataMeta):
        meta = FilteredGWASDataMeta(
            id=AssetId(asset_id),
            trait=source_meta.trait,
            project=source_meta.project,
            sub_dir=source_meta.sub_dir,
            read_spec=read_spec,
        )
    else:
        raise ValueError("unknown source meta")

    return cls(
        source_data_task=source_task,
        pipes=list(pipes),
        meta=meta,
        out_format=out_format,
        backend=backend,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/pipe_dataframe_task.py 
67
68
69
70
71
72
73
74
75
76
77
78
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    asset = fetch(self._source_id)
    df = scan_dataframe_asset(
        asset=asset, meta=self._source_meta, parquet_backend=self.backend
    )
    out_path = scratch_dir / "out_dataframe"
    for pipe in self.pipes:
        df = pipe.process(df)
    write_df_according_to_format(
        df=df, out_path=out_path, out_format=self.out_format
    )
    return FileAsset(out_path)