Source code for thesis.workflows.mrtrix3.operations.tckgen_adapter

"""Adapter from the canonical 5-tuple ROI fields to tckgen's input shape.

The reused HCP ROI pipeline produces:

* ``seed`` — single-path string for the seed mask
* ``waypoints_file`` — text file listing one waypoint mask path per line
* ``stop_mask``, ``avoid_mask``, ``target_mask`` — single-path strings (or empty)

MRtrix's ``tckgen`` instead expects:

* ``-seed_image`` — single path
* ``-include`` — repeated flag; the Nipype interface accepts a list under
  ``roi_incl``
* ``-exclude`` — repeated flag; ``roi_excl``
* ``-mask`` — single path; ``roi_mask``

This task adapts between the two shapes inside a single Function node so
the workflow only has to wire the canonical fields once. Targets are
folded into the include list (MRtrix has no separate target concept).
"""




def adapt_tckgen_inputs_task(
    seed: str,
    waypoints_file: str,
    target_mask: str,
    avoid_mask: str,
    stop_mask: str,
):
    """Translate canonical ROI fields into tckgen-friendly inputs.

    Args:
        seed: Path to the seed mask. Required.
        waypoints_file: Optional text file listing one waypoint mask path
            per line. Empty string disables.
        target_mask: Optional path to a target mask, folded into the
            include list. Empty string disables.
        avoid_mask: Optional path to an avoid (exclude) mask. Empty string
            disables.
        stop_mask: Optional path to a stop mask (truncates streamlines on
            exit). Empty string disables.

    Returns:
        Tuple of ``(seed_image, include_masks, exclude_masks, mask_stop)``
        where ``include_masks`` and ``exclude_masks`` are lists of strings
        (possibly empty) and ``mask_stop`` is a string (possibly empty).
    """
    from pathlib import Path

    include_masks: list[str] = []
    if waypoints_file:
        text = Path(waypoints_file).read_text(encoding="utf-8")
        for line in text.splitlines():
            stripped = line.strip()
            if stripped:
                include_masks.append(stripped)

    if target_mask:
        include_masks.append(str(target_mask))

    exclude_masks = [str(avoid_mask)] if avoid_mask else []
    mask_stop = str(stop_mask) if stop_mask else ""

    return str(seed), include_masks, exclude_masks, mask_stop