Skip to content

mecfs_bio.build_system.task.pipe_dataframe_task

Task to transform a dataframe asset using a DataProcessingPipes.

Classes:

Functions:

Attributes:

OutFormat module-attribute 

OutFormat = ParquetOutFormat | CSVOutFormat

CSVOutFormat

Attributes:

sep instance-attribute 

sep: str

ParquetOutFormat

PipeDataFrameTask

Bases: Task

Methods:

Attributes:

backend class-attribute instance-attribute 

backend: ValidBackend = 'ibis'

deps property 

deps: list[Task]

meta property 

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 
 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
137
138
139
140
141
142
143
144
145
146
147
@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 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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)
    if isinstance(self.out_format, CSVOutFormat):
        df.collect().to_pandas().to_csv(
            out_path, index=False, sep=self.out_format.sep
        )
    elif isinstance(self.out_format, ParquetOutFormat):
        df.sink_parquet(out_path)
    return FileAsset(out_path)

get_extension_and_read_spec_from_format 

get_extension_and_read_spec_from_format(
    out_format: OutFormat,
) -> tuple[str, DataFrameReadSpec]
Source code in mecfs_bio/build_system/task/pipe_dataframe_task.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_extension_and_read_spec_from_format(
    out_format: OutFormat,
) -> tuple[str, DataFrameReadSpec]:
    if isinstance(out_format, CSVOutFormat):
        read_spec = DataFrameReadSpec(DataFrameTextFormat(separator=out_format.sep))
        if out_format.sep == "\t":
            extension = ".tsv"
        elif out_format.sep == ",":
            extension = ".csv"
        else:
            raise ValueError("Unknown sep")
    elif isinstance(out_format, ParquetOutFormat):
        read_spec = DataFrameReadSpec(DataFrameParquetFormat())
        extension = ".parquet"
    else:
        raise ValueError(f"Unknown format {out_format}")
    return extension, read_spec