Skip to content

mecfs_bio.build_system.task.rename_cols_task

Rename dataframe columns. A deliberately narrow alternative to PipeDataFrameTask: because the rename map is the whole operation, create() can update HarmonizationInfo (ref/pos columns) accurately across the rename, which a general pipe cannot.

Classes:

Functions:

RenameColsTask

Bases: Task

Methods:

Attributes:

backend class-attribute instance-attribute 

backend: ValidBackend = 'polars'

deps property 

deps: list[Task]

meta instance-attribute 

meta: Meta

out_format class-attribute instance-attribute 

out_format: OutFormat = ParquetOutFormat()

renames instance-attribute 

renames: Mapping[str, str]

source_task instance-attribute 

source_task: Task

create classmethod 

create(
    source_task: Task,
    asset_id: str,
    renames: Mapping[str, str],
    out_format: OutFormat = ParquetOutFormat(),
    backend: ValidBackend = "polars",
) -> RenameColsTask
Source code in mecfs_bio/build_system/task/rename_cols_task.py 
 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
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,
    renames: Mapping[str, str],
    out_format: OutFormat = ParquetOutFormat(),
    backend: ValidBackend = "polars",
) -> "RenameColsTask":
    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, HarmonizableReferenceTableMeta):
        meta = HarmonizableReferenceTableMeta(
            group=source_meta.group,
            sub_group=source_meta.sub_group,
            sub_folder=PurePath("processed"),
            extension=extension,
            id=AssetId(asset_id),
            filename=None,
            read_spec=read_spec,
            harmonization_info=(
                remap_harmonization_info(source_meta.harmonization_info, renames)
                if source_meta.harmonization_info is not None
                else None
            ),
        )
    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,
            harmonization_info=(
                remap_harmonization_info(source_meta.harmonization_info, renames)
                if source_meta.harmonization_info is not None
                else None
            ),
        )
    elif isinstance(source_meta, ReferenceFileMeta):
        meta = ReferenceFileMeta(
            group=source_meta.group,
            sub_group=source_meta.sub_group,
            sub_folder=PurePath("processed"),
            extension=extension,
            id=AssetId(asset_id),
            read_spec=read_spec,
        )
    else:
        raise ValueError(
            f"RenameColsTask: unsupported source meta {type(source_meta).__name__}"
        )
    return cls(
        meta=meta,
        source_task=source_task,
        renames=dict(renames),
        out_format=out_format,
        backend=backend,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/rename_cols_task.py 
63
64
65
66
67
68
69
70
71
72
73
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    source = fetch(self.source_task.asset_id)
    frame: narwhals.LazyFrame = scan_dataframe_asset(
        asset=source, meta=self.source_task.meta, parquet_backend=self.backend
    )
    frame = frame.rename(dict(self.renames))
    out_path = scratch_dir / _OUT_FILENAME
    write_df_according_to_format(
        df=frame, out_path=out_path, out_format=self.out_format
    )
    return FileAsset(out_path)

remap_harmonization_info 

remap_harmonization_info(
    info: HarmonizationInfo, renames: Mapping[str, str]
) -> HarmonizationInfo
Source code in mecfs_bio/build_system/task/rename_cols_task.py 
40
41
42
43
44
45
46
47
48
def remap_harmonization_info(
    info: HarmonizationInfo, renames: Mapping[str, str]
) -> HarmonizationInfo:
    new_ref = renames.get(info.ref_allele_col, info.ref_allele_col)
    new_pos = renames.get(info.pos_col, info.pos_col)
    assert new_ref != new_pos, (
        f"rename collides the ref-allele column and position column onto {new_ref!r}"
    )
    return HarmonizationInfo(build=info.build, ref_allele_col=new_ref, pos_col=new_pos)