smftools.preprocessing.partitioned_executor

Execute bounded preprocessing work units into derived partition stores.

Functions

append_modification_qc_mask(obs_path, cfg)	Apply legacy modification thresholds to reduced metrics as a mask.
execute_partitioned_preprocessing(...[, tasks])	Execute planned tasks sequentially and publish catalogs plus a derived spine.
execute_preprocess_task(spine_path, task, ...)	Materialize, transform, core-crop, and write one preprocessing task.
fit_direct_modality_youden_thresholds(...)	Fit per-position Youden methylation thresholds once per reference.
reduce_duplicate_reads(...)	Run bounded duplicate comparisons and reconcile clusters across cores.
reduce_partial_coverage(catalog_path, spine, ...)	Reduce task-local coverage counts into one row per genomic position.
reduce_read_modification_stats(catalog_path, ...)	Reduce non-overlapping core partials into legacy-compatible read metrics.
write_read_qc_sidecar(spine, cfg, output_path)	Compute read-level QC as an aligned mask without deleting spine rows.

smftools.preprocessing.partitioned_executor.fit_direct_modality_youden_thresholds(spine_path, cfg, references, output_dir)

Fit per-position Youden methylation thresholds once per reference.

Threshold fitting (calculate_position_Youden) needs the whole control-sample population covering a position at once -- it can't run per bounded task the way the rest of partitioned preprocessing does. This runs as a pre-pass before task dispatch, reusing the same per-reference windowing (reference_windows) that task planning uses: for locus references that's a single window spanning the whole reference (small, so no tiling needed); for genome-mode references it's the same haloed tiles the rest of partitioned preprocessing already bounds memory with, since a read only ever covers a small local span of a chromosome-scale reference. Each window's fit keeps only its own core range (the halo is covered by an adjacent window) before results from all windows for a reference are concatenated into one lookup table.

Return type:

dict[str, DataFrame]

Returns:

Mapping of reference -> DataFrame indexed by genomic position (int) with columns threshold, j_statistic, passed_qc.

smftools.preprocessing.partitioned_executor.execute_preprocess_task(spine_path, task, cfg, output_dir, *, youden_thresholds=None)

Materialize, transform, core-crop, and write one preprocessing task.

Layers are streamed to disk (incremental_zarr.append_zarr_layer) as soon as each is computed, then freed, instead of being collected in memory and written together at the end -- see dev/pipeline_scaling_audit.md and the plan that implemented this. Ordering is modality-aware: for direct, the binarized layer is both a reportable derived layer and clean_NaN's read source, so it can't be written+freed until after clean_NaN returns.

Return type:

dict[str, object]

smftools.preprocessing.partitioned_executor.reduce_partial_coverage(catalog_path, spine, output_path, *, minimum_valid_fraction)

Reduce task-local coverage counts into one row per genomic position.

Return type:

Path

smftools.preprocessing.partitioned_executor.write_read_qc_sidecar(spine, cfg, output_path)

Compute read-level QC as an aligned mask without deleting spine rows.

Entry point of the preprocess-stage QC/dedup chain. The full chain, in execution order, is: :rtype: Path

1. write_read_qc_sidecar (this function) -> obs["passes_read_qc"]: length/quality/mapping thresholds (filter_reads_on_length_quality_mapping: read_len_filter_thresholds, mapped_len_filter_thresholds, ratio thresholds, read_quality_filter_thresholds, read_mapping_quality_filter_thresholds) AND CIGAR indel size limits (filter_reads_on_cigar_indels: max_internal_insertion_length, max_internal_deletion_length). For deaminase modality, reads labeled as PCR chimeras (deaminase_chimera_mask: ct_event_count, ga_event_count, strand_segment_purity thresholded by deaminase_chimera_min_events_per_span, deaminase_chimera_min_segment_purity, deaminase_chimera_max_single_strand_fraction) are ANDed out of passes_read_qc too, and recorded separately on obs["deaminase_PCR_chimera"] for inspection.

2. append_modification_qc_mask -> obs["passes_modification_qc"] (filter_reads_on_modification_thresholds: per-site-type modification fraction thresholds plus min_valid_fraction_positions_in_read_vs_ref) and obs["passes_qc"] = passes_read_qc & passes_modification_qc.

3. reduce_duplicate_reads -> obs["passes_dedup"] = passes_qc & ~is_duplicate (windowed Hamming-distance duplicate detection over duplicate_detection_site_types, gated by bypass_flag_duplicate_reads).

Downstream, execute_partitioned_spatial/execute_partitioned_hmm both select next(col for col in ("passes_dedup", "passes_qc") if col in spine.obs) as plan_preprocess_tasks's filter_mask, so only reads passing the full chain (or passes_qc alone, if dedup hasn't run) are ever included in a task's read_ids -- reads failing any step never reach spatial/HMM materialization, not merely get flagged.

smftools.preprocessing.partitioned_executor.reduce_read_modification_stats(catalog_path, var_path, obs_path)

Reduce non-overlapping core partials into legacy-compatible read metrics.

Return type:

Path

smftools.preprocessing.partitioned_executor.append_modification_qc_mask(obs_path, cfg)

Apply legacy modification thresholds to reduced metrics as a mask.

Return type:

Path

smftools.preprocessing.partitioned_executor.reduce_duplicate_reads(preprocess_spine_path, obs_path, cfg)

Run bounded duplicate comparisons and reconcile clusters across cores.

Return type:

Path

smftools.preprocessing.partitioned_executor.execute_partitioned_preprocessing(spine_path, cfg, output_dir, *, tasks=None)

Execute planned tasks sequentially and publish catalogs plus a derived spine.

Return type:

dict[str, Path]