diff --git a/BRB/PushButton.py b/BRB/PushButton.py
index a79cfab..b1eda29 100644
--- a/BRB/PushButton.py
+++ b/BRB/PushButton.py
@@ -1,6 +1,7 @@
import concurrent.futures
import glob
import os
+import re
import shutil
import stat
from collections import namedtuple
@@ -26,6 +27,19 @@
# Name of the per-group ownership marker written into the group's outputDir.
MARKER_NAME = "running.pid"
+# Parkour protocols of the two halves of a 10x Multiome sample. They are
+# processed jointly by cellranger-arc (scMultiome), never by scRNAseq/scATAC.
+MULTIOME_PROTOCOLS = {
+ "Chromium_NextGEM_Multiome_GEX": "_RNA",
+ "Chromium_NextGEM_Multiome_ATAC": "_ATAC",
+}
+
+
+# Written into a scMultiome outputDir when some pairs could not be run in this
+# pass (waiting for the other lane, unpaired, ...). One reason per line. Its
+# presence turns the e-mail row into ATTENTION and is printed by run_brb.
+MULTIOME_NOTE = "multiome.attention"
+
# Fixed for now, deliberately not .ini-configurable. Kept low because not all
# of a group's work is Slurm-offloaded: RELACS runs `demultiplex_relacs -p 10`
# locally, each snakePipes driver is a real local process, and the --sequencer
@@ -116,7 +130,11 @@ def createPath(config, group, project, org_label, analysisType, tuples):
BRB.misc.pacifier(project),
)
os.makedirs(baseDir, mode=0o700, exist_ok=True)
- oDir = os.path.join(baseDir, f"{BRB.misc.pacifier(analysisType)}_{org_label}")
+ dirName = f"{BRB.misc.pacifier(analysisType)}_{org_label}"
+ if len(tuples[0]) > 4:
+ # Mixed-protocol split (see GetResults): keep the groups' dirs apart.
+ dirName += f"_{tuples[0][4]}"
+ oDir = os.path.join(baseDir, dirName)
os.makedirs(oDir, mode=0o700, exist_ok=True)
return oDir
@@ -1018,6 +1036,180 @@ def scATAC(config, group, project, organism, analysisType, tuples):
return outputDir, 0, True
+def scMultiome(config, group, project, organism, analysisType, tuples):
+ """
+ 10x Multiome (GEX + ATAC), one joint cellranger-arc run per sample.
+
+ GEX and ATAC are paired by Parkour sample name: _RNA with
+ _ATAC. With dissectBCL --forcelanesplit the two halves can sit in
+ different _lanes_ folders, which BRB processes one at a time
+ and in no fixed order. A pair is therefore run by the first lane folder
+ whose pass finds both halves ready: FASTQs in this folder, or in a sibling
+ folder that already has fastq.made. A pair that has .done (written
+ by the 10x_ARC wrapper) in a sibling lane's output is not run again.
+ Pairs that are not ready yet are left to the sibling lane's pass, so this
+ group is only marked analysis.done when it has nothing left to wait for.
+
+ Anything that keeps a pair from running is written, one line each, to
+ MULTIOME_NOTE in outputDir and logged as a warning:
+ - waiting: one half's FASTQs are not in any ready lane folder yet;
+ - unpaired: Parkour has only the GEX or only the ATAC library on this
+ flowcell (other flowcell, different organism, or a name that does not
+ end in _RNA/_ATAC);
+ - duplicate: two GEX (or two ATAC) libraries share one base name;
+ - split: one half has FASTQs in more than one lane folder, while
+ cellranger-arc takes one input folder per modality.
+ Only "waiting" holds back analysis.done; the others never resolve
+ by themselves and need the operator.
+ """
+ project = BRB.misc.pacifier(project)
+ _org_name, org_label, org_yaml = organism
+ outputDir = createPath(config, group, project, org_label, analysisType, tuples)
+ if os.path.exists(os.path.join(outputDir, "analysis.done")):
+ return outputDir, 0, True
+ runID = config.get("Options", "runID")
+ baseData = config.get("Paths", "baseData")
+ lane = re.match(r"^(.*)_lanes_\d+$", runID)
+ if lane:
+ laneIDs = sorted(
+ (
+ p.name
+ for p in Path(baseData).glob(f"{lane.group(1)}_lanes_*")
+ if re.search(r"_lanes_\d+$", p.name)
+ ),
+ key=lambda n: int(n.rsplit("_", 1)[1]),
+ )
+ else:
+ laneIDs = [runID]
+ ready = [
+ laneID
+ for laneID in laneIDs
+ if laneID == runID or Path(baseData, laneID, "fastq.made").exists()
+ ]
+
+ def inLane(path, laneID):
+ # Same path with the run folder component swapped for laneID.
+ return Path(*(laneID if part == runID else part for part in Path(path).parts))
+
+ # Project folder of this run: where linkFiles/scATAC also read FASTQs.
+ if tuples[0][3]:
+ projectDir = Path(baseData, runID, f"Project_{project}")
+ else:
+ projectDir = Path(
+ config.get("Paths", "groupData"),
+ BRB.misc.pacifier(group),
+ BRB.misc.getLatestSeqdir(config.get("Paths", "groupData"), group),
+ runID,
+ f"Project_{project}",
+ )
+
+ # Locate each Parkour library: pair base, FASTQ sample name, and the
+ # ready lane folders that hold its FASTQs.
+ issues = []
+ halves = {"_RNA": {}, "_ATAC": {}}
+ for t in tuples:
+ suffix = MULTIOME_PROTOCOLS[t[2]]
+ base = BRB.misc.pacifier(t[1]).removesuffix(suffix)
+ found = [
+ inLane(projectDir, laneID)
+ for laneID in ready
+ if any(inLane(projectDir, laneID).glob(f"Sample_{t[0]}/*.fastq.gz"))
+ ]
+ # dissectBCL rewrites umlauts and spaces in FASTQ names, so the name
+ # the wrapper searches for is taken from disk, not from Parkour.
+ r1 = sorted(p for d in found for p in d.glob(f"Sample_{t[0]}/*_R1.fastq.gz"))
+ fastqName = r1[0].name.removesuffix("_R1.fastq.gz") if r1 else t[1]
+ if base in halves[suffix]:
+ issues.append(
+ f"{base}: two {suffix[1:]} libraries in Parkour "
+ f"({halves[suffix][base]['lib']}, {t[0]}), not run."
+ )
+ halves[suffix][base]["dup"] = True
+ continue
+ halves[suffix][base] = {
+ "lib": t[0],
+ "name": fastqName,
+ "dirs": found,
+ "dup": False,
+ }
+
+ runs = {}
+ waiting = []
+ for base in sorted(set(halves["_RNA"]) | set(halves["_ATAC"])):
+ gex, atac = halves["_RNA"].get(base), halves["_ATAC"].get(base)
+ if gex is None or atac is None:
+ have, miss = ("ATAC", "GEX") if gex is None else ("GEX", "ATAC")
+ issues.append(
+ f"{base}: {have} library {(gex or atac)['lib']} has no {miss} "
+ "partner on this flowcell in Parkour, not run."
+ )
+ continue
+ if gex["dup"] or atac["dup"]:
+ continue
+ doneIn = [
+ laneID
+ for laneID in laneIDs
+ if laneID != runID
+ and inLane(outputDir, laneID).joinpath(f"{base}.done").exists()
+ ]
+ if doneIn:
+ log.info(f"{project}: Multiome sample {base} already done in {doneIn[0]}.")
+ continue
+ split = [
+ f"{kind} {h['lib']} has FASTQs in {len(h['dirs'])} lane folders"
+ for kind, h in (("GEX", gex), ("ATAC", atac))
+ if len(h["dirs"]) > 1
+ ]
+ if split:
+ issues.append(
+ f"{base}: {'; '.join(split)}, cellranger-arc takes one folder "
+ "per modality, not run."
+ )
+ continue
+ if not gex["dirs"] or not atac["dirs"]:
+ waiting.append(base)
+ continue
+ runs.setdefault((gex["dirs"][0], atac["dirs"][0]), []).append(
+ f"--pair {base}:{gex['name']}:{atac['name']}"
+ )
+ if waiting:
+ issues.append(
+ f"{', '.join(waiting)}: waiting for the other half's FASTQs (lane "
+ "folder without fastq.made yet). The pass of the lane folder that "
+ "completes them runs them; if it never comes, run them by hand."
+ )
+ else:
+ issues = [i for i in issues if not i.startswith("waiting:")]
+ note = Path(outputDir, MULTIOME_NOTE)
+ for issue in issues:
+ log.warning(f"{runID} {project} Multiome: {issue}")
+ if issues:
+ note.write_text("\n".join(issues) + "\n")
+ else:
+ note.unlink(missing_ok=True)
+ if not runs:
+ if not waiting:
+ touchDone(outputDir)
+ return outputDir, 0, False
+
+ snakeMakePath = "{}/bin".format(config.get("Options", "snakemakeWorkflowBaseDir"))
+ for (gexDir, atacDir), pairs in runs.items():
+ CMD = config.get("10x", "ARC")
+ CMD += f" --gex-indir {gexDir} --atac-indir {atacDir} " + " ".join(pairs)
+ CMD += f" -o {outputDir} {org_yaml} --snakemakePath {snakeMakePath}"
+ log.info(f"scMultiome wf CMD: {CMD}")
+ try:
+ runManagedSubprocess(CMD)
+ except:
+ return outputDir, 1, False
+ copyCellRanger(config, outputDir, tuples[0][3])
+ stripRights(outputDir)
+ tidyUpABit(outputDir)
+ if not waiting:
+ touchDone(outputDir)
+ return outputDir, 0, True
+
+
def runOneGroup(config, item, registry):
"""
Ownership-guarded dispatch of a single WorkItem, coordinated through
@@ -1140,19 +1332,19 @@ def skipRow(status):
f"Processed project {BRB.misc.pacifier(item.project)} with the {item.pipeline} pipeline. {item.analysisType}, {org_name}. Rerun = {reruncount}"
)
crashed = False
- return [
- BRB.ET.phoneHome(
- config,
- outputDir,
- item.pipeline,
- item.tuples,
- org_name,
- org_label,
- item.project,
- item.analysisType,
- )
- + [sambaUpdate, reruncount]
- ]
+ row = BRB.ET.phoneHome(
+ config,
+ outputDir,
+ item.pipeline,
+ item.tuples,
+ org_name,
+ org_label,
+ item.project,
+ item.analysisType,
+ ) + [sambaUpdate, reruncount]
+ if Path(outputDir, MULTIOME_NOTE).exists():
+ row[4] = f"ATTENTION (see {outputDir}/{MULTIOME_NOTE})"
+ return [row]
log.warning(
f"FAILED project {BRB.misc.pacifier(item.project)} with the {item.pipeline} pipeline. {item.analysisType}, {org_name}. Rerun = {reruncount}"
)
@@ -1175,6 +1367,36 @@ def skipRow(status):
registry.unregister_group(outputDir)
+def filterLaneSamples(config, project, libraries):
+ """
+ On a single-lane folder (_lanes_, from dissectBCL
+ --forcelanesplit), keep only the libraries that have FASTQs in that
+ folder's Project_/Sample_. Other run folders hold all
+ their samples, so they are returned unchanged. Multiome libraries are
+ always kept: scMultiome pairs them across lane folders and decides
+ whether a pair is run, waits for the other lane, or needs the operator.
+ """
+ runID = config.get("Options", "runID")
+ lane = re.match(r"^(.*)_lanes_\d+$", runID)
+ if not lane:
+ return libraries
+ baseData = config.get("Paths", "baseData")
+ pacified = BRB.misc.pacifier(project)
+ projectDir = Path(baseData, runID, f"Project_{pacified}")
+ kept = {}
+ for barcode, v in libraries.items():
+ if v[2] in MULTIOME_PROTOCOLS or any(
+ projectDir.glob(f"Sample_{barcode}/*.fastq.gz")
+ ):
+ kept[barcode] = v
+ else:
+ log.info(
+ f"{runID}: no FASTQs for {barcode} ({v[0]}) in "
+ f"{Path(baseData, runID, f'Project_{pacified}')}, it lives on another lane."
+ )
+ return kept
+
+
def GetResults(config, project, libraries):
"""
Project is something like '352_Grzes_PearceEd' and libraries is a dictionary with libraries as keys:
@@ -1257,15 +1479,21 @@ def GetResults(config, project, libraries):
):
if org_label not in org_dict:
org_dict[org_label] = organism
- idx = validAnalysisTypes[analysisType]
- pipeline = pipelines[idx]
+ if libraryProtocol in MULTIOME_PROTOCOLS:
+ # GEX and ATAC halves of one Multiome sample go to one joint
+ # cellranger-arc run, not to the per-modality pipelines.
+ pipeline = "scMultiome"
+ groupKey = "Multiome"
+ else:
+ pipeline = pipelines[validAnalysisTypes[analysisType]]
+ groupKey = analysisType
if pipeline not in pipelineGroups:
pipelineGroups[pipeline] = {}
if org_label not in pipelineGroups[pipeline]:
pipelineGroups[pipeline][org_label] = {}
- if analysisType not in pipelineGroups[pipeline][org_label]:
- pipelineGroups[pipeline][org_label][analysisType] = []
- pipelineGroups[pipeline][org_label][analysisType].append(
+ if groupKey not in pipelineGroups[pipeline][org_label]:
+ pipelineGroups[pipeline][org_label][groupKey] = []
+ pipelineGroups[pipeline][org_label][groupKey].append(
[library, sampleName, libraryProtocol, ignore]
)
log.debug(f"Considering pipeline groups: {pipelineGroups}")
@@ -1297,18 +1525,36 @@ def GetResults(config, project, libraries):
for analysisType, tuples in v2.items():
log.debug("Queueing analysisType " + analysisType)
log.debug(tuples)
+ # One pipeline run assumes one protocol (tuples[0][2]). If an
+ # analysisType holds several (e.g. one per lane), split them
+ # and give each its own output dir via a 5th tuple element.
+ byProtocol = {}
+ for t in tuples:
+ byProtocol.setdefault(t[2], []).append(t)
+ if len(byProtocol) > 1 and pipeline != "scMultiome":
+ log.info(
+ f"{project}/{analysisType}: {len(byProtocol)} protocols, "
+ f"splitting: {sorted(byProtocol)}"
+ )
+ groups = [
+ [t + [BRB.misc.pacifier(proto)] for t in ts]
+ for proto, ts in byProtocol.items()
+ ]
+ else:
+ groups = [tuples]
# RELACS needs the unpacified project name to copy the original sample sheet to the dest dir
# hence the pacifier is applied on the project in each pipeline separately
- workItems.append(
- WorkItem(
- project=project,
- group=group,
- pipeline=pipeline,
- organism=organism,
- analysisType=analysisType,
- tuples=tuples,
+ for groupTuples in groups:
+ workItems.append(
+ WorkItem(
+ project=project,
+ group=group,
+ pipeline=pipeline,
+ organism=organism,
+ analysisType=analysisType,
+ tuples=groupTuples,
+ )
)
- )
# In case there is an external_skipList, there shouldn't be a skipList !
if external_skipList:
assert not skipList
@@ -1351,6 +1597,14 @@ def runFlowcell(config, ParkourDict, registry=None, maxWorkers=None):
f"{bdir}/Project_{BRB.misc.pacifier(k)} doesn't exist, probably lives on another lane."
)
continue
+ laneLibraries = filterLaneSamples(config, k, v)
+ if v and not laneLibraries:
+ log.info(
+ f"{bdir}/Project_{BRB.misc.pacifier(k)} has no samples of this "
+ "lane, skipping project."
+ )
+ continue
+ v = laneLibraries
projectItems, projectMsg = GetResults(config, k, v)
workItems.extend(projectItems)
msg.extend(projectMsg)
diff --git a/BRB/email.py b/BRB/email.py
index fbcf857..7ecacba 100644
--- a/BRB/email.py
+++ b/BRB/email.py
@@ -45,10 +45,11 @@ def finishedEmail(config, msg):
# Only inform deepseq if no workflow failed, and no group was
# skipped (a flowcell with a skipped group -- e.g. owned by a live
# PID from another process -- was not actually fully processed this
- # pass, so deepSeq should not be told it's ready).
+ # pass, so deepSeq should not be told it's ready), and no Multiome
+ # group needs attention (pairs waiting for another lane or unpaired).
statuses = [i[4] for i in msg]
if statuses.count("FAILED") == 0 and not any(
- s.startswith("SKIPPED") for s in statuses
+ str(s).startswith(("SKIPPED", "ATTENTION")) for s in statuses
):
recipient = config.get("Email", "deepSeq")
_html.add(
diff --git a/BRB/findFinishedFlowCells.py b/BRB/findFinishedFlowCells.py
index c7ff619..a6c312f 100644
--- a/BRB/findFinishedFlowCells.py
+++ b/BRB/findFinishedFlowCells.py
@@ -14,6 +14,11 @@ def flowCellProcessed(config):
def markFinished(config):
+ """
+ Write analysis.done marker for a processed flow cell.
+ Creates/updates analysis.done under config[Paths][baseData]/config[Options][runID]
+ to signal the flowcell has been fully processed.
+ """
_p = Path(config["Paths"]["baseData"], config["Options"]["runID"], "analysis.done")
_p.touch()
log.info(f"{_p} created, flow cell processed.")
diff --git a/BRB/run.py b/BRB/run.py
index 1e5f919..90795c9 100755
--- a/BRB/run.py
+++ b/BRB/run.py
@@ -80,6 +80,14 @@ def run_brb(configfile, sequencer):
print(errMsg, file=sys.stderr)
raise
+ # Multiome groups with pairs that could not run this pass (waiting
+ # for another lane folder, unpaired, ...): make them hard to miss.
+ for row in msg:
+ if str(row[4]).startswith("ATTENTION"):
+ warn = f"Multiome {row[0]} ({config.get('Options', 'runID')}): {row[4]}"
+ log.warning(warn)
+ print(f"[bold red]!!! {warn}[/bold red]")
+
# Email finished message
log.info("Create e-mail")
log.info(msg)
diff --git a/BigRedButton.ini b/BigRedButton.ini
index bdf708e..8b5a87e 100644
--- a/BigRedButton.ini
+++ b/BigRedButton.ini
@@ -76,6 +76,7 @@ Users=path_to_user_file
# separate pipelines specific for 10x Single Cell processing
RNA=/path/to/repo/10X_snakepipe/10x_scRNA/10x_scRNA
ATAC=/path/to/repo/10X_snakepipe/10x_scATAC/10x_scATAC
+ARC=/path/to/repo/10X_snakepipe/10x_ARC/10x_ARC
[external]
diff --git a/configs/BigRedButton_dev.ini b/configs/BigRedButton_dev.ini
new file mode 100644
index 0000000..7f3822d
--- /dev/null
+++ b/configs/BigRedButton_dev.ini
@@ -0,0 +1,105 @@
+[Paths]
+#Where full/partial flowcell demultiplexing results are stored. Copies of parts of this are delivered to groups
+baseData=/path/to/dissectBCL_results
+# groupdata = /path if PI A and pi B have folders as /path/A/sequencing_data and /path/B/sequencing_data
+groupData=/path/to/basedir/of/PI/folders
+# where sequencing facility inspects QC reports
+seqFacDir=/path/to/seqfacdir
+# where the log-files show appear
+logPath=/path/to/logdir
+
+#Groups are stored under $groupData/groupName
+groupData=/data
+bioinfoCoreDir=/path/to/bioinfocoreDir
+
+[Options]
+#How long to sleep between runs, in hours, may be fractional
+sleepTime=1
+#valid library types, separate by a comma
+validAnalysisTypes=ChIP-Seq,stranded mRNA-Seq,stranded RNA-Seq,mRNA-Seq,RNA-Seq,WGS,ATAC-Seq
+#Pipeline names, one for each of the aforementioned validAnalysisTypes. Each pipeline is actually a function in the PushButton submodule
+pipelines=DNA,RNA,RNA,RNA,RNA,DNA,ATAC
+#valid organisms, separate by a comma
+validOrganisms=human,mouse,drosophila
+#An ordered list of the organism names used in the pipeline (the order matches validOrganisms)
+organismNames=hg38,mm10,dm6
+#Snakemake workflow base directory
+snakemakeWorkflowBaseDir=/path/anaconda3/envs/snakePipes-V.x.y/
+# How many library-groups to dispatch concurrently per flowcell. Each one is a
+# local snakePipes driver; RELACS groups additionally fan out to 10 local
+# demultiplex_relacs processes. When BigRedButton is started with -s/--sequencer,
+# two processes share this host and each takes half of this value (minimum 1).
+poolSize=2
+# Path to the scancel binary used to cancel a crashed flowcell's Slurm jobs.
+# Defaults to "scancel" (resolved off PATH). On hosts where Slurm's client
+# binaries only appear on PATH after `module load slurm` (rapidus, at
+# least), run_brb won't see them: set this to the absolute path scancel
+# resolves to after that module load (e.g. `which scancel` in that shell).
+scancelBin=scancel
+#Don't set these, they're just for passing info around
+runID=
+
+[Queue]
+queueName=queue_name
+parallelProcesses=10
+
+[Parkour]
+QueryURL=http://someserver.com/api/analysis_list/analysis_list/
+ResultsURL=
+# Parkour's internal_pis endpoint: PIs whose /data directory diverges from their
+# Parkour name carry a deliver_to there (see BRB.misc.resolveDeliverTo,
+# PushButton.GetResults). Required; BRB stops with an error if unreachable.
+InternalPIsURL=http://someserver.com/api/internal_pis/
+user=foo@bar.com
+password=supersecret
+cert=path/to/cert.pem
+
+[Internals]
+# Passed as the `organizations` param to InternalPIsURL above.
+Organizations=MPI-IE
+
+[Email]
+#Lists of recipients can be comma separated
+errorTo=mustermann@ie-freiburg.mpg.de
+finishedTo=mustermann@ie-freiburg.mpg.de
+deepSeq=mustermann@ie-freiburg.mpg.de, another.example@ie-freiburg.mpg.de
+fromAddress=some.user@ie-freiburg.mpg.de
+host=localhost
+
+[Galaxy]
+URL=URL_to_galaxy_server
+API key=the_api_key
+verify=False
+Users=path_to_user_file
+
+[10x]
+# separate pipelines specific for 10x Single Cell processing
+RNA=/path/to/repo/10X_snakepipe/10x_scRNA/10x_scRNA
+ATAC=/path/to/repo/10X_snakepipe/10x_scATAC/10x_scATAC
+ARC=/path/to/repo/10X_snakepipe/10x_ARC/10x_ARC
+ARC=/path/to/repo/10X_snakepipe/10x_ARC/10x_ARC
+
+
+[external]
+# Projects that don't belong to any PI/group under $groupData (e.g. external
+# collaborators) are skipped by default, regardless of library type. The two
+# settings below opt specific libraries back in for processing anyway; their
+# output goes under $baseData//Analysis_/ instead of a PI's
+# groupData tree, since there's no group directory to put it in.
+#
+# Library types (comma separated) that should still be processed for
+# external projects. Matched against the Parkour "library type" field (see
+# validAnalysisTypes above).
+AnalysisTypes =
+# Library protocols (comma separated) that should still be processed for
+# external projects, matched via startswith() against the Parkour "library
+# protocol" field. Use this for protocol-specific opt-ins (e.g. "ChIP RELACS
+# high-throughput") without opening up every library of that type to
+# external processing.
+LibraryProtocols =
+# Where BRB fexsends the final results tarball for external projects
+# processed via AnalysisTypes/LibraryProtocols above (currently only used by
+# the RELACS pipeline).
+fexRecipient = mustermann@ie-freiburg.mpg.de
+# Path to the fexsend binary used for the above.
+fexsendBin = ~/.local/bin/fexsend
diff --git a/pyproject.toml b/pyproject.toml
index 70b1b23..52d9ef9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,6 +35,9 @@ dependencies = [
BigRedButton = "BRB.run:run_brb"
demultiplex_relacs = "BRB.demultiplex_relacs:main"
+
+[tool.setuptools]
+packages = ["BRB"]
[tool.setuptools_scm]
[tool.ruff]
diff --git a/tests/test_PushButton.py b/tests/test_PushButton.py
index 51c210c..5b6fd1e 100644
--- a/tests/test_PushButton.py
+++ b/tests/test_PushButton.py
@@ -637,9 +637,9 @@ def test_no_check_call_remains(self):
source = Path(PushButton.__file__).read_text()
assert "subprocess.check_call" not in source
- def test_all_twelve_sites_converted(self):
+ def test_all_thirteen_sites_converted(self):
source = Path(PushButton.__file__).read_text()
- assert source.count("runManagedSubprocess(") == 12
+ assert source.count("runManagedSubprocess(") == 13
def test_the_two_cwd_sites_keep_their_cwd(self):
source = Path(PushButton.__file__).read_text()
diff --git a/tests/test_PushButton_lanesplit.py b/tests/test_PushButton_lanesplit.py
new file mode 100644
index 0000000..44c6e43
--- /dev/null
+++ b/tests/test_PushButton_lanesplit.py
@@ -0,0 +1,447 @@
+"""
+Lane-split flowcells (dissectBCL --forcelanesplit): one project, one pool per
+lane, demultiplexed into _lanes_1 and _lanes_2. All names
+below are fake; protocol strings are the Parkour protocol names from the
+parkour2 fixtures (library_sample_shared/fixtures/libraryprotocol.json).
+"""
+
+import configparser
+
+import pytest
+
+import BRB.ET
+from BRB import PushButton
+
+HUMAN = ("human", "hg38", "/yaml/hg38.yaml")
+FLOWCELL = "261028_A00931_0938_BXXXXXXXX"
+PROJECT = "9999_Foo_Bar"
+GEX = "Chromium_NextGEM_SingleCell3Prime_GeneExpression_v3.1_DualIndex"
+ATAC = "Chromium Next GEM Single Cell ATAC v2"
+
+# barcode -> [name, analysisType, protocol, organism, indexType, depth]
+PARKOUR = {
+ PROJECT: {
+ "99L001": ["gex_a", "RNA-Seq single cell", GEX, HUMAN, "i7", 30],
+ "99L002": ["gex_b", "RNA-Seq single cell", GEX, HUMAN, "i7", 30],
+ "99L003": ["atac_a", "ATAC-Seq single cell", ATAC, HUMAN, "i7", 30],
+ "99L004": ["atac_b", "ATAC-Seq single cell", ATAC, HUMAN, "i7", 30],
+ }
+}
+LANE1 = {"99L001": "gex_a", "99L002": "gex_b"}
+LANE2 = {"99L003": "atac_a", "99L004": "atac_b"}
+
+
+def laneConfig(tmp_path, lane):
+ config = configparser.ConfigParser()
+ config["Paths"] = {
+ "baseData": str(tmp_path / "base"),
+ # groupData/ is absent -> GetResults ignore=True, so the
+ # [external] allowlist keeps the libraries and output is under baseData.
+ "groupData": str(tmp_path / "group"),
+ "seqFacDir": str(tmp_path / "seqfac"),
+ "bioinfoCoreDir": str(tmp_path / "bioinfocore"),
+ }
+ config["Options"] = {
+ "runID": f"{FLOWCELL}_lanes_{lane}",
+ "sequencerType": "Illumina",
+ "validAnalysisTypes": "RNA-Seq single cell,ATAC-Seq single cell",
+ "pipelines": "scRNAseq,scATAC",
+ "snakemakeWorkflowBaseDir": "/snakepipes",
+ }
+ config["external"] = {
+ "AnalysisTypes": "RNA-Seq single cell,ATAC-Seq single cell",
+ "LibraryProtocols": "",
+ }
+ config["10x"] = {"RNA": "/tenx/rna_wf", "ATAC": "/tenx/atac_wf"}
+ (tmp_path / "group").mkdir(parents=True, exist_ok=True)
+ (tmp_path / "bioinfocore").mkdir(parents=True, exist_ok=True)
+ return config
+
+
+def makeLane(tmp_path, lane, samples, suffixes):
+ """Fake dissectBCL output: _lanes_/Project_*/Sample_/_.fastq.gz"""
+ project = tmp_path / "base" / f"{FLOWCELL}_lanes_{lane}" / f"Project_{PROJECT}"
+ for barcode, name in samples.items():
+ sampleDir = project / f"Sample_{barcode}"
+ sampleDir.mkdir(parents=True)
+ for sfx in suffixes:
+ (sampleDir / f"{name}_{sfx}.fastq.gz").touch()
+
+
+@pytest.fixture
+def flowcell(tmp_path, monkeypatch):
+ makeLane(tmp_path, 1, LANE1, ["RNA_R1", "RNA_R2", "RNA_R1_optical_duplicates"])
+ makeLane(tmp_path, 2, LANE2, ["ATAC_I1", "ATAC_I2", "ATAC_R1", "ATAC_R2"])
+ monkeypatch.setattr(PushButton, "POOL_SIZE", 1)
+ monkeypatch.setattr(
+ BRB.ET,
+ "phoneHome",
+ lambda config, outputDir, pipeline, tuples, org, label, project, libType: [
+ project,
+ org,
+ libType,
+ pipeline,
+ "success",
+ "PARKOUR_OK",
+ ],
+ )
+ for fn in ("copyCellRanger", "tidyUpABit", "stripRights", "touchDone"):
+ monkeypatch.setattr(PushButton, fn, lambda *a, **k: None)
+ return tmp_path
+
+
+def recordDispatch(monkeypatch):
+ """Record (pipeline, [sample names]) per dispatch, without running anything."""
+ seen = []
+
+ def make(pipeline):
+ def stub(config, group, project, organism, analysisType, tuples):
+ seen.append((pipeline, [t[1] for t in tuples]))
+ return "/out", 0, False
+
+ return stub
+
+ monkeypatch.setattr(PushButton, "scRNAseq", make("scRNAseq"))
+ monkeypatch.setattr(PushButton, "scATAC", make("scATAC"))
+ return seen
+
+
+class TestLaneFilter:
+ def test_each_lane_dispatches_only_its_own_samples(self, flowcell, monkeypatch):
+ seen = recordDispatch(monkeypatch)
+
+ msg1 = PushButton.runFlowcell(laneConfig(flowcell, 1), PARKOUR)
+ msg2 = PushButton.runFlowcell(laneConfig(flowcell, 2), PARKOUR)
+
+ assert [s for p, s in seen if p == "scRNAseq"] == [["gex_a", "gex_b"]]
+ assert [s for p, s in seen if p == "scATAC"] == [["atac_a", "atac_b"]]
+ assert [m[3] for m in msg1] == ["scRNAseq"]
+ assert [m[3] for m in msg2] == ["scATAC"]
+
+ def test_project_without_samples_in_this_lane_is_skipped_and_logged(
+ self, flowcell, monkeypatch, caplog
+ ):
+ # Project dir exists in lane 2, but none of lane-1's samples are in it.
+ seen = recordDispatch(monkeypatch)
+ lane1Only = {PROJECT: {k: v for k, v in PARKOUR[PROJECT].items() if k in LANE1}}
+ caplog.set_level("INFO")
+
+ msg = PushButton.runFlowcell(laneConfig(flowcell, 2), lane1Only)
+
+ assert msg == []
+ assert seen == []
+ assert f"Project_{PROJECT}" in caplog.text
+ assert "no samples" in caplog.text.lower()
+
+ def test_scatac_gets_only_lane_samples_in_samples_arg(self, flowcell, monkeypatch):
+ """Real scATAC: the --samples argument must not name the other lane's samples."""
+ cmds = []
+ monkeypatch.setattr(
+ PushButton, "runManagedSubprocess", lambda cmd, **k: cmds.append(cmd)
+ )
+
+ PushButton.runFlowcell(laneConfig(flowcell, 2), PARKOUR)
+
+ assert len(cmds) == 1
+ samples = cmds[0].split(" --samples ")[1].split(" --snakemakePath")[0]
+ assert samples.split() == ["atac_a", "atac_b"]
+
+
+class TestProtocolGrouping:
+ def test_mixed_protocols_in_one_analysis_type_get_separate_work_items(
+ self, tmp_path
+ ):
+ config = laneConfig(tmp_path, 1)
+ other = "Chromium_GEM-X_SingleCell_3primeRNA-seq_v4"
+ libraries = {
+ "99L001": ["a", "RNA-Seq single cell", GEX, HUMAN, "i7", 30],
+ "99L002": ["b", "RNA-Seq single cell", other, HUMAN, "i7", 30],
+ }
+
+ items, _ = PushButton.GetResults(config, PROJECT, libraries)
+
+ for item in items:
+ assert len({t[2] for t in item.tuples}) == 1
+
+
+MULTI_GEX = "Chromium_NextGEM_Multiome_GEX"
+MULTI_ATAC = "Chromium_NextGEM_Multiome_ATAC"
+# Real Multiome names: _RNA / _ATAC, FASTQs named the same way.
+MULTIOME = {
+ PROJECT: {
+ "99L001": ["con_RNA", "RNA-Seq single cell", MULTI_GEX, HUMAN, "i7", 30],
+ "99L002": ["ko_RNA", "RNA-Seq single cell", MULTI_GEX, HUMAN, "i7", 30],
+ "99L003": ["con_ATAC", "ATAC-Seq single cell", MULTI_ATAC, HUMAN, "i7", 30],
+ "99L004": ["ko_ATAC", "ATAC-Seq single cell", MULTI_ATAC, HUMAN, "i7", 30],
+ }
+}
+
+
+@pytest.fixture
+def multiome(tmp_path, monkeypatch):
+ makeLane(tmp_path, 1, {"99L001": "con_RNA", "99L002": "ko_RNA"}, ["R1", "R2"])
+ makeLane(
+ tmp_path,
+ 2,
+ {"99L003": "con_ATAC", "99L004": "ko_ATAC"},
+ ["I1", "I2", "R1", "R2"],
+ )
+ for lane in (1, 2):
+ (tmp_path / "base" / f"{FLOWCELL}_lanes_{lane}" / "fastq.made").touch()
+ monkeypatch.setattr(PushButton, "POOL_SIZE", 1)
+ monkeypatch.setattr(
+ BRB.ET,
+ "phoneHome",
+ lambda config, outputDir, pipeline, tuples, org, label, project, libType: [
+ project,
+ org,
+ libType,
+ pipeline,
+ "success",
+ "PARKOUR_OK",
+ ],
+ )
+ for fn in ("copyCellRanger", "tidyUpABit", "stripRights"):
+ monkeypatch.setattr(PushButton, fn, lambda *a, **k: None)
+ cmds = []
+ monkeypatch.setattr(
+ PushButton, "runManagedSubprocess", lambda cmd, **k: cmds.append(cmd)
+ )
+ return tmp_path, cmds
+
+
+def multiomeConfig(tmp_path, lane):
+ config = laneConfig(tmp_path, lane)
+ config["10x"]["ARC"] = "/tenx/arc_wf"
+ return config
+
+
+class TestMultiome:
+ def test_multiome_protocols_route_to_scMultiome_not_per_lane_pipelines(
+ self, tmp_path
+ ):
+ items, _ = PushButton.GetResults(
+ multiomeConfig(tmp_path, 2), PROJECT, MULTIOME[PROJECT]
+ )
+
+ assert {i.pipeline for i in items} == {"scMultiome"}
+
+ def test_first_lane_with_both_halves_ready_runs_one_joint_arc(self, multiome):
+ tmp_path, cmds = multiome
+
+ msg = PushButton.runFlowcell(multiomeConfig(tmp_path, 1), MULTIOME)
+
+ assert len(cmds) == 1
+ cmd = cmds[0]
+ assert cmd.startswith("/tenx/arc_wf ")
+ assert (
+ f"--gex-indir {tmp_path}/base/{FLOWCELL}_lanes_1/Project_{PROJECT} " in cmd
+ )
+ assert (
+ f"--atac-indir {tmp_path}/base/{FLOWCELL}_lanes_2/Project_{PROJECT} " in cmd
+ )
+ assert "--pair con:con_RNA:con_ATAC" in cmd
+ assert "--pair ko:ko_RNA:ko_ATAC" in cmd
+ # Output lives under the lane folder that ran it.
+ assert f"-o {tmp_path}/base/{FLOWCELL}_lanes_1/Analysis_{PROJECT}/" in cmd
+ assert [m[3] for m in msg] == ["scMultiome"]
+ assert list((tmp_path / "base" / f"{FLOWCELL}_lanes_1").rglob("analysis.done"))
+
+ def test_pairs_done_in_a_sibling_lane_are_not_rerun(self, multiome):
+ tmp_path, cmds = multiome
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 1), MULTIOME)
+ lane1Out = next(
+ (tmp_path / "base" / f"{FLOWCELL}_lanes_1").glob("Analysis_*/*")
+ )
+ for base in ("con", "ko"): # what the 10x_ARC wrapper writes per pair
+ (lane1Out / f"{base}.done").touch()
+ cmds.clear()
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), MULTIOME)
+
+ assert cmds == []
+
+ def test_lane_processed_before_its_partner_is_demultiplexed_hands_over(
+ self, multiome
+ ):
+ # BRB may pick up lanes_2 while lanes_1 is still being demultiplexed
+ # (no fastq.made). run.py then marks lanes_2 analysis.done, so the
+ # joint run must happen in lanes_1's later pass.
+ tmp_path, cmds = multiome
+ (tmp_path / "base" / f"{FLOWCELL}_lanes_1" / "fastq.made").unlink()
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), MULTIOME)
+
+ assert cmds == []
+ assert not list(
+ (tmp_path / "base" / f"{FLOWCELL}_lanes_2").rglob("analysis.done")
+ )
+
+ (tmp_path / "base" / f"{FLOWCELL}_lanes_1" / "fastq.made").touch()
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 1), MULTIOME)
+
+ assert len(cmds) == 1
+ assert f"-o {tmp_path}/base/{FLOWCELL}_lanes_1/Analysis_{PROJECT}/" in cmds[0]
+
+ def test_unpaired_multiome_sample_is_not_run(self, multiome):
+ tmp_path, cmds = multiome
+ # ko has no GEX library in Parkour: only con can be paired.
+ partial = {
+ PROJECT: {k: v for k, v in MULTIOME[PROJECT].items() if k != "99L002"}
+ }
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), partial)
+
+ assert len(cmds) == 1
+ assert "--pair con:con_RNA:con_ATAC" in cmds[0]
+ assert "ko" not in cmds[0].split("--pair ", 1)[1]
+
+
+def laneDir(tmp_path, lane):
+ return tmp_path / "base" / f"{FLOWCELL}_lanes_{lane}"
+
+
+def note(tmp_path, lane):
+ notes = list(laneDir(tmp_path, lane).rglob(PushButton.MULTIOME_NOTE))
+ return notes[0].read_text() if notes else None
+
+
+class TestMultiomeEdgeCases:
+ def test_partner_lane_not_demultiplexed_at_all_waits_and_flags(
+ self, tmp_path, monkeypatch
+ ):
+ # lanes_2 picked up while dissectBCL has not written lanes_1 yet:
+ # the GEX libraries must not be dropped as "on another lane".
+ makeLane(tmp_path, 2, {"99L003": "con_ATAC"}, ["I1", "I2", "R1", "R2"])
+ (laneDir(tmp_path, 2) / "fastq.made").touch()
+ monkeypatch.setattr(PushButton, "POOL_SIZE", 1)
+ monkeypatch.setattr(
+ BRB.ET,
+ "phoneHome",
+ lambda c, o, p, t, org, lab, proj, lt: [proj, org, lt, p, "success", "OK"],
+ )
+ cmds = []
+ monkeypatch.setattr(
+ PushButton, "runManagedSubprocess", lambda c, **k: cmds.append(c)
+ )
+ lib = {k: MULTIOME[PROJECT][k] for k in ("99L001", "99L003")}
+
+ msg = PushButton.runFlowcell(multiomeConfig(tmp_path, 2), {PROJECT: lib})
+
+ assert cmds == []
+ assert not list(laneDir(tmp_path, 2).rglob("analysis.done"))
+ assert "con: waiting" in note(tmp_path, 2)
+ assert msg[0][4].startswith("ATTENTION")
+
+ def test_unpaired_library_flags_attention_but_group_is_done(self, multiome):
+ tmp_path, _cmds = multiome
+ partial = {
+ PROJECT: {k: v for k, v in MULTIOME[PROJECT].items() if k != "99L002"}
+ }
+
+ msg = PushButton.runFlowcell(multiomeConfig(tmp_path, 2), partial)
+
+ assert "ko: ATAC library 99L004 has no GEX partner" in note(tmp_path, 2)
+ assert msg[0][4].startswith("ATTENTION")
+ # Nothing will ever complete ko by itself: do not hold the group open.
+ assert list(laneDir(tmp_path, 2).rglob("analysis.done"))
+
+ def test_clean_pass_removes_a_stale_note(self, multiome):
+ tmp_path, _cmds = multiome
+ partial = {
+ PROJECT: {k: v for k, v in MULTIOME[PROJECT].items() if k != "99L002"}
+ }
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), partial)
+ for done in laneDir(tmp_path, 2).rglob("analysis.done"):
+ done.unlink() # operator forces a rerun after fixing Parkour
+
+ msg = PushButton.runFlowcell(multiomeConfig(tmp_path, 2), MULTIOME)
+
+ assert note(tmp_path, 2) is None
+ assert msg[0][4] == "success"
+
+ def test_half_with_fastqs_in_two_lanes_is_not_run(self, multiome):
+ tmp_path, cmds = multiome
+ # con_ATAC also sequenced on lane 1 (pool loaded on both lanes).
+ makeLane(tmp_path, 1, {"99L003": "con_ATAC"}, ["I1", "I2", "R1", "R2"])
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), MULTIOME)
+
+ assert len(cmds) == 1
+ assert "--pair ko:ko_RNA:ko_ATAC" in cmds[0]
+ assert "--pair con:" not in cmds[0]
+ assert "con: ATAC 99L003 has FASTQs in 2 lane folders" in note(tmp_path, 2)
+
+ def test_duplicate_base_name_is_not_run(self, multiome):
+ tmp_path, cmds = multiome
+ makeLane(tmp_path, 1, {"99L005": "con_RNA_b"}, ["R1", "R2"])
+ dup = {PROJECT: dict(MULTIOME[PROJECT])}
+ dup[PROJECT]["99L005"] = [
+ "con_RNA",
+ "RNA-Seq single cell",
+ MULTI_GEX,
+ HUMAN,
+ "i7",
+ 30,
+ ]
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), dup)
+
+ assert "--pair con:" not in cmds[0]
+ assert "con: two RNA libraries in Parkour (99L001, 99L005)" in note(tmp_path, 2)
+
+ def test_fastq_names_are_taken_from_disk_not_parkour(self, multiome):
+ # dissectBCL strips umlauts from FASTQ names; the wrapper searches
+ # for the on-disk name, so Parkour's spelling must not leak through.
+ tmp_path, cmds = multiome
+ umlaut = {PROJECT: dict(MULTIOME[PROJECT])}
+ umlaut[PROJECT]["99L001"] = [
+ "cön_RNA",
+ "RNA-Seq single cell",
+ MULTI_GEX,
+ HUMAN,
+ "i7",
+ 30,
+ ]
+ umlaut[PROJECT]["99L003"] = [
+ "cön_ATAC",
+ "ATAC-Seq single cell",
+ MULTI_ATAC,
+ HUMAN,
+ "i7",
+ 30,
+ ]
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), umlaut)
+
+ assert ":con_RNA:con_ATAC" in cmds[0]
+
+ def test_genome_arg_is_the_full_yaml_path_passed_to_wrapper(self, multiome):
+ # Parkour's organism "yaml" is a snakePipes yaml path in production.
+ # The 10x_ARC wrapper (10X_snakepipe#21) now handles genome normalization,
+ # so BRB passes the full org_yaml path directly to the wrapper.
+ tmp_path, cmds = multiome
+ prod = {
+ PROJECT: {
+ bc: [
+ v[0],
+ v[1],
+ v[2],
+ (
+ "human",
+ "hg38",
+ "/data/repository/organisms/hg38/snakePipes_organism.yaml",
+ ),
+ v[4],
+ v[5],
+ ]
+ for bc, v in MULTIOME[PROJECT].items()
+ }
+ }
+
+ PushButton.runFlowcell(multiomeConfig(tmp_path, 2), prod)
+
+ assert (
+ "/data/repository/organisms/hg38/snakePipes_organism.yaml --snakemakePath"
+ in cmds[0]
+ )
diff --git a/tests/test_email.py b/tests/test_email.py
index 39d5059..7c15fe7 100644
--- a/tests/test_email.py
+++ b/tests/test_email.py
@@ -109,6 +109,31 @@ def test_skipped_entry_suppresses_deepseq_notification(self, mockSMTP):
recipients = mockInstance.sendmail.call_args[0][1]
assert recipients == ["bioinfocore@example.org"]
+ @patch("BRB.email.smtplib.SMTP")
+ def test_multiome_attention_entry_suppresses_deepseq_notification(self, mockSMTP):
+ # A Multiome group with pairs waiting for another lane folder (or
+ # unpaired) is not finished either.
+ config = make_config()
+ mockInstance = MagicMock()
+ mockSMTP.return_value = mockInstance
+ msg = [
+ [
+ "9999_Foo_Bar",
+ "human",
+ "RNA-Seq single cell",
+ "scMultiome",
+ "ATTENTION (see /x/multiome.attention)",
+ "PARKOUR_OK",
+ True,
+ 0,
+ ],
+ ]
+
+ email.finishedEmail(config, msg)
+
+ recipients = mockInstance.sendmail.call_args[0][1]
+ assert recipients == ["bioinfocore@example.org"]
+
@patch("BRB.email.smtplib.SMTP")
def test_no_skipped_or_failed_entries_still_notifies_deepseq(self, mockSMTP):
config = make_config()