spacr.spacrops
==============

.. py:module:: spacr.spacrops






Module Contents
---------------

.. py:class:: spacrStitcher(detector: str = 'ORB', nfeatures: int = 6000, max_keypoints: Optional[int] = 2000, downsample: float = 0.5, ransac_thresh_px: float = 3.0, allow_scale: bool = False, allow_rotation: bool = False, outline_source: str = 'otsu', canny: Tuple[int, int] = (40, 120), blur_sigma: float = 0.0, dilate_ksize: int = 0, line_thickness: int = 1, outline_alpha: float = 1.0, outdir: str = './sbs_out', save_qc: bool = True, save_stitched_default: bool = True, all_scores: bool = False, score_threshold: Optional[float] = None, verbose: bool = False, feature_cache_mode: str = 'disk', feature_cache_dir: Optional[str] = None, max_ram_features: int = 256, n_workers_features: Optional[int] = None, pair_batch_size: int = 8000, stream_csv: bool = True, opencv_threads: int = 1, arr_axes: str = 'AUTO', mip: bool = False, z_index: int = 0, t_index: int = 0, squeeze_singleton: bool = True)

   Pairwise stitcher with downsampled scoring and whole-well mosaic assembly.

   Robustness features for very large datasets:
     - feature_cache_mode: "disk" (default) writes DS features to disk with an LRU RAM cap
     - pair_batch_size:    limit number of concurrent pair futures
     - stream_csv:         write pairwise rows immediately (low RAM)
     - opencv_threads:     limit OpenCV internal threading to avoid oversubscription

   Transform control
   -----------------
     allow_scale:    if True  → use RANSAC affine (rotation+scale+translation)
                     if False → enforce unit scale (see allow_rotation)
     allow_rotation: if True  → rotation+translation
                     if False → translation-only

   Mosaic
   ------
     Uses topology-aware pruning + maximum spanning tree on pairwise scores
     to build a cohesive grid. Exports mosaic image and a mosaic.csv manifest.

   Axis & Z handling
   -----------------
     arr_axes : str in {"AUTO"} or a string over {T,C,Z,Y,X} (e.g. "CZYX", "CYX", "ZYX").
                Determines how to interpret multi-dimensional TIFFs.
     mip      : bool. If True and Z exists, perform max-intensity projection over Z.
     z_index  : int  (if mip=False) choose Z slice index.
     t_index  : int  choose time index if T exists.


   .. py:attribute:: detector
      :value: ''



   .. py:attribute:: nfeatures
      :value: 6000



   .. py:attribute:: max_keypoints
      :value: None



   .. py:attribute:: downsample


   .. py:attribute:: ransac_thresh_px


   .. py:attribute:: allow_scale
      :value: False



   .. py:attribute:: allow_rotation
      :value: False



   .. py:attribute:: outline_source
      :value: ''



   .. py:attribute:: canny
      :value: (40, 120)



   .. py:attribute:: blur_sigma


   .. py:attribute:: dilate_ksize
      :value: 0



   .. py:attribute:: line_thickness
      :value: 1



   .. py:attribute:: outline_alpha


   .. py:attribute:: outdir
      :value: b'.'



   .. py:attribute:: save_qc
      :value: True



   .. py:attribute:: save_stitched_default
      :value: True



   .. py:attribute:: all_scores
      :value: False



   .. py:attribute:: score_threshold
      :value: None



   .. py:attribute:: verbose
      :value: False



   .. py:attribute:: feature_cache_mode
      :value: ''



   .. py:attribute:: n_workers_features
      :value: 0



   .. py:attribute:: pair_batch_size
      :value: 8000



   .. py:attribute:: stream_csv
      :value: True



   .. py:attribute:: arr_axes
      :value: ''



   .. py:attribute:: mip
      :value: False



   .. py:attribute:: z_index
      :value: 0



   .. py:attribute:: t_index
      :value: 0



   .. py:attribute:: squeeze_singleton
      :value: True



   .. py:method:: set_meta_regex(pattern: Union[str, re.Pattern])


   .. py:method:: prepare_features(paths: List[str], channel_index: int, num_workers: Optional[int] = None)

      Precompute features. In 'disk' mode, every computed feature is flushed to disk
      and only an LRU-sized subset is kept in RAM. In 'ram' mode, behaves like before.



   .. py:method:: stitch_pair(pathA: str, pathB: str, channel_index: int = 0, score_threshold: Optional[float] = None, save_stitched: Optional[bool] = None, force_no_qc: bool = False, qc_only_if_score_ge: Optional[float] = None) -> Optional[Dict]


   .. py:method:: run_folder(folder: str, csv_path: str, *, channel_index: int = 0, exts: Tuple[str, Ellipsis] = ('.tif', '.tiff'), recursive: bool = False, same_well_only: bool = True, max_site_gap: int = 3, n_workers: int = 8, stitch: bool = True, score_threshold: Optional[float] = None, meta_regex: Optional[Union[str, re.Pattern]] = None, mosaic: bool = False, mosaic_out: Optional[str] = None, mosaic_min_score: Optional[float] = None, mosaic_csv_out: Optional[str] = None, mosaic_all_channels: bool = False, mosaic_channel_count: Optional[int] = None, mosaic_channel_index_order: Optional[List[int]] = None, qc_pairs_threshold: int = 1000, qc_only_above_threshold_when_many: bool = True) -> str

      When number of candidate pairs exceeds `qc_pairs_threshold`, QC plotting is suppressed
      during the first (threaded) scoring pass and, once a threshold is known, QC is produced
      only for pairs with score >= threshold (in the second pass).



   .. py:method:: build_multichannel_mosaic_from_manifest(manifest_csv: str, out_tif: str, out_png: Optional[str] = None, channel_indices: Optional[List[int]] = None, blend: str = 'max', tmp_dir: Optional[str] = None, preview_downsample: int = 8)

      Build a CYX BigTIFF mosaic from a 'mosaic.csv' manifest written by
      spacrStitcher.render_mosaic_from_csv(...).

      Expected CSV columns (per tile row):
        - path, H, W,
          M00, M01, M02,
          M10, M11, M12,
          canvas_x, canvas_y, best_pair_score

      .. rubric:: Notes

      * Uses the 2x3 affine in the manifest (already includes global offset).
      * If `channel_indices` is None, infers the channel count from the first valid tile.
      * `blend="max"` takes per-pixel max across overlapping tiles;
        `blend="overwrite"` writes the latest tile over earlier ones where coverage>0.
      * If `tmp_dir` is provided (or available from self.feature_cache_dir), the output
        workspace uses a disk memmap to reduce RAM.



   .. py:method:: mosaic_all_channels_from_csv_v1(csv_path: str, out_tif: str, *, min_score: Optional[float] = None, channel_count: Optional[int] = None, channel_index_order: Optional[List[int]] = None, angle_tol_deg: float = 30.0, step_tol_frac: float = 0.25, rot_tol_deg: float = 5.0, scale_tol: float = 0.03, cap_one_per_dir: bool = True, out_csv: Optional[str] = None) -> str

      Build a CYX mosaic by reusing pairwise transforms computed on (typically) the nuclei channel.

      :param csv_path: Pairwise results CSV produced by `run_folder(...)`.
      :type csv_path: str
      :param out_tif: Output TIFF path (saved with axes='CYX').
      :type out_tif: str
      :param min_score: Minimum pairwise score to keep when building the mosaic graph. If None, auto-knee.
      :type min_score: Optional[float]
      :param channel_count: If provided, force this many channels to be mosaicked (0..channel_count-1).
                            Otherwise inferred as the minimum channel count across nodes.
      :type channel_count: Optional[int]
      :param channel_index_order: Optional explicit channel indices to mosaic (e.g., [0,3,1]). Overrides `channel_count` if given.
      :type channel_index_order: Optional[List[int]]
      :param angle_tol_deg: Same gating params as single-channel mosaic.
      :param step_tol_frac: Same gating params as single-channel mosaic.
      :param rot_tol_deg: Same gating params as single-channel mosaic.
      :param scale_tol: Same gating params as single-channel mosaic.
      :param cap_one_per_dir: Same gating params as single-channel mosaic.
      :param out_csv: If provided, write a manifest with per-node canvas transforms.
      :type out_csv: Optional[str]

      :returns: Path to the saved multi-channel mosaic TIFF (CYX).
      :rtype: str



   .. py:method:: render_mosaic_from_csv_v1(csv_path: str, out_tif: str, out_png: Optional[str] = None, channel_index: int = 0, min_score: Optional[float] = None, *, angle_tol_deg: float = 30.0, step_tol_frac: float = 0.25, rot_tol_deg: float = 5.0, scale_tol: float = 0.03, cap_one_per_dir: bool = True, out_csv: Optional[str] = None) -> Tuple[str, Optional[str]]


   .. py:method:: render_mosaic_from_csv(csv_path: str, out_tif: str, out_png: Optional[str] = None, channel_index: int = 0, min_score: Optional[float] = None, *, angle_tol_deg: float = 30.0, step_tol_frac: float = 0.25, rot_tol_deg: float = 5.0, scale_tol: float = 0.03, cap_one_per_dir: bool = True, out_csv: Optional[str] = None) -> Tuple[str, Optional[str]]


   .. py:method:: mosaic_all_channels_from_csv(csv_path: str, out_tif: Optional[str], *, min_score: Optional[float] = None, channel_count: Optional[int] = None, channel_index_order: Optional[List[int]] = None, angle_tol_deg: float = 30.0, step_tol_frac: float = 0.25, rot_tol_deg: float = 5.0, scale_tol: float = 0.03, cap_one_per_dir: bool = True, out_csv: Optional[str] = None) -> Optional[str]

      Build a CYX mosaic by reusing pairwise transforms computed on (typically) the nuclei channel.

      If out_csv is not None and out_tif is None, run in "manifest-only" mode:
      compute transforms + canvas geometry + per-node canvas transforms and write the manifest CSV,
      but DO NOT render/write the mosaic TIFF.



.. py:class:: StitchedMultiAligner(detector: str = 'ORB', nfeatures: int = 6000, max_keypoints: Optional[int] = 2000, downsample: float = 0.5, ransac_thresh_px: float = 3.0, allow_scale: bool = False, allow_rotation: bool = False, outdir: str = './align_out', opencv_threads: int = 1, arr_axes: str = 'AUTO', mip: bool = False, z_index: int = 0, t_index: int = 0, squeeze_singleton: bool = True)

   Align an arbitrary number of stitched mosaics (multi-channel, arbitrary axes) to a common reference
   using the nuclei/ Hoechst channel. Saves a channel-concatenated aligned stack and a CSV manifest
   describing the mapping from input channels to output channels and the estimated transforms/scores.

   Output image axes: CYX (channels stacked in the order inputs are provided).

   :param detector: Feature detector for keypoint matching.
   :type detector: {"ORB","SIFT"}
   :param nfeatures: Feature budget for detector.
   :type nfeatures: int
   :param max_keypoints: Hard cap on kept keypoints after detection (by detector’s internal ranking).
   :type max_keypoints: Optional[int]
   :param downsample: Downsample factor for feature/score pass.
   :type downsample: float in (0,1]
   :param ransac_thresh_px: Reprojection threshold (pixels) for affine estimation (downsampled space).
   :type ransac_thresh_px: float
   :param allow_scale: If False, constrain to rotation+translation (or translation only if allow_rotation=False).
   :type allow_scale: bool
   :param allow_rotation: If False, constrain to translation only.
   :type allow_rotation: bool
   :param outdir: Output directory for images/csv.
   :type outdir: str
   :param opencv_threads: Limit OpenCV internal threading (avoid oversubscription).
   :type opencv_threads: int
   :param # Axis/Z/time handling (for TIFF reading):
   :param arr_axes:
   :type arr_axes: "AUTO" or a string over {T,C,Z,Y,X}
   :param mip: If True and Z exists, max-project Z.
   :type mip: bool
   :param z_index: If mip=False, choose Z slice.
   :type z_index: int
   :param t_index: Choose T index if T exists.
   :type t_index: int
   :param squeeze_singleton: Squeeze 1-length axes after slicing.
   :type squeeze_singleton: bool

   .. rubric:: Notes

   - Alignment is done to the first image in `paths` (reference).
   - For each image, you can provide a per-image nuclei channel index via `nuclei_channel_indices`.
     If None, defaults to 0 for all.


   .. py:attribute:: detector
      :value: ''



   .. py:attribute:: nfeatures
      :value: 6000



   .. py:attribute:: max_keypoints
      :value: None



   .. py:attribute:: downsample


   .. py:attribute:: ransac_thresh_px


   .. py:attribute:: allow_scale
      :value: False



   .. py:attribute:: allow_rotation
      :value: False



   .. py:attribute:: outdir
      :value: b'.'



   .. py:attribute:: arr_axes
      :value: ''



   .. py:attribute:: mip
      :value: False



   .. py:attribute:: z_index
      :value: 0



   .. py:attribute:: t_index
      :value: 0



   .. py:attribute:: squeeze_singleton
      :value: True



   .. py:method:: align(paths: List[str], nuclei_channel_indices: Optional[List[int]] = None, out_tif: Optional[str] = None, out_png_preview: Optional[str] = None, csv_path: Optional[str] = None) -> Tuple[str, Optional[str], Optional[str]]


.. py:function:: stitch_cycle_wells(settings)

.. py:function:: get_preprocess_ops_settings(settings)

.. py:class:: FOVAlignAndCropper(detector: str = 'ORB', nfeatures: int = 6000, max_keypoints: Optional[int] = 2000, downsample: float = 0.5, ransac_thresh_px: float = 3.0, allow_scale: bool = False, allow_rotation: bool = False, outdir: str = './fov_out', opencv_threads: int = 1, arr_axes: str = 'AUTO', mip: bool = False, z_index: int = 0, t_index: int = 0, squeeze_singleton: bool = True, folder_image_scale: float = 1.0)

   Align each image in a folder (arbitrary channels) to a stitched mosaic (arbitrary channels) using
   the Hoechst/nuclei channel, then extract the FOV region from the mosaic at the aligned location.
   For each input FOV, saves:
     - a .npy array with shape (C_fov + C_mosaic, H_fov, W_fov):
         [FOV channels stacked; mosaic channels warped into FOV frame and stacked]
     - a CSV row with file paths, transform, score, and the mosaic-space top-left of the aligned FOV bbox.

   :param detector: Same semantics as in StitchedMultiAligner.
   :param nfeatures: Same semantics as in StitchedMultiAligner.
   :param max_keypoints: Same semantics as in StitchedMultiAligner.
   :param downsample: Same semantics as in StitchedMultiAligner.
   :param ransac_thresh_px: Same semantics as in StitchedMultiAligner.
   :param allow_scale: Same semantics as in StitchedMultiAligner.
   :param allow_rotation: Same semantics as in StitchedMultiAligner.
   :param outdir: Same semantics as in StitchedMultiAligner.
   :param opencv_threads: Same semantics as in StitchedMultiAligner.
   :param arr_axes: Same TIFF axis handling semantics as in StitchedMultiAligner.
   :param mip: Same TIFF axis handling semantics as in StitchedMultiAligner.
   :param z_index: Same TIFF axis handling semantics as in StitchedMultiAligner.
   :param t_index: Same TIFF axis handling semantics as in StitchedMultiAligner.
   :param squeeze_singleton: Same TIFF axis handling semantics as in StitchedMultiAligner.

   .. rubric:: Notes

   - Alignment is (FOV -> mosaic). Extraction uses the inverse transform to warp the mosaic into the FOV frame,
     guaranteeing the combined array has the FOV’s native (H_fov, W_fov).


   .. py:attribute:: outdir
      :value: b'.'



   .. py:attribute:: folder_image_scale


   .. py:method:: run(stitched_path: str, folder: str, *, stitched_nuclei_idx: int = 0, fov_nuclei_idx: int = 0, exts: Tuple[str, Ellipsis] = ('.tif', '.tiff'), recursive: bool = False, csv_path: Optional[str] = None, npy_dir: Optional[str] = None, folder_image_scale: Optional[float] = None) -> str

      Align each FOV in `folder` to the stitched mosaic at `stitched_path` using the nuclei channel,
      then save a stacked array [FOV channels; mosaic channels warped into FOV] to .npy and a CSV row.

      Known scale handling:
        - If the FOV magnification differs from the mosaic, set `folder_image_scale` to the
          FOV→mosaic pixel-scale factor (e.g., mosaic 10× vs FOV 20× -> 0.5; mosaic 20× vs FOV 10× -> 2.0).

      :returns: Path to the CSV manifest.
      :rtype: str



.. py:function:: align_image_to_stitch(stitch_dst_root: str, align_src: str, *, meta_regex: str = '(?P<mag>\\d+X)_c(?P<chan>\\d+)_?(?P<well>[A-H]\\d{1,2}).*?Site[-_](?P<site>\\d+)\\.(?:tif|tiff)$', well_group: str = 'well', channel_index: int = 0, relative_scale: float = 2.0, downsample: float = 0.5, nfeatures: int = 4000, ransac_thresh_px: float = 3.0, allow_scale: bool = False, allow_rotation: bool = False, qc_outlines: bool = False, recursive_align_src: bool = True, exts: tuple = ('.tif', '.tiff')) -> Dict[str, Dict[str, str]]

   For each WELL that already has a stitched mosaic under stitch_dst_root/<WELL>/_stitch/mosaic_allc.tif,
   align all 20× images from align_src (grouped by WELL via meta_regex) to that mosaic using
   FOVAlignAndCropper, and write per-well crop manifests and .npy crops.

   :returns: {"mosaic": <path>, "align_folder": <per-well link>, "manifest_csv": <path>} }
   :rtype: { WELL


.. py:function:: ops_preprocess(settings)

