Skip to content

mecfs_bio.build_system.task.mixer.mixer_bivariate_results

Classes:

Attributes:

BIVARIATE_OUTPUT_PREFIX module-attribute 

BIVARIATE_OUTPUT_PREFIX = 'bivariate_results'

MixerBivariateSummarizeResultsTask

Bases: Task

Task to take the results of a Bivarirate MiXeR run and produce tables and plots.

Methods:

Attributes:

combine_task instance-attribute 

combine_task: MixerBivariateCombine

deps property 

deps: list[Task]

meta property 

meta: Meta

override_trait_1_name class-attribute instance-attribute 

override_trait_1_name: str | None = None

override_trait_2_name class-attribute instance-attribute 

override_trait_2_name: str | None = None

trait_1_name property 

trait_1_name: str

trait_2_name property 

trait_2_name: str

create classmethod 

create(
    asset_id: str,
    combine_task: MixerBivariateCombine,
    override_trait_1_name: str | None = None,
    override_trait_2_name: str | None = None,
)
Source code in mecfs_bio/build_system/task/mixer/mixer_bivariate_results.py 
 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
@classmethod
def create(
    cls,
    asset_id: str,
    combine_task: MixerBivariateCombine,
    override_trait_1_name: str | None = None,
    override_trait_2_name: str | None = None,
):
    source_meta = combine_task.meta
    meta: Meta
    if isinstance(source_meta, ResultDirectoryMeta):
        meta = ResultDirectoryMeta(
            trait=source_meta.trait,
            project=source_meta.project,
            id=AssetId(asset_id),
            sub_dir=PurePath("analysis") / "mixer_results",
        )
    elif isinstance(source_meta, SimpleDirectoryMeta):
        meta = SimpleDirectoryMeta(
            id=AssetId(asset_id),
        )
    else:
        raise ValueError(f"Unknown meta {source_meta}")
    return cls(
        meta=meta,
        combine_task=combine_task,
        override_trait_1_name=override_trait_1_name,
        override_trait_2_name=override_trait_2_name,
    )

execute 

execute(scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset
Source code in mecfs_bio/build_system/task/mixer/mixer_bivariate_results.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def execute(self, scratch_dir: Path, fetch: Fetch, wf: WF) -> Asset:
    source_asset = fetch(self.combine_task.asset_id)
    assert isinstance(source_asset, DirectoryAsset)

    plots_mounts = {
        scratch_dir.resolve(): _CONTAINER_PLOT_DIR,
        source_asset.path.resolve(): _CONTAINER_COMBINED_DIR,
    }
    trait_1_name = self.override_trait_1_name or self.trait_1_name
    trait_2_name = self.override_trait_2_name or self.trait_2_name
    invoke_mixer_figures(
        args=[
            "two",
            "--json-test",
            str(
                _CONTAINER_COMBINED_DIR
                / (BIVARIATE_COMBINED_TEST_FILENAME_PREFIX + ".json")
            ),
            "--json-fit",
            str(
                _CONTAINER_COMBINED_DIR
                / (BIVARIATE_COMBINED_FIT_FILENAME_PREFIX + ".json")
            ),
            "--out",
            str(_CONTAINER_PLOT_DIR / (BIVARIATE_OUTPUT_PREFIX)),
            "--statistic",
            "mean std",
            "--ext",
            "png",
            "--trait1",
            _normalize_name(trait_1_name),
            "--trait2",
            _normalize_name(trait_2_name),
        ],
        extra_mounts=plots_mounts,
    )
    return DirectoryAsset(scratch_dir)