Skip to content

Commit bc8e0af

Browse files
ghar1821LuLeom
ghar1821
and
LuLeom
authored
Setup run hpc (#119)
* add scripts to process raw dataset * editing config to set apptainer cache dir * editing pre-run scripts and trying to fix R methods not running. * add h5py to setup * reverting changes to setup * separate submit scripts * finally the first setting that works!!!! * update config and settings for control methods * adjusted resources for metrics and methods * update cytovi to use A30 gpu * add numba cache dir export to allow jit caching * update cytovi implementation * force recompute for all cytonorm * add temp dir resolution for hpc * remove transpose from harmonypy * adding support for hpc * update temp dir again * latest config file that works reasonably well with hpc * add some job submit scripts for SLURM * update tmp_path for cytonorm * redirect numba cache dir away from /tmp and to its own folder. * update batch adjust non control samples naming * fix bug in perfect integration subsetting * fix bug where we can't replace the batch column if it is not integer * fix bug where the donor loc are somewhat mismatched.. * update ratio inconsistent peak where corrected data return only zero * Update script.py * update scripts * remove average batch r2 global * add seed setting for cytovi * remove env for viash temp files * update lisi to allow anndata write * update cycombine * more updates to cycombine * minor change of script type * update cytonorm * fixed gaussnorm * fixed limma * Fixed harmonypy and combat * Fixed rPCA * update batchadjust and add copy to subset * remove cytovi and some obsolete metrics * renamed shuffle control methods * missed label change * reorganising scripts for hpc * update changelog * update changelog again * update changelog * update changelog * update description. * manually adding some dependencies for flowCore and flowStats * update ratio inconsistent peaks * update inconsistent peaks * add print statements to subset functions * add print statements when writing files out * add utility scripts for pulling intermediate files * update methods and metrics labels * fix bug where subsetting was not done on ilisi and fsom mapping metrics * Update CHANGELOG.md --------- Co-authored-by: Luqui12 <luca.leomazzi@gmail.com>
1 parent 37b439b commit bc8e0af

77 files changed

Lines changed: 1524 additions & 411 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎CHANGELOG.md‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,12 @@
6767

6868
* Added new metric `ratio_inconsistent_peaks` (PR #114).
6969

70+
* Added processing scripts for Lille dataset and remove ones for CLL dataset (PR #118).
71+
72+
* Added config and run scripts for running the benchmark on WEHI HPC (PR #119).
73+
74+
* Added utility scripts to pull intermediate files (PR #119).
75+
7076
## MAJOR CHANGES
7177

7278
* Updated file schema (PR #18):
@@ -103,6 +109,24 @@
103109

104110
* Update CytoVI (PR #114).
105111

112+
* Update CytoVI to normalise using minmax scaler fitted on batch 1 post correction (PR #119).
113+
114+
* Update batchadjust, cytonorm to use HPC temp dir if the environment variable is set or else
115+
default to what is set by viash. This is to prevent collision in temp files when the jobs are running (PR #119).
116+
117+
* Update ratio inconsistent peaks to handle edge cases where methods return only zero values
118+
for a marker/cell type/donor combination, causing sd to be zero and division by zero (PR #119).
119+
120+
* One control and no control method will only get either samples from one control plus non-control samples or just no control samples.
121+
They will no longer be given access to other samples to correct or to train the model.
122+
Notably, the included control samples may still be corrected (PR #119).
123+
124+
* Change temp folder for methods which rely on writing out FCS files.
125+
Temp folders are now created by a new helper function which will create a subdirectory under `meta[["temp_dir"]]`.
126+
This will be used as the temp directory (PR #119).
127+
128+
* Change inconsistent peaks metrics to consistent peaks (PR #119).
129+
106130
## MINOR CHANGES
107131

108132
* Enabled unit tests (PR #2).
@@ -134,6 +158,11 @@
134158

135159
* Removed EMD max from calculation (PR #113).
136160

161+
* Tune the resource requirement for each method (PR #119).
162+
* Low time, mem, cpu for control methods.
163+
* Mid time, mem, cpu for most methods, except below.
164+
* High (or very high) time, mem, cpu for computationally expensive methods like rPCA.
165+
137166

138167
## BUG FIXES
139168

@@ -170,3 +199,17 @@
170199
* Fix bug in EMD vertical where sample combination was malformed (PR #113)
171200

172201
* Fix lisi inconsistent naming (PR #117) for issue #116.
202+
203+
* Fix bug in perfect integration where if batch is str (not int), it only returns control samples (PR #119).
204+
205+
* Fix bug in batchadjust needing "Batch_" in the sample names for non-control samples (PR #119).
206+
207+
* Fix bug in cytonorm to mid where recompute was set to FALSE. It is now set to TRUE (PR #119).
208+
209+
* Remove transpose in harmonypy as new updates to harmonypy no longer need the transpose (PR #119).
210+
211+
* Fix bug in get_obs_var_for_integrated to handle the cases where batch column in obs is str
212+
and thus can't be directly overriden (new values given by get_donor_batch_map is int) (PR #119).
213+
214+
* Update flowsom mapping similarity so we subset to just markers to correct, and lisi to remove control samples
215+
and unlabelled cells (PR #119).

‎scripts/create_resources/process_raw_datasets_hpc.sh‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
#!/bin/bash
2+
3+
# script to launch the process raw dataset workflow on slurm via seqera tower.
4+
# leave the input_states to s3 bucket as the datasets raw files are stored there.
5+
6+
cat > /tmp/params.yaml << 'HERE'
7+
input_states: s3://openproblems-data/resources/task_cyto_batch_integration/datasets_raw/**/state.yaml
8+
rename_keys: 'input:output_dataset'
9+
output_state: '$id/state.yaml'
10+
settings: '{"output_unintegrated": "$id/unintegrated.h5ad", "output_censored_split1": "$id/censored_split1.h5ad", "output_censored_split2": "$id/censored_split2.h5ad"}'
11+
publish_dir: /vast/scratch/users/putri.g/cytobenchmark/benchmark_out_hpc/datasets/
12+
HERE
13+
14+
tw launch https://github.com/openproblems-bio/task_cyto_batch_integration.git \
15+
--revision build/main \
16+
--pull-latest \
17+
--main-script target/nextflow/workflows/process_datasets/main.nf \
18+
--workspace 80689470953249 \
19+
--params-file /tmp/params.yaml \
20+
--entry-name auto \
21+
--config scripts/labels_tw_wehi.config \
22+
--labels task_cyto_batch_integration,process_datasets

‎scripts/get_intermediate_files/copy_intermediate_files.py‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
# Using the csv file produced by `find_intermediate_files.py`
2+
# copy the intermediate files out.
3+
# The csv file must be in the same directory as this script, and the files will be copied to the same directory as well.
4+
5+
import os
6+
import re
7+
import shutil
8+
from pathlib import Path
9+
10+
import pandas as pd
11+
12+
base_dir = str(Path(__file__).resolve().parent)
13+
print(f"Downloading to {base_dir}")
14+
15+
s3_bucket_files = pd.read_csv(f"{base_dir}/task_s3_bucket_map.csv")
16+
17+
18+
def process_row(row):
19+
print(f"Processing {row.method}, dataset {row.dataset}, metric {row.metric}")
20+
21+
data_dir = row.s3_path
22+
23+
if pd.isna(row.metric):
24+
for itemname in os.listdir(data_dir):
25+
item_path = os.path.join(data_dir, itemname)
26+
27+
# Skip if it's a directory or not .h5ad file
28+
if os.path.isdir(item_path) or not itemname.endswith(".h5ad"):
29+
continue
30+
31+
if (
32+
"no_integration" in row.process_name
33+
or "perfect_integration" in row.process_name
34+
):
35+
split_name = re.search(r"split\d+", itemname)
36+
if not split_name:
37+
exit(f"Error: cannot find split part in {itemname}")
38+
split_name = split_name.group() # Extract the matched string
39+
new_filename = f"{base_dir}/{row.dataset}/method_out/{row.method}_{split_name}.h5ad"
40+
41+
# create directory if not exists
42+
os.makedirs(os.path.dirname(new_filename), exist_ok=True)
43+
44+
print(f"Copying {itemname} to {os.path.basename(new_filename)}")
45+
shutil.copy2(item_path, new_filename)
46+
47+
else:
48+
split_name = "2" if "process1" in row.process_name else "1"
49+
new_filename = f"{base_dir}/{row.dataset}/method_out/{row.method}_split{split_name}.h5ad"
50+
51+
# create directory if not exists
52+
os.makedirs(os.path.dirname(new_filename), exist_ok=True)
53+
54+
print(f"Copying {itemname} to {os.path.basename(new_filename)}")
55+
shutil.copy2(item_path, new_filename)
56+
57+
else:
58+
new_filename = (
59+
f"{base_dir}/{row.dataset}/metric_out/{row.metric}_{row.method}.h5ad"
60+
)
61+
62+
# create directory if not exists
63+
os.makedirs(os.path.dirname(new_filename), exist_ok=True)
64+
65+
for itemname in os.listdir(data_dir):
66+
item_path = os.path.join(data_dir, itemname)
67+
68+
# Skip if it's a directory or not .h5ad file
69+
if os.path.isdir(item_path) or not itemname.endswith(".h5ad"):
70+
continue
71+
72+
print(f"Copying {itemname} to {os.path.basename(new_filename)}")
73+
shutil.copy2(item_path, new_filename)
74+
75+
76+
for row in s3_bucket_files.itertuples(index=True):
77+
process_row(row)

‎scripts/get_intermediate_files/download_intermediate_files.py‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
# Use me to download intermediate files from AWS S3 bucket.
2+
3+
import argparse
4+
import os
5+
import re
6+
import shutil
7+
8+
import pandas as pd
9+
10+
parser = argparse.ArgumentParser(description="Download intermediate files from S3")
11+
parser.add_argument(
12+
"--base_dir", type=str, required=True, help="Base directory to download files to"
13+
)
14+
args = parser.parse_args()
15+
16+
17+
s3_bucket_files = pd.read_csv(f"{args.base_dir}/task_s3_bucket_map.csv")
18+
19+
for row in s3_bucket_files.itertuples(index=True):
20+
# skip files that are already downloaded
21+
file_exists = []
22+
for split in ["1", "2"]:
23+
new_filename = (
24+
f"{args.base_dir}/{row.dataset}/method_out/{row.method}_{split}.h5ad"
25+
)
26+
file_exists.append(os.path.exists(new_filename))
27+
28+
if all(file_exists):
29+
print(f"Files for {row.method} already exist, skipping...")
30+
continue
31+
32+
print(f"Processing row: {row}")
33+
if pd.isna(row.metric):
34+
outdir = f"{args.base_dir}/{row.dataset}/method_out/temp"
35+
os.makedirs(outdir, exist_ok=True)
36+
os.system(
37+
f"common/scripts/fetch_task_run --input {row.s3_path} --output {outdir}/"
38+
)
39+
for itemname in os.listdir(outdir):
40+
item_path = os.path.join(outdir, itemname)
41+
42+
# Skip if it's a directory or not .h5ad file
43+
if os.path.isdir(item_path) or not itemname.endswith(".h5ad"):
44+
continue
45+
46+
elif (
47+
"no_integration" in row.process_name
48+
or "perfect_integration" in row.process_name
49+
):
50+
split_name = re.search(r"split\d+", itemname)
51+
if not split_name:
52+
exit(f"Error: cannot find split part in {itemname}")
53+
split_name = split_name.group() # Extract the matched string
54+
new_filename = f"{args.base_dir}/{row.dataset}/method_out/{row.method}_{split_name}.h5ad"
55+
print(f"Renaming {item_path} to {new_filename}")
56+
shutil.move(item_path, new_filename)
57+
58+
else:
59+
split_name = "2" if "process1" in row.process_name else "1"
60+
new_filename = f"{args.base_dir}/{row.dataset}/method_out/{row.method}_split{split_name}.h5ad"
61+
print(f"Renaming {item_path} to {new_filename}")
62+
shutil.move(item_path, new_filename)
63+
shutil.rmtree(outdir)
64+
65+
else:
66+
new_filename = (
67+
f"{args.base_dir}/{row.dataset}/metric_out/{row.metric}_{row.method}.h5ad"
68+
)
69+
70+
if os.path.exists(new_filename):
71+
print(f"File {new_filename} already exists, skipping...")
72+
continue
73+
74+
outdir = f"{args.base_dir}/{row.dataset}/metric_out/temp"
75+
os.makedirs(outdir, exist_ok=True)
76+
os.system(
77+
f"common/scripts/fetch_task_run --input {row.s3_path} --output {outdir}/"
78+
)
79+
for itemname in os.listdir(outdir):
80+
item_path = os.path.join(outdir, itemname)
81+
82+
# Skip if it's a directory or the .h5ad file
83+
if os.path.isdir(item_path) or not itemname.endswith(".h5ad"):
84+
continue
85+
86+
else:
87+
# TODO the split is not needed.. made a mistake before.
88+
print(f"Renaming {item_path} to {new_filename}")
89+
shutil.move(item_path, new_filename)
90+
91+
# clean up temp folder
92+
shutil.rmtree(outdir)

‎scripts/get_intermediate_files/find_intermediate_files.py‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
# Run me to find the AWS s3 paths for intermediate files for each task.
2+
3+
import argparse
4+
import re
5+
6+
import pandas as pd
7+
8+
parser = argparse.ArgumentParser(
9+
description="Extract intermediate file paths from log file"
10+
)
11+
parser.add_argument(
12+
"--log_file_path", type=str, required=True, help="Path to the log file to parse"
13+
)
14+
parser.add_argument(
15+
"--output_csv_path",
16+
type=str,
17+
default="task_s3_bucket_map.csv",
18+
help="Path to save the output CSV file with task and S3 path mapping",
19+
)
20+
args = parser.parse_args()
21+
22+
# Feb-04 16:09:49.696 [Task submitter] DEBUG nextflow.executor.GridTaskHandler - [SLURM] submitted process auto:run_benchmark:run_wf:runEachWf:harmonypy:processWf:harmonypy_process (human_blood_mass_cytometry.harmonypy) > jobId: 25169450; workDir: /vast/scratch/users/putri.g/nextflow/work/6f/488d302c4e33171f2d69c8107b7a9c
23+
24+
25+
# to pull log lines with:
26+
# Sep-01 00:25:30.723 [AWSBatch-executor-26] DEBUG n.c.aws.batch.AwsBatchTaskHandler - [AWS BATCH] Process `auto:run_benchmark:run_wf:runEachWf:harmonypy:processWf:harmonypy_process (human_blood_mass_cytometry.harmonypy)` submitted > job=70af3d53-8070-45a6-96f4-fd37580b9d00; work-dir=s3://openproblems-work/scratch/2e8BJJovTcoBGj/60/c803767cf237028c908fadf447b1b9
27+
# 1. Match only lines that contain "[AWS BATCH] Process `...`"
28+
# 2. Extract the last part inside the backticks (e.g. "extract_uns_metadata_process (mouse_spleen_flow_cytometry)")
29+
# 3. Extract the work-dir S3 path
30+
# zooming into the ([\w\d_]+ \([^)]+\)) part:
31+
# | Component | Meaning |
32+
# | ---------- | --------------------------------------------------------------------------------------------- |
33+
# | `[\w\d_]+` | Matches the process name: letters, digits, underscores (e.g., `extract_uns_metadata_process`) |
34+
# | ` ` | Space between process and dataset |
35+
# | `\(` | Opening parenthesis (escaped) |
36+
# | `[^)]+` | Match everything inside the parentheses (dataset name), up to the next `)` |
37+
# | `\)` | Closing parenthesis |
38+
# that captures harmonypy_process (human_blood_mass_cytometry.harmonypy)
39+
# the work-dir part:
40+
# | Component | Meaning |
41+
# | --------------- | ----------------------------------------------------------------------------- |
42+
# | `.*?` | Non-greedy match of any characters between the backtick block and `work-dir=` |
43+
# | `work-dir=` | Literal match |
44+
# | `(s3://[^ ;]+)` | The full S3 path. Match any characters except space or semicolon |
45+
46+
47+
aws_batch_pattern = re.compile(
48+
r"\[AWS BATCH\] Process `.*:([\w\d_]+ \([^)]+\))`.*?work-dir=(s3://[^ ;]+)"
49+
)
50+
51+
slurm_pattern = re.compile(
52+
r"\[SLURM\] submitted process .*:([\w\d_]+ \([^)]+\)).*?workDir: ([^ ]+)"
53+
)
54+
55+
is_aws = False
56+
57+
task_s3_bucket_map = []
58+
with open(args.log_file_path, "r") as log_file:
59+
for line_number, line in enumerate(log_file, start=1):
60+
if is_aws:
61+
match = aws_batch_pattern.search(line)
62+
else:
63+
match = slurm_pattern.search(line)
64+
if match:
65+
# e.g., "extract_uns_metadata_process (mouse_spleen_flow_cytometry)"
66+
task_component = match.group(1)
67+
# task can be like this. so can pull the dataset name, task, and whether it is method or metric
68+
# harmonypy_process (human_blood_mass_cytometry.harmonypy)
69+
# flowsom_mapping_similarity_process (mouse_spleen_flow_cytometry.cytonorm_no_controls_to_mid.flowsom_mapping_similarity)
70+
process = task_component.split(" ")[0]
71+
task_info = task_component.split(" ")[1].strip("()").split(".")
72+
dataset_name = task_info[0]
73+
metric_name = None
74+
method_name = None
75+
if 2 <= len(task_info) <= 3:
76+
method_name = task_info[1]
77+
if len(task_info) == 3:
78+
metric_name = task_info[2]
79+
80+
# e.g., "s3://openproblems-work/scratch/2e8"
81+
s3_path = match.group(2).strip()
82+
task_s3_bucket_map.append(
83+
(line_number, process, dataset_name, method_name, metric_name, s3_path)
84+
)
85+
86+
task_s3_bucket_map = pd.DataFrame(
87+
task_s3_bucket_map,
88+
columns=["line", "process_name", "dataset", "method", "metric", "s3_path"],
89+
)
90+
task_s3_bucket_map.sort_values(by="process_name", inplace=True)
91+
task_s3_bucket_map.to_csv(args.output_csv_path, index=False)

0 commit comments

Comments
 (0)