If you want, you can always perform a similar job type manually - just split the dataset in half using particle sets, run ab initio on one half, then (after confirming that it has the right sym axis assignment), clone & run with the other half of the particles, but the same random seed.
You can then use Align 3D maps with update particle orientations to align the two maps & particle sets, and use a CS tools script (below) to assign these as the half-sets of a new particle set (which you can then use for homogeneous reconstruction & local refinement).
Script:
#!/usr/bin/env python3
# Combine two particle sets and enforce half-sets via alignments3D/split (A->0, B->1).
# Works with CryoSPARC v4.7x and v5. Minimal, dtype-agnostic (reuses input dtype).
from pathlib import Path
import json
import argparse
import numpy as np
try:
from cryosparc.tools import CryoSPARC, Dataset, APIError
except Exception:
from cryosparc.tools import CryoSPARC, Dataset
try:
from cryosparc.errors import APIError
except Exception:
class APIError(Exception): pass
CANDIDATE_OUTPUTS = ("particles", "particles_selected", "particles_keep", "particles_out", "particles_split")
WANTED_PREFIXES = ("blob/", "ctf/", "alignments3D/")
def autodetect_particles(job):
for nm in CANDIDATE_OUTPUTS:
try:
ds = job.load_output(nm)
if ds and len(ds) and "blob/path" in ds.fields():
return nm, ds
except Exception:
pass
raise RuntimeError(f"{job.uid}: no usable particle output found.")
def trailing_shape(arr):
try: return arr.shape[1:]
except Exception: return ()
def empty_like(ref, n):
tsh = trailing_shape(ref)
dt = getattr(ref, "dtype", object)
if hasattr(dt, "kind") and dt.kind in "iu":
return np.zeros((n,)+tsh, dtype=dt)
if hasattr(dt, "kind") and dt.kind == "f":
return np.zeros((n,)+tsh, dtype=dt)
out = np.empty((n,)+tsh, dtype=object); out[...] = None
return out
def empty_for(field, refds, n):
return empty_like(refds[field], n) if field in refds.fields() else np.array([None]*n, dtype=object)
def counts(arr):
u, c = np.unique(arr, return_counts=True)
return {int(k): int(v) for k, v in zip(u, c)}
# ---------------- CLI ----------------
p = argparse.ArgumentParser(description="Combine two CryoSPARC particle sets and enforce half-sets via alignments3D/split.")
p.add_argument("--project","-p", required=True)
p.add_argument("--workspace","-w", required=True)
p.add_argument("--jobA","-a", required=True, help="First job (becomes half-set 0)")
p.add_argument("--jobB","-b", required=True, help="Second job (becomes half-set 1)")
p.add_argument("--title","-t", default=None)
p.add_argument("--instance-info","-i", default=str(Path.home()/ "instance_info.json"))
args = p.parse_args()
print(">> combine_halfsets_min starting...")
# Connect
cfg = json.loads(Path(args.instance_info).read_text())
cs = CryoSPARC(**cfg)
if not cs.test_connection(): raise RuntimeError("Could not connect to CryoSPARC")
project = cs.find_project(args.project) or (_ for _ in ()).throw(RuntimeError(f"Project not found: {args.project}"))
jobA = project.find_job(args.jobA) or (_ for _ in ()).throw(RuntimeError(f"Job not found: {args.jobA}"))
jobB = project.find_job(args.jobB) or (_ for _ in ()).throw(RuntimeError(f"Job not found: {args.jobB}"))
# Load inputs
outA_name, A = autodetect_particles(jobA)
outB_name, B = autodetect_particles(jobB)
nA, nB = len(A), len(B)
print(f"Loaded: {jobA.uid}:{outA_name} n={nA:,} | {jobB.uid}:{outB_name} n={nB:,}")
if nA == 0 or nB == 0: raise RuntimeError("One input is empty.")
fA, fB = set(A.fields()), set(B.fields())
# Sanity fields
req = {"blob/path", "blob/psize_A", "alignments3D/pose", "alignments3D/shift"}
missA, missB = req - fA, req - fB
if missA or missB:
raise RuntimeError(f"Missing required fields. A:{sorted(missA)} B:{sorted(missB)}")
# Strict pixel-size equality
uA = np.unique(np.asarray(A["blob/psize_A"]).ravel())
uB = np.unique(np.asarray(B["blob/psize_A"]).ravel())
if len(uA) != 1: raise RuntimeError(f"Set A has mixed pixel sizes: {uA}")
if len(uB) != 1: raise RuntimeError(f"Set B has mixed pixel sizes: {uB}")
if uA[0] != uB[0]: raise RuntimeError(f"Pixel sizes differ (A:{uA[0]}, B:{uB[0]}). Re-extract/rebin to match.")
# Require alignments3D/split in at least one input
if "alignments3D/split" not in (fA | fB):
raise RuntimeError("Neither input has 'alignments3D/split'. On v4.x this column is required for the gold-standard split.")
split_dtype = (A["alignments3D/split"].dtype if "alignments3D/split" in fA else B["alignments3D/split"].dtype)
# Build combined table with only wanted groups and compatible trailing shapes
keepA = [f for f in fA if f.startswith(WANTED_PREFIXES)]
keepB = [f for f in fB if f.startswith(WANTED_PREFIXES)]
union = sorted(set(keepA).union(keepB))
# Ensure blob/idx exists on both sides
idxA = A["blob/idx"] if "blob/idx" in fA else np.zeros(nA, dtype=np.int64)
idxB = B["blob/idx"] if "blob/idx" in fB else np.zeros(nB, dtype=np.int64)
cols = {}
for f in union:
colA = A[f] if f in fA else empty_for(f, B, nA)
colB = B[f] if f in fB else empty_for(f, A, nB)
if f == "blob/idx": colA, colB = idxA, idxB
if trailing_shape(colA) != trailing_shape(colB):
continue
cols[f] = np.concatenate([np.asarray(colA), np.asarray(colB)], axis=0)
# Overwrite alignments3D/split to A->0, B->1 using the existing dtype
N = nA + nB
new_split = np.empty(N, dtype=split_dtype)
new_split[:nA] = 0
new_split[nA:] = 1
cols["alignments3D/split"] = new_split
combined = Dataset(sorted(cols.items(), key=lambda kv: kv[0]))
report = counts(combined["alignments3D/split"])
print(f"alignments3D/split (pre-save): {report} (expected 0:{nA}, 1:{nB})")
# External job
title = args.title or f"Combined half-sets from {jobA.uid} ({outA_name}) + {jobB.uid} ({outB_name})"
ej = project.create_external_job(workspace_uid=args.workspace, title=title)
try:
major = int(str(cs.version()).lstrip("v").split(".")[0])
except Exception:
major = 4
if major >= 5:
in_slots = [{"name":"blob","dtype":"particle_blob"},
{"name":"ctf","dtype":"ctf"},
{"name":"alignments3D","dtype":"alignments3D"}]
out_slots = [{"name":"blob","dtype":"particle_blob"},
{"name":"ctf","dtype":"ctf"},
{"name":"alignments3D","dtype":"alignments3D"}]
else:
in_slots, out_slots = ["blob","ctf","alignments3D"], ["blob","ctf","alignments3D"]
ej.add_input(type="particle", name="input_particles_A", slots=in_slots)
ej.add_input(type="particle", name="input_particles_B", slots=in_slots)
ej.connect(target_input="input_particles_A", source_job_uid=jobA.uid, source_output=outA_name)
ej.connect(target_input="input_particles_B", source_job_uid=jobB.uid, source_output=outB_name)
ej.add_output(type="particle", name="particles", slots=out_slots, alloc=combined)
with ej.run():
ej.log(f"Overwrote alignments3D/split → A:0 (n={nA:,}), B:1 (n={nB:,})")
ej.log(f"alignments3D/split counts (pre-save): {report}")
ej.save_output("particles", combined)
ej.stop()
# Verify
try:
saved = project.find_job(ej.uid).load_output("particles")
post = counts(saved["alignments3D/split"])
print(f"[VERIFY] alignments3D/split (post-save): {post}")
except Exception as e:
print(f"[VERIFY] reload failed: {e}")
print(f"Created External Job: '{title}'")
print("Done.")
(this should make an external job with both sets combined as half-sets of a single particle output)
