Skip to content

Commit 94ce9aa

Browse files
ChromojonesChromojones
committed
updated flowRNAanalysis script to flowbio library
1 parent 539a77e commit 94ce9aa

1 file changed

Lines changed: 128 additions & 135 deletions

File tree

‎analysis/flowRNAanalysis.py‎

Lines changed: 128 additions & 135 deletions
Original file line numberDiff line numberDiff line change
@@ -6,37 +6,31 @@
66
from typing import Dict, List, Tuple
77
import requests
88
import numpy as np
9-
import fnmatch
9+
import re
10+
from flowbio import Client
1011

1112
# -------------------------
12-
# CLI Usage - python3 ./flowRNAanalysis.py --pid ######### --filter sample_name "*A" -n #ofbatches
13+
# CLI Usage - python3 ./flowRNAanalysis.py --pid ######### --filter sample_name 'STAU2_HepG2.*$' -n #ofbatches --start-batch 1 --end-batch 10
1314
# -------------------------
1415
def parse_args():
1516
p = argparse.ArgumentParser(description="Run Flow.bio RNA-seq analysis (fetch + client-side filter by sample name)")
1617
p.add_argument("--pid", "--PID", dest="project_id", required=True,
1718
help="Flow.bio Project ID (string)")
1819
p.add_argument("--filter", nargs=2, metavar=("KEY", "VALUE"), default=None,
19-
help='Metadata filter. Supported now: --filter sample_name "*A"')
20+
help='Metadata filter. Supported now: --filter sample_name "<regex>"')
2021
p.add_argument("-n", "--num-chunks", type=int, default=1,
2122
help="Split selected samples into N executions using numpy.array_split (default: 1)")
22-
p.add_argument("--dry-run", action="store_true",
23-
help="Resolve everything and print payloads without submitting")
24-
p.add_argument("--verbose", action="store_true",
25-
help="Enable DEBUG logging")
26-
p.add_argument("--log-level", choices=["DEBUG", "INFO", "WARNING", "ERROR"], default=None,
27-
help="Explicit log level (overrides --verbose)")
23+
p.add_argument("--start-batch", type=int, default=1,
24+
help="Start execution from batch number (1-based, default: 1)")
25+
p.add_argument("--end-batch", type=int, default=None,
26+
help="End execution at batch number (1-based, default: all batches)")
2827
return p.parse_args()
2928

3029
# -------------------------
3130
# Logging
3231
# -------------------------
33-
def setup_logging(args):
34-
level = logging.INFO
35-
if args.verbose:
36-
level = logging.DEBUG
37-
if args.log_level:
38-
level = getattr(logging, args.log_level)
39-
logging.basicConfig(level=level, format="%(asctime)s | %(levelname)-8s | %(message)s")
32+
def setup_logging():
33+
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)-8s | %(message)s")
4034

4135
# -------------------------
4236
# CLIP pipeline settings
@@ -96,15 +90,6 @@ def fetch_all_project_samples(session: requests.Session, token: str, project_id:
9690
logging.info("Fetched %d samples from project %s", len(collected), project_id)
9791
return collected
9892

99-
def resolve_pipeline_version_id(session: requests.Session, token: str, pipeline_id: str, version_label: str) -> str:
100-
headers = {"Authorization": f"Bearer {token}"}
101-
r = session.get(f"{API_BASE}/pipelines/{pipeline_id}", headers=headers, timeout=30)
102-
_raise_for_status(r)
103-
pipeline = r.json()
104-
match = next((v for v in pipeline.get("versions", []) if v.get("name") == version_label), None)
105-
if not match:
106-
raise RuntimeError(f"Version {version_label!r} not found on pipeline {pipeline_id}")
107-
return match["id"]
10893

10994
def fetch_prep_execution(session: requests.Session, token: str, prep_execution_id: str) -> Dict:
11095
headers = {"Authorization": f"Bearer {token}"}
@@ -158,55 +143,55 @@ def build_data_params_from_execution(execution: Dict, file_map: Dict[str, str])
158143
# -------------------------
159144
# Client-side filter (sample_name glob)
160145
# -------------------------
161-
def filter_by_sample_name(samples: List[Dict], glob_expr: str | None) -> List[Dict]:
162-
if not glob_expr:
146+
def filter_by_sample_name(samples: List[Dict], regex_expr: str | None) -> List[Dict]:
147+
if not regex_expr:
163148
return samples
164-
matched = [s for s in samples if fnmatch.fnmatch((s.get("name") or ""), glob_expr)]
165-
logging.info("Filter sample_name=%r matched %d / %d samples", glob_expr, len(matched), len(samples))
149+
try:
150+
pattern = re.compile(regex_expr)
151+
except re.error as e:
152+
raise SystemExit(f"Invalid regex for sample_name: {e}")
153+
matched = [s for s in samples if pattern.search((s.get("name") or ""))]
154+
logging.info("Filter sample_name=%r matched %d / %d samples", regex_expr, len(matched), len(samples))
166155
return matched
167156

168157
# -------------------------
169158
# Main
170159
# -------------------------
171160
def main():
172161
args = parse_args()
173-
setup_logging(args)
162+
setup_logging()
174163

175164
project_id = args.project_id
176165

177166
# RNA pipeline settings
178167
prep_execution_id = PIPELINE_RNA["prep_execution_id"]
179-
pipeline_id = PIPELINE_RNA["pipeline_id"]
180-
pipeline_version = PIPELINE_RNA["pipeline_version"]
168+
pipeline_name = "RNA-Seq" # Using standard RNA-Seq pipeline name
169+
pipeline_version = PIPELINE_RNA["pipeline_version"]
170+
nextflow_version = "24.04.2"
171+
172+
# Initialize flowbio client (for submission only)
173+
client = Client()
181174

182-
# HTTP session & auth
175+
# REST session & auth for discovery endpoints
183176
session = requests.Session()
184177
token = login(session)
185-
headers = {"Authorization": f"Bearer {token}"}
186-
187-
# Resolve pipeline version ID
188-
version_id = resolve_pipeline_version_id(session, token, pipeline_id, pipeline_version)
189-
logging.debug("Resolved pipeline version id=%s for label=%s", version_id, pipeline_version)
190178

191-
# Prep execution & reference files
179+
# Prep execution & reference files via REST
192180
ex = fetch_prep_execution(session, token, prep_execution_id)
193-
fileset_id = (ex.get("fileset") or {}).get("id")
194-
if not fileset_id:
195-
raise RuntimeError("Prep execution has no fileset id")
196181
data_params = build_data_params_from_execution(ex, FILE_MAP)
197182

198-
# Fetch all samples for the project (no TSV), then client-side filter by sample name glob
183+
# Fetch all samples for the project via REST
199184
project_samples = fetch_all_project_samples(session, token, project_id, page_size=100)
200185

201-
# parse --filter
202-
name_glob = None
186+
# Parse --filter
187+
name_regex = None
203188
if args.filter:
204189
key, value = args.filter
205190
if key.lower() != "sample_name":
206-
raise SystemExit("Only --filter sample_name \"<glob>\" is supported in this iteration.")
207-
name_glob = value
191+
raise SystemExit("Only --filter sample_name \"<regex>\" is supported in this iteration.")
192+
name_regex = value
208193

209-
selected = filter_by_sample_name(project_samples, name_glob)
194+
selected = filter_by_sample_name(project_samples, name_regex)
210195

211196
# Print filtered list
212197
print(f"\nFiltered {len(selected)} samples:")
@@ -221,106 +206,114 @@ def main():
221206
chunks = [list(chunk) for chunk in np.array_split(np.array(selected, dtype=object), n_chunks)]
222207
logging.info("Prepared %d execution batch(es)", len(chunks))
223208

224-
# Build and submit one execution per chunk
209+
# Determine which batches to execute
210+
start_batch = max(1, args.start_batch)
211+
end_batch = args.end_batch if args.end_batch is not None else len(chunks)
212+
end_batch = min(end_batch, len(chunks))
213+
214+
if start_batch > len(chunks):
215+
raise SystemExit(f"Start batch {start_batch} exceeds total batches {len(chunks)}")
216+
217+
logging.info("Will execute batches %d to %d (out of %d total batches)", start_batch, end_batch, len(chunks))
218+
219+
# Build and submit one execution per chunk in the specified range
225220
run_urls = []
226221
for i, chunk in enumerate(chunks, start=1):
227-
rows = [{
228-
"sample": s["id"],
229-
"values": {
222+
# Skip batches outside the specified range
223+
if i < start_batch or i > end_batch:
224+
logging.info("Skipping batch %d (not in range %d-%d)", i, start_batch, end_batch)
225+
continue
226+
# Build sample_params in the format expected by flowbio
227+
sample_params = {}
228+
for s in chunk:
229+
sample_params[s["id"]] = {
230230
"group": s.get("name", ""),
231231
"replicate": "1",
232232
}
233-
} for s in chunk]
234-
235-
payload = {
236-
"params": {
237-
# UMI
238-
"with_umi": "true",
239-
"umitools_extract_method": "regex",
240-
"umitools_bc_pattern": "^(?P<discard_1>.{4})(?P<umi_1>.{5})",
241-
"skip_umi_extract": "false",
242-
"umitools_dedup_stats": "false",
243-
"save_umi_intermeds": "false",
244-
245-
# Annotation/grouping
246-
"gencode": "false",
247-
"gtf_extra_attributes": "gene_name",
248-
"gtf_group_features": "gene_id",
249-
"featurecounts_group_type": "gene_biotype",
250-
"featurecounts_feature_type": "exon",
251-
252-
# Align/quant
253-
"aligner": "star_salmon",
254-
"pseudo_aligner": "",
255-
"bam_csi_index": "false",
256-
"star_ignore_sjdbgtf": "false",
257-
"stringtie_ignore_gtf": "false",
258-
"save_unaligned": "false",
259-
"save_align_intermeds": "false",
260-
"skip_markduplicates": "false",
261-
"skip_alignment": "false",
262-
"skip_pseudo_alignment": "false",
263-
264-
# Trimming
265-
"trimmer": "trimgalore",
266-
"skip_trimming": "false",
267-
"save_trimmed": "false",
268-
269-
# rRNA options
270-
"remove_ribo_rna": "false",
271-
"ribo_database_manifest": "./assets/rrna-db-defaults.txt",
272-
"save_non_ribo_reads": "false",
273-
274-
# QC
275-
"deseq2_vst": "true",
276-
"skip_bigwig": "false",
277-
"skip_stringtie": "false",
278-
"skip_fastqc": "false",
279-
"skip_preseq": "true",
280-
"skip_qualimap": "false",
281-
"skip_rseqc": "false",
282-
"skip_biotype_qc": "false",
283-
"skip_deseq2_qc": "false",
284-
"skip_multiqc": "false",
285-
"skip_qc": "false",
286-
},
287-
# File/data bindings are provided via data_params (from FILE_MAP)
288-
"data_params": data_params,
289-
"csv_params": {"samplesheet": {"rows": rows, "paired": "both"}},
290-
"retries": None,
291-
"nextflow_version": "24.04.2",
292-
"fileset": fileset_id,
293-
"resequence_samples": False,
294-
}
295233

296-
if args.dry_run:
297-
logging.info("DRY RUN: Batch %d/%d: %d samples", i, len(chunks), len(chunk))
298-
for s in chunk[:10]:
299-
logging.info(" %s | id=%s", s.get("name", ""), s.get("id", ""))
300-
continue
234+
# Pipeline parameters (RNA-seq specific)
235+
params = {
236+
# UMI
237+
"with_umi": "true",
238+
"umitools_extract_method": "regex",
239+
"umitools_bc_pattern": "^(?P<discard_1>.{4})(?P<umi_1>.{5})",
240+
"skip_umi_extract": "false",
241+
"umitools_dedup_stats": "false",
242+
"save_umi_intermeds": "false",
243+
244+
# Annotation/grouping
245+
"gencode": "false",
246+
"gtf_extra_attributes": "gene_name",
247+
"gtf_group_features": "gene_id",
248+
"featurecounts_group_type": "gene_biotype",
249+
"featurecounts_feature_type": "exon",
250+
251+
# Align/quant
252+
"aligner": "star_salmon",
253+
"pseudo_aligner": "",
254+
"bam_csi_index": "false",
255+
"star_ignore_sjdbgtf": "false",
256+
"stringtie_ignore_gtf": "false",
257+
"save_unaligned": "false",
258+
"save_align_intermeds": "false",
259+
"skip_markduplicates": "false",
260+
"skip_alignment": "false",
261+
"skip_pseudo_alignment": "false",
262+
263+
# Trimming
264+
"trimmer": "trimgalore",
265+
"skip_trimming": "false",
266+
"save_trimmed": "false",
267+
268+
# rRNA options
269+
"remove_ribo_rna": "false",
270+
"ribo_database_manifest": "./assets/rrna-db-defaults.txt",
271+
"save_non_ribo_reads": "false",
272+
273+
# QC
274+
"deseq2_vst": "true",
275+
"skip_bigwig": "false",
276+
"skip_stringtie": "false",
277+
"skip_fastqc": "false",
278+
"skip_preseq": "true",
279+
"skip_qualimap": "false",
280+
"skip_rseqc": "false",
281+
"skip_biotype_qc": "false",
282+
"skip_deseq2_qc": "false",
283+
"skip_multiqc": "false",
284+
"skip_qc": "false",
285+
}
301286

302287
if i == 1:
303288
proceed = input("Submit? (y/n): ").strip().lower()
304289
if proceed != "y":
305290
logging.info("Aborted by user.")
306291
sys.exit(0)
307292

308-
r = session.post(
309-
f"{API_BASE}/pipelines/versions/{version_id}/run",
310-
headers=headers,
311-
json=payload,
312-
timeout=60,
313-
)
314-
_raise_for_status(r)
315-
run_id = r.json().get("id")
316-
if not run_id:
317-
raise RuntimeError(f"Submission succeeded but no run id in response: {r.text}")
318-
url = f"https://app.flow.bio/executions/{run_id}"
319-
run_urls.append(url)
320-
print(url)
321-
322-
if args.dry_run:
323-
logging.info("DRY RUN complete: %d batch(es) prepared.", len(chunks))
293+
try:
294+
# Use flowbio client to run pipeline
295+
execution = client.run_pipeline(
296+
name=pipeline_name,
297+
version=pipeline_version,
298+
nextflow_version=nextflow_version,
299+
params=params,
300+
data_params=data_params,
301+
sample_params={"samples": sample_params}
302+
)
303+
304+
run_id = execution.get("id")
305+
if not run_id:
306+
raise RuntimeError(f"Submission succeeded but no run id in response: {execution}")
307+
308+
url = f"https://app.flow.bio/executions/{run_id}"
309+
run_urls.append(url)
310+
print(f"Batch {i}: {url}")
311+
312+
except Exception as e:
313+
logging.error("Failed to submit batch %d: %s", i, e)
314+
continue
315+
316+
logging.info("Completed submission of %d batches", len(run_urls))
324317

325318
if __name__ == "__main__":
326319
try:

0 commit comments

Comments
 (0)