Python API¶

from manifold_genetics import run_pipeline, load_config, PCA, PHATE, UMAP, TSNE, DiffusionMap

Pipeline¶

manifold_genetics.pipeline.runner.run_pipeline ¶

run_pipeline(fit_plink, project_plink, output_dir, labels=None, colormap=None, fit_labels=None, project_labels=None, fit_colormap=None, project_colormap=None, geographic_coords=None, n_pcs=50, pca_backend='python', max_fit_memory_gb=8.0, max_project_memory_gb=8.0, k_min=2, k_max=10, admix_threads=None, admix_gpus=None, admix_batch_size=400, admixture_backend=None, embedding='phate', embedding_params=None, embedding_input='both', admix_group_column=None, admix_within_group_order='chron', projection_plot_fit_column=None, projection_plot_project_column=None, skip_pca=False, skip_admixture=False, skip_embedding=False, skip_visualization=False, skip_pca_visualization=False, skip_admixture_visualization=False, skip_metrics=False)

Run the complete manifold-genetics pipeline.

This is the canonical entry point for running the full genetic analysis pipeline. It coordinates PCA, Admixture, Embeddings, Visualization, and Metrics computation.

Which fields are populated depends on what ran. admixture and embedding are None when their stage was skipped, as are geographic_metrics and admixture_metrics. Figure families are empty rather than None.

pca is the exception: it is never None. Under skip_pca it is still a PCAStepResult, with skipped=True and fit_pca/project_pca filled only from cached output already on disk — possibly both None. Check .skipped rather than truthiness to tell whether PCA ran.

Parameters:

Name	Type	Description	Default
`fit_plink`	`Union[str, Path]`	Path to fit subset PLINK files (prefix for .bed/.bim/.fam)	required
`project_plink`	`Union[str, Path]`	Path to project subset PLINK files (prefix for .bed/.bim/.fam)	required
`output_dir`	`Union[str, Path]`	Directory for all outputs	required
`labels`	`Optional[Union[str, Path]]`	Path to labels CSV (used for both fit and project if not overridden)	`None`
`colormap`	`Optional[Union[str, Path]]`	Path to colormap JSON (used for both fit and project if not overridden)	`None`
`fit_labels`	`Optional[Union[str, Path]]`	Optional override labels CSV for fit dataset	`None`
`project_labels`	`Optional[Union[str, Path]]`	Optional override labels CSV for project dataset	`None`
`fit_colormap`	`Optional[Union[str, Path]]`	Optional override colormap JSON for fit dataset	`None`
`project_colormap`	`Optional[Union[str, Path]]`	Optional override colormap JSON for project dataset	`None`
`geographic_coords`	`Optional[Union[str, Path]]`	Optional path to geographic coordinates CSV for metrics	`None`
`n_pcs`	`int`	Number of principal components (default: 50)	`50`
`pca_backend`	`str`	'flashpca' (external binary) or 'python' (in process)	`'python'`
`max_fit_memory_gb`	`float`	GB budget for the dense standardised matrix when fitting the in-process PCA backend. Above it the fit streams, which bounds memory to about 110 MB at roughly nineteen times the wall clock. Raise it to match the node: a 60,000 x 120,849 cohort is 54 GB dense and streams under the default.	`8.0`
`max_project_memory_gb`	`float`	GB budget for one chunk when projecting. Peak resident memory runs to about three times this. Both are ignored by the flashpca backend.	`8.0`
`k_min`	`int`	Minimum K for admixture (default: 2)	`2`
`k_max`	`int`	Maximum K for admixture (default: 10)	`10`
`admix_threads`	`Optional[int]`	Number of threads for neural admixture (None = auto-detect)	`None`
`admix_gpus`	`Optional[int]`	Number of GPUs for neural admixture (None = auto-detect)	`None`
`admix_batch_size`	`Optional[int]`	Batch size for neural admixture training	`400`
`admixture_backend`	`Optional[object]`	Optional AdmixtureBackend instance for testing (if None, constructs a real NeuralAdmixtureBackend, which runs neural-admixture in its own child process)	`None`
`embedding`	`str`	Embedding method - 'phate', 'umap', 'tsne', or 'diffusion_map' (default: 'phate')	`'phate'`
`embedding_params`	`Optional[Dict]`	Optional dictionary of embedding-specific parameters	`None`
`embedding_input`	`str`	Which dataset to embed - 'fit', 'project', or 'both' (default: 'both')	`'both'`
`admix_group_column`	`Optional[str]`	Column for grouping in admixture barplots (None = use first colormap key)	`None`
`admix_within_group_order`	`Optional[str]`	Method for ordering samples within groups ('chron', 'tree', or None)	`'chron'`
`projection_plot_fit_column`	`Optional[str]`	Column from fit colormap to use for projection plot	`None`
`projection_plot_project_column`	`Optional[str]`	Column from project colormap to use for projection plot	`None`
`skip_pca`	`bool`	Skip PCA step	`False`
`skip_admixture`	`bool`	Skip admixture step	`False`
`skip_embedding`	`bool`	Skip embedding step	`False`
`skip_visualization`	`bool`	Skip embedding visualization step	`False`
`skip_pca_visualization`	`bool`	Skip PCA visualization step	`False`
`skip_admixture_visualization`	`bool`	Skip admixture visualization step	`False`
`skip_metrics`	`bool`	Skip metrics computation	`False`

Returns:

Name	Type	Description
`PipelineResult`	`PipelineResult`	The typed outputs of every stage that ran.

Examples:

>>> # Basic usage with shared labels/colormap
>>> results = run_pipeline(
...     fit_plink="data/fit_subset",
...     project_plink="data/project_subset",
...     labels="labels.csv",
...     colormap="colormap.json",
...     output_dir="results/",
...     n_pcs=50,
...     k_min=2,
...     k_max=10,
...     embedding="phate",
...     embedding_params={"knn": 100, "t": 3}
... )

>>> # Cross-cohort analysis with separate labels/colormaps
>>> results = run_pipeline(
...     fit_plink="data/hgdp_fit",
...     project_plink="data/ukbb_project",
...     fit_labels="hgdp_labels.csv",
...     project_labels="ukbb_labels.csv",
...     fit_colormap="hgdp_colors.json",
...     project_colormap="ukbb_colors.json",
...     output_dir="cross_cohort_results/",
... )

>>> # Skip steps selectively
>>> results = run_pipeline(
...     fit_plink="data/fit",
...     project_plink="data/project",
...     labels="labels.csv",
...     colormap="colormap.json",
...     output_dir="results/",
...     skip_pca=True,  # Use existing PCA
...     skip_metrics=True  # Don't compute metrics
... )

Note

You must provide either:

A shared set of labels and colormap, which will be used for both the fit and project cohorts, or
Separate values for all of fit_labels, project_labels, fit_colormap, and project_colormap for cross-cohort analysis.

Source code in src/manifold_genetics/pipeline/runner.py

def run_pipeline(
    fit_plink: Union[str, Path],
    project_plink: Union[str, Path],
    output_dir: Union[str, Path],
    labels: Optional[Union[str, Path]] = None,
    colormap: Optional[Union[str, Path]] = None,
    # Optional overrides for cross-cohort analysis
    fit_labels: Optional[Union[str, Path]] = None,
    project_labels: Optional[Union[str, Path]] = None,
    fit_colormap: Optional[Union[str, Path]] = None,
    project_colormap: Optional[Union[str, Path]] = None,
    # Geographic coordinates for metrics
    geographic_coords: Optional[Union[str, Path]] = None,
    # PCA parameters
    n_pcs: int = 50,
    pca_backend: str = "python",
    max_fit_memory_gb: float = 8.0,
    max_project_memory_gb: float = 8.0,
    # Admixture parameters
    k_min: int = 2,
    k_max: int = 10,
    admix_threads: Optional[int] = None,
    admix_gpus: Optional[int] = None,
    admix_batch_size: Optional[int] = 400,
    admixture_backend: Optional[object] = None,
    # Embedding parameters
    embedding: str = "phate",
    embedding_params: Optional[Dict] = None,
    embedding_input: str = "both",
    # Visualization parameters
    admix_group_column: Optional[str] = None,
    admix_within_group_order: Optional[str] = "chron",
    projection_plot_fit_column: Optional[str] = None,
    projection_plot_project_column: Optional[str] = None,
    # Skip flags
    skip_pca: bool = False,
    skip_admixture: bool = False,
    skip_embedding: bool = False,
    skip_visualization: bool = False,
    skip_pca_visualization: bool = False,
    skip_admixture_visualization: bool = False,
    skip_metrics: bool = False,
) -> PipelineResult:
    """
    Run the complete manifold-genetics pipeline.

    This is the canonical entry point for running the full genetic analysis pipeline.
    It coordinates PCA, Admixture, Embeddings, Visualization, and Metrics computation.

    Which fields are populated depends on what ran. ``admixture`` and
    ``embedding`` are None when their stage was skipped, as are
    ``geographic_metrics`` and ``admixture_metrics``. Figure families are empty
    rather than None.

    ``pca`` is the exception: it is never None. Under ``skip_pca`` it is still a
    ``PCAStepResult``, with ``skipped=True`` and ``fit_pca``/``project_pca``
    filled only from cached output already on disk — possibly both None. Check
    ``.skipped`` rather than truthiness to tell whether PCA ran.

    Args:
        fit_plink: Path to fit subset PLINK files (prefix for .bed/.bim/.fam)
        project_plink: Path to project subset PLINK files (prefix for .bed/.bim/.fam)
        output_dir: Directory for all outputs
        labels: Path to labels CSV (used for both fit and project if not overridden)
        colormap: Path to colormap JSON (used for both fit and project if not overridden)
        fit_labels: Optional override labels CSV for fit dataset
        project_labels: Optional override labels CSV for project dataset
        fit_colormap: Optional override colormap JSON for fit dataset
        project_colormap: Optional override colormap JSON for project dataset
        geographic_coords: Optional path to geographic coordinates CSV for metrics
        n_pcs: Number of principal components (default: 50)
        pca_backend: 'flashpca' (external binary) or 'python' (in process)
        max_fit_memory_gb: GB budget for the dense standardised matrix when
            fitting the in-process PCA backend. Above it the fit streams, which
            bounds memory to about 110 MB at roughly nineteen times the wall
            clock. Raise it to match the node: a 60,000 x 120,849 cohort is 54 GB
            dense and streams under the default.
        max_project_memory_gb: GB budget for one chunk when projecting. Peak
            resident memory runs to about three times this. Both are ignored by
            the flashpca backend.
        k_min: Minimum K for admixture (default: 2)
        k_max: Maximum K for admixture (default: 10)
        admix_threads: Number of threads for neural admixture (None = auto-detect)
        admix_gpus: Number of GPUs for neural admixture (None = auto-detect)
        admix_batch_size: Batch size for neural admixture training
        admixture_backend: Optional AdmixtureBackend instance for testing
                          (if None, constructs a real NeuralAdmixtureBackend, which
                          runs neural-admixture in its own child process)
        embedding: Embedding method - 'phate', 'umap', 'tsne', or 'diffusion_map' (default: 'phate')
        embedding_params: Optional dictionary of embedding-specific parameters
        embedding_input: Which dataset to embed - 'fit', 'project', or 'both' (default: 'both')
        admix_group_column: Column for grouping in admixture barplots (None = use first colormap key)
        admix_within_group_order: Method for ordering samples within groups ('chron', 'tree', or None)
        projection_plot_fit_column: Column from fit colormap to use for projection plot
        projection_plot_project_column: Column from project colormap to use for projection plot
        skip_pca: Skip PCA step
        skip_admixture: Skip admixture step
        skip_embedding: Skip embedding step
        skip_visualization: Skip embedding visualization step
        skip_pca_visualization: Skip PCA visualization step
        skip_admixture_visualization: Skip admixture visualization step
        skip_metrics: Skip metrics computation

    Returns:
        PipelineResult: The typed outputs of every stage that ran.

    Examples:
        >>> # Basic usage with shared labels/colormap
        >>> results = run_pipeline(
        ...     fit_plink="data/fit_subset",
        ...     project_plink="data/project_subset",
        ...     labels="labels.csv",
        ...     colormap="colormap.json",
        ...     output_dir="results/",
        ...     n_pcs=50,
        ...     k_min=2,
        ...     k_max=10,
        ...     embedding="phate",
        ...     embedding_params={"knn": 100, "t": 3}
        ... )

        >>> # Cross-cohort analysis with separate labels/colormaps
        >>> results = run_pipeline(
        ...     fit_plink="data/hgdp_fit",
        ...     project_plink="data/ukbb_project",
        ...     fit_labels="hgdp_labels.csv",
        ...     project_labels="ukbb_labels.csv",
        ...     fit_colormap="hgdp_colors.json",
        ...     project_colormap="ukbb_colors.json",
        ...     output_dir="cross_cohort_results/",
        ... )

        >>> # Skip steps selectively
        >>> results = run_pipeline(
        ...     fit_plink="data/fit",
        ...     project_plink="data/project",
        ...     labels="labels.csv",
        ...     colormap="colormap.json",
        ...     output_dir="results/",
        ...     skip_pca=True,  # Use existing PCA
        ...     skip_metrics=True  # Don't compute metrics
        ... )

    Note:
        You must provide either:

        * A shared set of `labels` and `colormap`, which will be used for both the
          fit and project cohorts, **or**
        * Separate values for all of `fit_labels`, `project_labels`, `fit_colormap`,
          and `project_colormap` for cross-cohort analysis.
    """
    # Labels/colormap argument-shape validation now lives in Pipeline.__init__
    # (via build_configs()), which runs before output_dir is created — no need
    # to duplicate it here, and duplicating it would risk the two messages
    # drifting apart.

    # Create Pipeline instance
    pipeline = Pipeline(
        fit_plink_prefix=fit_plink,
        project_plink_prefix=project_plink,
        labels=labels,
        colormap=colormap,
        output_dir=output_dir,
        geographic_coords=geographic_coords,
        fit_labels=fit_labels,
        project_labels=project_labels,
        fit_colormap=fit_colormap,
        project_colormap=project_colormap,
        admixture_backend=admixture_backend,
        projection_plot_fit_column=projection_plot_fit_column,
        projection_plot_project_column=projection_plot_project_column,
    )

    # Run pipeline with all parameters
    results = pipeline.run(
        n_pcs=n_pcs,
        pca_backend=pca_backend,
        max_fit_memory_gb=max_fit_memory_gb,
        max_project_memory_gb=max_project_memory_gb,
        k_min=k_min,
        k_max=k_max,
        embedding=embedding,
        embedding_params=embedding_params,
        embedding_input=embedding_input,
        skip_pca=skip_pca,
        skip_admixture=skip_admixture,
        skip_embedding=skip_embedding,
        skip_visualization=skip_visualization,
        skip_pca_visualization=skip_pca_visualization,
        skip_admixture_visualization=skip_admixture_visualization,
        skip_metrics=skip_metrics,
        admix_group_column=admix_group_column,
        admix_within_group_order=admix_within_group_order,
        admix_threads=admix_threads,
        admix_gpus=admix_gpus,
        admix_batch_size=admix_batch_size,
    )

    logger.info("Pipeline execution complete")
    return results

manifold_genetics.pipeline.configfile.load_config ¶

load_config(path, base_dir=None)

Read path and return keyword arguments for run_pipeline.

Parameters:

Name	Type	Description	Default
`path`	`PathLike`	The YAML config file.	required
`base_dir`	`Optional[PathLike]`	Directory relative paths resolve against. Defaults to the config file's own directory, which is what makes an example runnable from anywhere.	`None`

Raises:

Type	Description
`ConfigFileError`	missing file, malformed YAML, unknown section or key, unknown preset, or a contradictory embedding setting.

Source code in src/manifold_genetics/pipeline/configfile.py

def load_config(path: PathLike, base_dir: Optional[PathLike] = None) -> Dict[str, Any]:
    """Read ``path`` and return keyword arguments for ``run_pipeline``.

    Args:
        path: The YAML config file.
        base_dir: Directory relative paths resolve against. Defaults to the
            config file's own directory, which is what makes an example runnable
            from anywhere.

    Raises:
        ConfigFileError: missing file, malformed YAML, unknown section or key,
            unknown preset, or a contradictory embedding setting.
    """
    path = Path(path).expanduser()
    data = _load_yaml(path)
    base = Path(base_dir) if base_dir is not None else path.parent.resolve()

    _reject_unknown(data, _SECTIONS, f"section(s) in {path.name}")

    preset_name = data.get("preset")
    if preset_name in PRESET_ALIASES:
        replacement = PRESET_ALIASES[preset_name]
        warnings.warn(
            f"The preset {preset_name!r} has been renamed to {replacement!r}, because "
            f"{preset_name!r} is the name of a method rather than of a mode. The old "
            "name still works and will be removed in a future release.",
            DeprecationWarning,
            stacklevel=2,
        )
        preset_name = replacement
    if preset_name is not None and preset_name not in PRESETS:
        raise ConfigFileError(
            f"Unknown preset {preset_name!r}. Choose from: {', '.join(sorted(PRESETS))}"
        )
    preset = PRESETS.get(preset_name, {})

    if "data" not in data:
        raise ConfigFileError(
            f"{path.name} has no 'data' section. It must name at least fit_plink, "
            "project_plink, labels, colormap and output_dir."
        )

    kwargs: Dict[str, Any] = {}

    section = data["data"] or {}
    _reject_unknown(section, _DATA_KEYS, "key(s) in the 'data' section")
    for key, value in section.items():
        kwargs[_DATA_KEYS[key]] = _resolve(value, base) if key in _PATH_KEYS else value

    for name, mapping in (
        ("pca", _PCA_KEYS),
        ("admixture", _ADMIXTURE_KEYS),
        ("visualization", _VIZ_KEYS),
        ("skip", _SKIP_KEYS),
    ):
        section = data.get(name) or {}
        _reject_unknown(section, mapping, f"key(s) in the {name!r} section")
        for key, value in section.items():
            kwargs[mapping[key]] = value

    for key, arg in _SKIP_KEYS.items():
        kwargs.setdefault(arg, False)

    # --- embedding: preset defaults, then the file's own values ---
    for key, value in preset.items():
        if key != "embedding":
            kwargs.setdefault(key, value)
    params: Dict[str, Any] = dict(preset.get("embedding", {}))

    section = data.get("embedding") or {}
    _reject_unknown(
        section, set(_EMBEDDING_ARGS) | _EMBEDDING_PARAMS, "key(s) in the 'embedding' section"
    )
    for key, value in section.items():
        if key in _EMBEDDING_ARGS:
            kwargs[_EMBEDDING_ARGS[key]] = value
        else:
            params[key] = value

    if isinstance(params.get("n_landmark"), str):
        if params["n_landmark"].lower() == "none":
            params["n_landmark"] = None
        else:
            params["n_landmark"] = int(params["n_landmark"])

    if params.get("random_landmarking") and params.get("n_landmark") is None:
        raise ConfigFileError(
            "random_landmarking is set but n_landmark is not. Landmarking is disabled "
            "when n_landmark is none, so random_landmarking would have no effect."
        )

    kwargs["embedding_params"] = params
    return kwargs

manifold_genetics.pipeline.orchestrator.Pipeline ¶

Pipeline(fit_plink_prefix, project_plink_prefix, labels=None, colormap=None, output_dir=None, geographic_coords=None, fit_labels=None, project_labels=None, fit_colormap=None, project_colormap=None, admixture_backend=None, projection_plot_fit_column=None, projection_plot_project_column=None)

End-to-end pipeline for genetic analysis.

Examples:

>>> # Full pipeline
>>> pipeline = Pipeline(
...     fit_plink_prefix="data/fit_subset",
...     project_plink_prefix="data/project_subset",
...     labels="labels.csv",
...     colormap="colormap.json",
...     output_dir="results/"
... )
>>> results = pipeline.run(
...     n_pcs=50,
...     k_min=2, k_max=10,
...     embedding="phate", knn=25
... )

Initialize pipeline.

Parameters:

Name	Type	Description	Default
`fit_plink_prefix`	`Union[str, Path]`	Path to fit subset PLINK files	required
`project_plink_prefix`	`Union[str, Path]`	Path to project subset PLINK files	required
`labels`	`Optional[Union[str, Path]]`	Path to labels CSV (used for both fit and project if not overridden)	`None`
`colormap`	`Optional[Union[str, Path]]`	Path to colormap JSON (used for both fit and project if not overridden)	`None`
`output_dir`	`Union[str, Path]`	Directory for outputs	`None`
`geographic_coords`	`Optional[Union[str, Path]]`	Optional path to geographic coordinates	`None`
`fit_labels`	`Optional[Union[str, Path]]`	Optional override labels CSV for fit dataset	`None`
`project_labels`	`Optional[Union[str, Path]]`	Optional override labels CSV for project dataset	`None`
`fit_colormap`	`Optional[Union[str, Path]]`	Optional override colormap JSON for fit dataset	`None`
`project_colormap`	`Optional[Union[str, Path]]`	Optional override colormap JSON for project dataset	`None`
`admixture_backend`	`Optional[object]`	Optional AdmixtureBackend instance for testing (if None, constructs a real NeuralAdmixtureBackend, which runs neural-admixture in its own child process)	`None`
`projection_plot_fit_column`	`Optional[str]`	Column from fit colormap to use for projection plot	`None`
`projection_plot_project_column`	`Optional[str]`	Column from project colormap to use for projection plot	`None`

Note

Must provide either (labels + colormap) OR (fit_labels + project_labels + fit_colormap + project_colormap)

Source code in src/manifold_genetics/pipeline/orchestrator.py

def __init__(
    self,
    fit_plink_prefix: Union[str, Path],
    project_plink_prefix: Union[str, Path],
    labels: Optional[Union[str, Path]] = None,
    colormap: Optional[Union[str, Path]] = None,
    output_dir: Union[str, Path] = None,
    geographic_coords: Optional[Union[str, Path]] = None,
    fit_labels: Optional[Union[str, Path]] = None,
    project_labels: Optional[Union[str, Path]] = None,
    fit_colormap: Optional[Union[str, Path]] = None,
    project_colormap: Optional[Union[str, Path]] = None,
    admixture_backend: Optional[object] = None,
    projection_plot_fit_column: Optional[str] = None,
    projection_plot_project_column: Optional[str] = None,
):
    """
    Initialize pipeline.

    Args:
        fit_plink_prefix: Path to fit subset PLINK files
        project_plink_prefix: Path to project subset PLINK files
        labels: Path to labels CSV (used for both fit and project if not overridden)
        colormap: Path to colormap JSON (used for both fit and project if not overridden)
        output_dir: Directory for outputs
        geographic_coords: Optional path to geographic coordinates
        fit_labels: Optional override labels CSV for fit dataset
        project_labels: Optional override labels CSV for project dataset
        fit_colormap: Optional override colormap JSON for fit dataset
        project_colormap: Optional override colormap JSON for project dataset
        admixture_backend: Optional AdmixtureBackend instance for testing
                          (if None, constructs a real NeuralAdmixtureBackend, which
                          runs neural-admixture in its own child process)
        projection_plot_fit_column: Column from fit colormap to use for projection plot
        projection_plot_project_column: Column from project colormap to use for projection plot

    Note:
        Must provide either (labels + colormap) OR (fit_labels + project_labels + fit_colormap + project_colormap)
    """
    # build_configs() performs all labels/colormap argument-shape validation
    # (no filesystem access) and must run before output_dir is created below —
    # tests/unit/test_runner.py::test_validation_fires_before_output_dir_created
    # guards exactly this ordering.
    #
    # __init__ only receives the IO and viz-relevant arguments; run()-time
    # parameters (n_pcs, k_min/k_max, embedding, the skip flags, and the two
    # admixture-viz ordering knobs) are not passed here, so the pca/admixture/
    # embedding/skips configs this call produces are built from defaults and
    # discarded — run() builds its own from its own arguments, exactly as
    # before. This keeps `Pipeline(...)` and `run_pipeline(...)` signatures
    # unchanged and avoids storing per-run state on the instance.
    configs = build_configs(
        fit_plink=fit_plink_prefix,
        project_plink=project_plink_prefix,
        output_dir=output_dir,
        labels=labels,
        colormap=colormap,
        fit_labels=fit_labels,
        project_labels=project_labels,
        fit_colormap=fit_colormap,
        project_colormap=project_colormap,
        geographic_coords=geographic_coords,
        projection_plot_fit_column=projection_plot_fit_column,
        projection_plot_project_column=projection_plot_project_column,
    )

    self._io = configs.io
    self._viz_config = configs.viz

    # Loose attributes kept for backward compatibility (existing callers and
    # tests read these directly off the Pipeline instance).
    self.fit_plink_prefix = configs.io.fit_plink
    self.project_plink_prefix = configs.io.project_plink
    self.output_dir = configs.io.output_dir
    self.geographic_coords = configs.io.geographic_coords
    self.fit_labels = configs.io.fit_labels
    self.project_labels = configs.io.project_labels
    self.fit_colormap = configs.io.fit_colormap
    self.project_colormap = configs.io.project_colormap
    self.labels = Path(labels) if labels else None
    self.colormap = Path(colormap) if colormap else None

    self.output_dir.mkdir(parents=True, exist_ok=True)

    # Store admixture backend (for testing)
    self.admixture_backend = admixture_backend

    # Store projection column settings
    self.projection_plot_fit_column = projection_plot_fit_column
    self.projection_plot_project_column = projection_plot_project_column

run ¶

run(n_pcs=50, pca_backend='python', max_fit_memory_gb=8.0, max_project_memory_gb=8.0, k_min=2, k_max=10, embedding='phate', embedding_params=None, embedding_input='both', skip_pca=False, skip_admixture=False, skip_embedding=False, skip_visualization=False, skip_pca_visualization=False, skip_admixture_visualization=False, admix_group_column=None, admix_within_group_order='chron', skip_metrics=False, admix_threads=None, admix_gpus=None, admix_batch_size=400)

Run full pipeline.

Parameters:

Name	Type	Description	Default
`pca_backend`	`str`	'flashpca' (external binary) or 'python' (in process)	`'python'`
`max_fit_memory_gb`	`float`	GB budget for the dense standardised matrix when fitting the in-process PCA backend. Above it the fit streams, which bounds memory to about 110 MB at roughly nineteen times the wall clock. Raise it to match the node: a 60,000 x 120,849 cohort is 54 GB dense and streams under the default.	`8.0`
`max_project_memory_gb`	`float`	GB budget for one chunk when projecting. Peak resident memory runs to about three times this. Both are ignored by the flashpca backend.	`8.0`
`n_pcs`	`int`	Number of principal components	`50`
`k_min`	`int`	Minimum K for admixture	`2`
`k_max`	`int`	Maximum K for admixture	`10`
`embedding`	`str`	Embedding method ('phate', 'umap', 'tsne', 'diffusion_map')	`'phate'`
`embedding_params`	`Optional[Dict]`	Optional parameters for embedding	`None`
`embedding_input`	`str`	Which dataset to embed - 'fit', 'project', or 'both' (default)	`'both'`
`skip_pca`	`bool`	Skip PCA step	`False`
`skip_admixture`	`bool`	Skip admixture step	`False`
`skip_embedding`	`bool`	Skip embedding step	`False`
`skip_visualization`	`bool`	Skip embedding visualization step	`False`
`skip_pca_visualization`	`bool`	Skip PCA visualization step	`False`
`admix_group_column`	`Optional[str]`	Column for grouping in admixture barplots (None = use first colormap key)	`None`
`admix_within_group_order`	`Optional[str]`	Method for ordering samples within groups ('chron', 'tree', or None)	`'chron'`
`skip_metrics`	`bool`	Skip metrics computation	`False`
`admix_threads`	`Optional[int]`	Threads to use for neural admixture (None = auto-detect)	`None`
`admix_gpus`	`Optional[int]`	Number of GPUs for neural admixture (None = auto-detect)	`None`

Returns:

Name	Type	Description
`PipelineResult`	`PipelineResult`	The typed outputs of every stage that ran. Fields for skipped stages are None, except `pca`, which is always a `PCAStepResult` -- check its `.skipped` rather than truthiness. Figure families are empty rather than None.

Source code in src/manifold_genetics/pipeline/orchestrator.py

def run(
    self,
    n_pcs: int = 50,
    pca_backend: str = "python",
    max_fit_memory_gb: float = 8.0,
    max_project_memory_gb: float = 8.0,
    k_min: int = 2,
    k_max: int = 10,
    embedding: str = "phate",
    embedding_params: Optional[Dict] = None,
    embedding_input: str = "both",
    skip_pca: bool = False,
    skip_admixture: bool = False,
    skip_embedding: bool = False,
    skip_visualization: bool = False,
    skip_pca_visualization: bool = False,
    skip_admixture_visualization: bool = False,
    admix_group_column: Optional[str] = None,
    admix_within_group_order: Optional[str] = "chron",
    skip_metrics: bool = False,
    admix_threads: Optional[int] = None,
    admix_gpus: Optional[int] = None,
    admix_batch_size: Optional[int] = 400,
) -> PipelineResult:
    """
    Run full pipeline.

    Args:
        pca_backend: 'flashpca' (external binary) or 'python' (in process)
        max_fit_memory_gb: GB budget for the dense standardised matrix when
            fitting the in-process PCA backend. Above it the fit streams, which
            bounds memory to about 110 MB at roughly nineteen times the wall
            clock. Raise it to match the node: a 60,000 x 120,849 cohort is 54 GB
            dense and streams under the default.
        max_project_memory_gb: GB budget for one chunk when projecting. Peak
            resident memory runs to about three times this. Both are ignored by
            the flashpca backend.
        n_pcs: Number of principal components
        k_min: Minimum K for admixture
        k_max: Maximum K for admixture
        embedding: Embedding method ('phate', 'umap', 'tsne', 'diffusion_map')
        embedding_params: Optional parameters for embedding
        embedding_input: Which dataset to embed - 'fit', 'project', or 'both' (default)
        skip_pca: Skip PCA step
        skip_admixture: Skip admixture step
        skip_embedding: Skip embedding step
        skip_visualization: Skip embedding visualization step
        skip_pca_visualization: Skip PCA visualization step
        admix_group_column: Column for grouping in admixture barplots (None = use first colormap key)
        admix_within_group_order: Method for ordering samples within groups ('chron', 'tree', or None)
        skip_metrics: Skip metrics computation
        admix_threads: Threads to use for neural admixture (None = auto-detect)
        admix_gpus: Number of GPUs for neural admixture (None = auto-detect)

    Returns:
        PipelineResult: The typed outputs of every stage that ran. Fields
            for skipped stages are None, except ``pca``, which is always a
            ``PCAStepResult`` -- check its ``.skipped`` rather than
            truthiness. Figure families are empty rather than None.
    """
    failed = []

    io = self._io
    pca_cfg = PCAConfig(
        n_pcs=n_pcs,
        backend=pca_backend,
        max_fit_memory_gb=max_fit_memory_gb,
        max_project_memory_gb=max_project_memory_gb,
    )
    pca_paths = pca_output_paths(io, pca_cfg)
    # admix_group_column / admix_within_group_order are run()-time parameters,
    # not init-time ones — self._viz_config only carries the init-time
    # projection-plot columns, so those two fields are overridden per call.
    # Two run() calls with different values must not interfere with each
    # other, which a stored, mutated VizConfig on self would risk.
    viz_cfg = dataclasses.replace(
        self._viz_config,
        admix_group_column=admix_group_column,
        admix_within_group_order=admix_within_group_order,
    )

    # ---- Step 1: PCA ----
    if not skip_pca:
        logger.info("=" * 70)
        logger.info("STEP 1: PCA")
        logger.info("=" * 70)

        r_pca = run_pca_step(io, pca_cfg)
    else:
        # PCA skipped — resolve expected paths so embedding can still run
        r_pca = _resolve_skipped_pca(io, pca_cfg)

    # ---- Step 1.5: PCA Visualization (independent of PCA computation) ----
    pca_figures = ()
    if not skip_pca_visualization:
        logger.info("=" * 70)
        logger.info("STEP 1.5: PCA VISUALIZATION")
        logger.info("=" * 70)

        pca_file = pca_paths["project_pca"]

        if pca_file.exists():
            pca_viz_result = _run_viz(
                "pca_viz",
                failed,
                lambda: run_pca_viz_step(io, pca_file=pca_file, n_pcs=n_pcs),
            )
            if pca_viz_result is not None:
                pca_figures = tuple(pca_viz_result.figures)
        else:
            logger.warning(f"PCA file not found: {pca_file}")
            logger.warning("Run with --skip-pca=False to compute PCA first")

    # ---- Step 2: Admixture ----
    r_admix = None
    admixture_figures = {}
    if not skip_admixture:
        logger.info("=" * 70)
        logger.info("STEP 2: ADMIXTURE")
        logger.info("=" * 70)

        admix_cfg = AdmixtureConfig(
            k_min=k_min,
            k_max=k_max,
            threads=admix_threads,
            num_gpus=admix_gpus,
            batch_size=admix_batch_size,
        )

        r_admix = run_admixture_step(io, admix_cfg, backend=self.admixture_backend)

        # Admixture bar plot (placed in figures/admixture/)
        if not skip_admixture_visualization:
            admix_viz_result = _run_viz(
                "admixture_viz",
                failed,
                lambda: run_admixture_viz_step(io, viz_cfg, admixture=r_admix),
            )
            if admix_viz_result is not None:
                bars = admix_viz_result.figures[0] if admix_viz_result.figures else None
                if bars is not None:
                    admixture_figures["bars"] = bars

    # ---- Step 3: Embedding ----
    r_emb = None
    if not skip_embedding:
        logger.info("=" * 70)
        logger.info(f"STEP 3: EMBEDDING ({embedding.upper()})")
        logger.info("=" * 70)

        if r_pca.fit_pca is None and r_pca.project_pca is None:
            raise RuntimeError(
                "No PCA files found. PCA was skipped (--skip-pca) and no cached "
                f"output exists at {pca_paths['fit_pca']} or "
                f"{pca_paths['project_pca']}. Re-run without --skip-pca to "
                "compute PCA, or place existing PCA CSVs at those paths."
            )

        emb_cfg = EmbeddingConfig(
            method=embedding,
            input_mode=embedding_input,
            params=dict(embedding_params or {}),
        )

        # PCA outputs may have come from the step or been resolved from disk
        # under --skip-pca; either way the embedding step takes them as a result.
        r_emb = run_embedding_step(io, emb_cfg, pca=r_pca)

    # ---- Step 4: Embedding Visualization ----
    fit_embedding_figures = ()
    embedding_figures = ()
    projection_plot = None
    if not skip_visualization and not skip_embedding:
        logger.info("=" * 70)
        logger.info("STEP 4: EMBEDDING VISUALIZATION")
        logger.info("=" * 70)

        emb_viz_result = _run_viz(
            "embedding_viz",
            failed,
            lambda: run_embedding_viz_step(io, viz_cfg, embedding=r_emb, method=embedding),
        )
        if emb_viz_result is not None:
            if r_emb.fit_embedding_file is not None:
                fit_embedding_figures = tuple(emb_viz_result.fit_figures)
            embedding_figures = tuple(emb_viz_result.project_figures)
            if emb_viz_result.projection_plot is not None:
                projection_plot = emb_viz_result.projection_plot
            failed.extend(emb_viz_result.failed_substeps)

    # ---- Step 4.5: Admixture-Colored Embedding Visualization (requires embedding) ----
    if not skip_admixture_visualization and not skip_embedding and not skip_admixture:
        logger.info("=" * 70)
        logger.info("STEP 4.5: ADMIXTURE-COLORED EMBEDDING VISUALIZATION")
        logger.info("=" * 70)

        if r_emb is not None and r_admix is not None:
            admix_emb_viz_result = _run_viz(
                "admixture_embedding_viz",
                failed,
                lambda: run_admixture_embedding_viz_step(
                    io, embedding=r_emb, admixture=r_admix
                ),
            )
            if admix_emb_viz_result is not None:
                admix_emb = (
                    admix_emb_viz_result.figures[0] if admix_emb_viz_result.figures else None
                )
                if admix_emb is not None:
                    admixture_figures["admixture_colored_embedding"] = admix_emb
        else:
            logger.warning(
                "Skipping admixture-colored embedding visualization - missing embedding or admixture data"
            )

    # ---- Step 5: Metrics ----
    r_geo = None
    r_admix_metrics = None
    if not skip_metrics and not skip_embedding:
        logger.info("=" * 70)
        logger.info("STEP 5: METRICS")
        logger.info("=" * 70)

        metrics_paths = metrics_output_paths(io)
        # Created unconditionally: the output tree has always contained metrics/
        # even when neither metric runs.
        metrics_paths["geographic"].parent.mkdir(parents=True, exist_ok=True)

        # Geographic preservation
        if self.geographic_coords:
            r_geo = run_geographic_metrics_step(
                r_emb.embedding_file,
                self.geographic_coords,
                metrics_paths["geographic"],
            )

        # Admixture preservation
        if not skip_admixture and r_admix is not None:
            r_admix_metrics = run_admixture_metrics_step(
                r_emb.embedding_file,
                r_admix.q_prefix,
                range(k_min, k_max + 1),
                metrics_paths["admixture"],
            )

    # Summary
    logger.info("=" * 70)
    logger.info("PIPELINE COMPLETE")
    logger.info("=" * 70)
    logger.info(f"Output directory: {self.output_dir}")

    return PipelineResult(
        pca=r_pca,
        admixture=r_admix,
        embedding=r_emb,
        geographic_metrics=r_geo,
        admixture_metrics=r_admix_metrics,
        pca_figures=pca_figures,
        fit_embedding_figures=fit_embedding_figures,
        embedding_figures=embedding_figures,
        projection_plot=projection_plot,
        admixture_figures=admixture_figures,
        failed_steps=tuple(failed),
    )

manifold_genetics.pipeline.result.PipelineResult `dataclass` ¶

PipelineResult(pca=None, admixture=None, embedding=None, geographic_metrics=None, admixture_metrics=None, pca_figures=(), fit_embedding_figures=(), embedding_figures=(), projection_plot=None, admixture_figures=dict(), failed_steps=())

Typed outputs of a pipeline run. Replaces the loose results dict.

figures `property` ¶

figures

Every figure produced, in stage order.

metrics `property` ¶

metrics

The metrics mapping in the shape the CLI summary prints: {"geographic": {...}, "admixture": {"2": {...}}}; empty when neither ran.

PCA¶

manifold_genetics.pca.flashpca.PCA ¶

PCA(n_components=20, flashpca_path=None, force=False, backend=None, max_fit_memory_gb=8.0, max_project_memory_gb=8.0)

FlashPCA wrapper for principal component analysis.

Examples:

>>> # Fit PCA on reference data
>>> pca = PCA(n_components=50)
>>> pca_coords = pca.fit_transform(plink_prefix="data/hgdp")
>>> pca_coords.to_csv("pca_50.csv", index=False)
>>>
>>> # Project new samples onto reference PCA
>>> pca = PCA(n_components=50)
>>> pca.fit(plink_prefix="data/hgdp_ref")
>>> new_coords = pca.project(plink_prefix="data/ukbb")

Initialize PCA analyzer.

Parameters:

Name	Type	Description	Default
`n_components`	`int`	Number of principal components to compute	`20`
`flashpca_path`	`Optional[str]`	Path to flashpca executable (None = auto-detect)	`None`
`force`	`bool`	If True, recompute even if outputs exist	`False`
`backend`	`Optional[str]`	`"python"` uses the in-process implementation, which needs no binary; `"flashpca"` shells out to the external one. They match to 1.5e-7 and write the same artefact set (tests/integration/test_pca_flashpca_parity.py), so a model fitted by either is usable by the other. `None` (the default) means `"flashpca"` when `flashpca_path` was supplied and `"python"` otherwise -- so passing a path is never silently ignored, while a plain `PCA()` needs no binary.	`None`
`max_fit_memory_gb`	`float`	Budget for the dense standardised matrix when fitting. Above it the fit streams, which bounds memory to about 110 MB at roughly nineteen times the wall clock. Raise it on a large node: a 60,000 x 120,849 cohort is 54 GB dense, so it streams under the default and does not under 64.	`8.0`
`max_project_memory_gb`	`float`	Budget for one chunk when projecting. Lower it on a small machine; peak resident memory runs to about three times this figure. Both are ignored by the flashpca backend, which manages its own memory.	`8.0`

The binary is resolved only for the flashpca backend. Resolving it unconditionally would make merely constructing this object fail on a machine without it -- i.e. anywhere that is not Linux x86-64.

Source code in src/manifold_genetics/pca/flashpca.py

def __init__(
    self,
    n_components: int = 20,
    flashpca_path: Optional[str] = None,
    force: bool = False,
    backend: Optional[str] = None,
    max_fit_memory_gb: float = 8.0,
    max_project_memory_gb: float = 8.0,
):
    """
    Initialize PCA analyzer.

    Args:
        n_components: Number of principal components to compute
        flashpca_path: Path to flashpca executable (None = auto-detect)
        force: If True, recompute even if outputs exist
        backend: ``"python"`` uses the in-process implementation, which
            needs no binary; ``"flashpca"`` shells out to the external one.
            They match to 1.5e-7 and write the same artefact set
            (tests/integration/test_pca_flashpca_parity.py), so a model
            fitted by either is usable by the other. ``None`` (the default)
            means ``"flashpca"`` when ``flashpca_path`` was supplied and
            ``"python"`` otherwise -- so passing a path is never silently
            ignored, while a plain ``PCA()`` needs no binary.

        max_fit_memory_gb: Budget for the dense standardised matrix when
            fitting. Above it the fit streams, which bounds memory to about
            110 MB at roughly nineteen times the wall clock. Raise it on a
            large node: a 60,000 x 120,849 cohort is 54 GB dense, so it
            streams under the default and does not under 64.
        max_project_memory_gb: Budget for one chunk when projecting. Lower
            it on a small machine; peak resident memory runs to about three
            times this figure. Both are ignored by the flashpca backend,
            which manages its own memory.

    The binary is resolved only for the flashpca backend. Resolving it
    unconditionally would make merely constructing this object fail on a
    machine without it -- i.e. anywhere that is not Linux x86-64.
    """
    if backend is None:
        backend = "flashpca" if flashpca_path is not None else "python"
    if backend not in ("flashpca", "python"):
        raise ValueError(f"Unknown PCA backend {backend!r}; choose 'flashpca' or 'python'")

    self.n_components = n_components
    self.force = force
    self.backend = backend

    self.flashpca: Optional[str] = None
    self._py_backend: Optional[SklearnPCABackend] = None
    self._model: Optional[PCAModel] = None

    if backend == "flashpca":
        if flashpca_path is None:
            resolver = ToolResolver()
            flashpca_path = resolver.resolve_flashpca()
        self.flashpca = flashpca_path
        logger.debug(f"Using flashpca: {self.flashpca}")
    else:
        self._py_backend = SklearnPCABackend(
            n_components=n_components,
            max_fit_memory_gb=max_fit_memory_gb,
            max_project_memory_gb=max_project_memory_gb,
        )
        logger.debug("Using in-process Python PCA backend")

    # Fitted state
    self._is_fitted = False
    self._loadings_path: Optional[Path] = None
    self._meansd_path: Optional[Path] = None
    self._fit_output_dir: Optional[Path] = None

fit ¶

fit(plink_prefix, output_dir=None)

Fit PCA on reference data.

Parameters:

Name	Type	Description	Default
`plink_prefix`	`Union[str, Path]`	Path to PLINK file prefix (without extension)	required
`output_dir`	`Optional[Union[str, Path]]`	Directory to save outputs (default: ./pca_outputs)	`None`

Returns:

Type	Description
`PCA`	Self (for method chaining)

Source code in src/manifold_genetics/pca/flashpca.py

def fit(
    self,
    plink_prefix: Union[str, Path],
    output_dir: Optional[Union[str, Path]] = None,
) -> "PCA":
    """
    Fit PCA on reference data.

    Args:
        plink_prefix: Path to PLINK file prefix (without extension)
        output_dir: Directory to save outputs (default: ./pca_outputs)

    Returns:
        Self (for method chaining)
    """
    plink_prefix = validate_plink_files(plink_prefix)
    plink_prefix = Path(plink_prefix).expanduser()
    if not plink_prefix.is_absolute():
        plink_prefix = Path.cwd() / plink_prefix
    plink_prefix = Path(plink_prefix).expanduser()
    if not plink_prefix.is_absolute():
        plink_prefix = Path.cwd() / plink_prefix

    # Set default output directory
    if output_dir is None:
        output_dir = Path.cwd() / "pca_outputs"
    else:
        output_dir = Path(output_dir).expanduser()
        if not output_dir.is_absolute():
            output_dir = Path.cwd() / output_dir

    output_dir.mkdir(parents=True, exist_ok=True)
    self._fit_output_dir = output_dir

    if self.backend == "python":
        self._model = self._fit_python(plink_prefix, output_dir)
        self._is_fitted = True
        logger.info(f"PCA fitted with {self.n_components} components (python backend)")
        return self

    # Run FlashPCA fit
    output_prefix = output_dir / "fit"
    outputs = self._run_flashpca_fit(plink_prefix, output_prefix)

    # Store reference files for projection
    self._loadings_path = outputs["loadings"].resolve()
    self._meansd_path = outputs["meansd"].resolve()
    self._is_fitted = True

    logger.info(f"PCA fitted with {self.n_components} components")
    return self

project ¶

project(plink_prefix, output_path=None, output_dir=None)

Project samples onto fitted PCA space.

Parameters:

Name	Type	Description	Default
`plink_prefix`	`Union[str, Path]`	Path to PLINK file prefix	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save CSV output	`None`
`output_dir`	`Optional[Union[str, Path]]`	Optional directory for raw FlashPCA projection outputs	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and PCA coordinates (dim_1, dim_2, ...)

Source code in src/manifold_genetics/pca/flashpca.py

def project(
    self,
    plink_prefix: Union[str, Path],
    output_path: Optional[Union[str, Path]] = None,
    output_dir: Optional[Union[str, Path]] = None,
) -> pd.DataFrame:
    """
    Project samples onto fitted PCA space.

    Args:
        plink_prefix: Path to PLINK file prefix
        output_path: Optional path to save CSV output
        output_dir: Optional directory for raw FlashPCA projection outputs

    Returns:
        DataFrame with sample_id and PCA coordinates (dim_1, dim_2, ...)
    """
    if not self._is_fitted:
        raise RuntimeError("PCA not fitted. Call fit() first.")

    plink_prefix = validate_plink_files(plink_prefix)
    plink_prefix = Path(plink_prefix).expanduser()
    if not plink_prefix.is_absolute():
        plink_prefix = Path.cwd() / plink_prefix

    if self.backend == "python":
        coords = self._py_backend.project(plink_prefix, self._model)
        fids, iids = read_fam(plink_prefix)
        self._write_projection_pc(coords, fids, iids, plink_prefix, output_dir)
        df = self._coords_to_df(coords, iids)
        if output_path:
            write_embedding_csv(df, output_path)
        return df

    # Run FlashPCA projection
    if output_dir is None:
        output_dir = self._fit_output_dir or Path.cwd() / "pca_outputs"
    output_dir = Path(output_dir).expanduser()
    if not output_dir.is_absolute():
        output_dir = Path.cwd() / output_dir
    output_dir.mkdir(parents=True, exist_ok=True)

    # Use dataset-specific prefix so we don't reuse a cached projection from
    # a different PLINK dataset (e.g., fit vs. project subset).
    dataset_name = Path(plink_prefix).name
    output_prefix = output_dir / f"project_{dataset_name}"
    pc_file = self._run_flashpca_project(plink_prefix, output_prefix)

    # Convert to manylatents format
    df = self._convert_pc_to_csv(pc_file, plink_prefix)

    # Save if output path provided
    if output_path:
        write_embedding_csv(df, output_path)

    return df

fit_transform ¶

fit_transform(plink_prefix, output_path=None)

Fit PCA and transform the same data.

Parameters:

Name	Type	Description	Default
`plink_prefix`	`Union[str, Path]`	Path to PLINK file prefix	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save CSV output	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and PCA coordinates

Source code in src/manifold_genetics/pca/flashpca.py

def fit_transform(
    self, plink_prefix: Union[str, Path], output_path: Optional[Union[str, Path]] = None
) -> pd.DataFrame:
    """
    Fit PCA and transform the same data.

    Args:
        plink_prefix: Path to PLINK file prefix
        output_path: Optional path to save CSV output

    Returns:
        DataFrame with sample_id and PCA coordinates
    """
    plink_prefix = validate_plink_files(plink_prefix)

    # Determine output directory
    if output_path:
        output_dir = Path(output_path).parent / "pca_outputs"
    else:
        output_dir = Path.cwd() / "pca_outputs"
    output_dir = output_dir.expanduser()
    if not output_dir.is_absolute():
        output_dir = Path.cwd() / output_dir

    output_dir.mkdir(parents=True, exist_ok=True)

    if self.backend == "python":
        self._model = self._fit_python(plink_prefix, output_dir)
        self._is_fitted = True
        self._fit_output_dir = output_dir
        df = self._coords_to_df(self._model.fit_coords, self._model.fit_sample_ids)
        if output_path:
            write_embedding_csv(df, output_path)
        return df

    # Run FlashPCA fit
    output_dir = output_dir if output_dir.is_absolute() else (Path.cwd() / output_dir)
    output_prefix = output_dir / "fit"
    outputs = self._run_flashpca_fit(plink_prefix, output_prefix)

    # Convert to manylatents format
    df = self._convert_pc_to_csv(outputs["pc"], plink_prefix)

    # Save if output path provided
    if output_path:
        write_embedding_csv(df, output_path)

    # Update fitted state
    self._loadings_path = outputs["loadings"]
    self._meansd_path = outputs["meansd"]
    self._fit_output_dir = output_dir
    self._is_fitted = True

    return df

Embeddings¶

manifold_genetics.embeddings.phate.PHATE ¶

PHATE(n_components=2, knn=5, t='auto', decay=40, gamma=1.0, n_pca=100, n_landmark=2000, random_landmarking=False, random_state=42, n_jobs=-1, mds_solver='sgd', embed_batch_size=None)

Bases: EmbeddingBase

PHATE embedding for manifold learning.

Examples:

>>> # Basic usage
>>> phate_model = PHATE(n_components=2, knn=25)
>>> embedding = phate_model.fit_transform("pca_50.csv")
>>>
>>> # With custom parameters
>>> phate_model = PHATE(n_components=2, knn=50, t=15, gamma=1)
>>> embedding = phate_model.fit_transform("pca_50.csv", output_path="phate_2d.csv")

Initialize PHATE embedding.

Parameters:

Name	Type	Description	Default
`n_components`	`int`	Number of embedding dimensions	`2`
`knn`	`int`	Number of nearest neighbors	`5`
`t`	`Union[int, str]`	Diffusion time (int or 'auto')	`'auto'`
`decay`	`Optional[int]`	Alpha decay parameter for graph kernel	`40`
`gamma`	`float`	Informational distance parameter	`1.0`
`n_pca`	`Optional[int]`	Number of PCs to compute before PHATE (None to skip)	`100`
`n_landmark`	`Optional[int]`	Number of landmarks for efficiency	`2000`
`random_landmarking`	`bool`	Use random landmark selection instead of spectral clustering	`False`
`random_state`	`Optional[int]`	Random seed	`42`
`n_jobs`	`int`	Number of parallel jobs (-1 for all cores)	`-1`
`mds_solver`	`str`	MDS solver ('sgd' or 'smacof')	`'sgd'`
`embed_batch_size`	`Optional[int]`	Number of samples to transform at once (None for all)	`None`

Source code in src/manifold_genetics/embeddings/phate.py

def __init__(
    self,
    n_components: int = 2,
    knn: int = 5,
    t: Union[int, str] = "auto",
    decay: Optional[int] = 40,
    gamma: float = 1.0,
    n_pca: Optional[int] = 100,
    n_landmark: Optional[int] = 2000,
    random_landmarking: bool = False,
    random_state: Optional[int] = 42,
    n_jobs: int = -1,
    mds_solver: str = "sgd",
    embed_batch_size: Optional[int] = None,
):
    """
    Initialize PHATE embedding.

    Args:
        n_components: Number of embedding dimensions
        knn: Number of nearest neighbors
        t: Diffusion time (int or 'auto')
        decay: Alpha decay parameter for graph kernel
        gamma: Informational distance parameter
        n_pca: Number of PCs to compute before PHATE (None to skip)
        n_landmark: Number of landmarks for efficiency
        random_landmarking: Use random landmark selection instead of spectral clustering
        random_state: Random seed
        n_jobs: Number of parallel jobs (-1 for all cores)
        mds_solver: MDS solver ('sgd' or 'smacof')
        embed_batch_size: Number of samples to transform at once (None for all)
    """
    super().__init__(n_components=n_components, random_state=random_state)

    self.knn = knn
    self.t = t
    self.decay = decay
    self.gamma = gamma
    self.n_pca = n_pca
    self.n_landmark = n_landmark
    self.random_landmarking = random_landmarking
    self.n_jobs = n_jobs
    self.mds_solver = mds_solver
    self.embed_batch_size = embed_batch_size

    # Initialize PHATE model
    self.model = phate.PHATE(
        n_components=n_components,
        knn=knn,
        t=t,
        decay=decay,
        gamma=gamma,
        n_pca=n_pca,
        n_landmark=n_landmark,
        random_landmarking=random_landmarking,
        random_state=random_state,
        n_jobs=n_jobs,
        mds_solver=mds_solver,
        verbose=0,
    )

fit ¶

fit(X)

Fit PHATE on the data.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`PHATE`	Self (for method chaining)

Source code in src/manifold_genetics/embeddings/phate.py

def fit(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> "PHATE":
    """
    Fit PHATE on the data.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        Self (for method chaining)
    """
    X_array, self._sample_ids = self._load_input_data(X)

    logger.info(f"Fitting PHATE with knn={self.knn}, t={self.t}...")
    self.model.fit(X_array)
    self._is_fitted = True

    logger.info(f"✓ PHATE fitted on {len(X_array)} samples")
    return self

transform ¶

transform(X)

Transform data using fitted PHATE.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and PHATE coordinates

Source code in src/manifold_genetics/embeddings/phate.py

def transform(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> pd.DataFrame:
    """
    Transform data using fitted PHATE.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        DataFrame with sample_id and PHATE coordinates
    """
    if not self._is_fitted:
        raise RuntimeError("PHATE not fitted. Call fit() first.")

    X_array, sample_ids = self._load_input_data(X)
    n_samples = len(X_array)

    # Check if batch processing is needed
    if self.embed_batch_size is not None and n_samples > self.embed_batch_size:
        logger.info(
            f"🔄 BATCH MODE: Transforming {n_samples} samples with PHATE in batches of {self.embed_batch_size}..."
        )

        # Process in batches
        embeddings = []
        n_batches = int(np.ceil(n_samples / self.embed_batch_size))

        for i in range(n_batches):
            start_idx = i * self.embed_batch_size
            end_idx = min((i + 1) * self.embed_batch_size, n_samples)

            logger.info(
                f"  Processing batch {i + 1}/{n_batches} (samples {start_idx}-{end_idx})..."
            )
            batch_data = X_array[start_idx:end_idx]
            batch_embedding = self.model.transform(batch_data)
            embeddings.append(batch_embedding)

        # Concatenate all batches
        embedding = np.vstack(embeddings)
        logger.info(f"✓ Batch transformation complete: {embedding.shape}")
    else:
        # Process all at once
        if self.embed_batch_size is not None:
            logger.warning(
                f"embed_batch_size={self.embed_batch_size} set but not using batch mode "
                f"(data size {n_samples} <= batch size)"
            )
        logger.info(f"Transforming {n_samples} samples with PHATE (no batching)...")
        embedding = self.model.transform(X_array)

    return self._format_output(embedding, sample_ids)

fit_transform ¶

fit_transform(X, output_path=None)

Fit PHATE and transform data in one step.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save output CSV	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and PHATE coordinates

Source code in src/manifold_genetics/embeddings/phate.py

def fit_transform(
    self,
    X: Union[np.ndarray, pd.DataFrame, str, Path],
    output_path: Optional[Union[str, Path]] = None,
) -> pd.DataFrame:
    """
    Fit PHATE and transform data in one step.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)
        output_path: Optional path to save output CSV

    Returns:
        DataFrame with sample_id and PHATE coordinates
    """
    X_array, sample_ids = self._load_input_data(X)

    logger.info(f"Running PHATE (knn={self.knn}, t={self.t}) on {len(X_array)} samples...")
    embedding = self.model.fit_transform(X_array)
    self._is_fitted = True
    self._sample_ids = sample_ids

    logger.info(f"✓ PHATE embedding complete: {embedding.shape}")

    # Format output with sample_id column
    return self._format_output(embedding, sample_ids, output_path)

manifold_genetics.embeddings.umap.UMAP ¶

UMAP(n_components=2, n_neighbors=15, min_dist=0.1, metric='euclidean', random_state=42, n_epochs=None, learning_rate=1.0, n_jobs=-1)

Bases: EmbeddingBase

UMAP embedding for manifold learning.

Examples:

>>> # Basic usage
>>> umap_model = UMAP(n_components=2, n_neighbors=15)
>>> embedding = umap_model.fit_transform("pca_50.csv")
>>>
>>> # With custom parameters
>>> umap_model = UMAP(n_components=2, n_neighbors=50, min_dist=0.1)
>>> embedding = umap_model.fit_transform("pca_50.csv", output_path="umap_2d.csv")

Initialize UMAP embedding.

Parameters:

Name	Type	Description	Default
`n_components`	`int`	Number of embedding dimensions	`2`
`n_neighbors`	`int`	Number of nearest neighbors	`15`
`min_dist`	`float`	Minimum distance between points in embedding	`0.1`
`metric`	`str`	Distance metric to use	`'euclidean'`
`random_state`	`Optional[int]`	Random seed	`42`
`n_epochs`	`Optional[int]`	Number of training epochs (None = auto)	`None`
`learning_rate`	`float`	Learning rate for optimization	`1.0`
`n_jobs`	`int`	Number of parallel jobs (-1 for all cores)	`-1`

Source code in src/manifold_genetics/embeddings/umap.py

def __init__(
    self,
    n_components: int = 2,
    n_neighbors: int = 15,
    min_dist: float = 0.1,
    metric: str = "euclidean",
    random_state: Optional[int] = 42,
    n_epochs: Optional[int] = None,
    learning_rate: float = 1.0,
    n_jobs: int = -1,
):
    """
    Initialize UMAP embedding.

    Args:
        n_components: Number of embedding dimensions
        n_neighbors: Number of nearest neighbors
        min_dist: Minimum distance between points in embedding
        metric: Distance metric to use
        random_state: Random seed
        n_epochs: Number of training epochs (None = auto)
        learning_rate: Learning rate for optimization
        n_jobs: Number of parallel jobs (-1 for all cores)
    """
    super().__init__(n_components=n_components, random_state=random_state)

    self.n_neighbors = n_neighbors
    self.min_dist = min_dist
    self.metric = metric
    self.n_epochs = n_epochs
    self.learning_rate = learning_rate
    self.n_jobs = n_jobs

    # Initialize UMAP model
    self.model = umap.UMAP(
        n_components=n_components,
        n_neighbors=n_neighbors,
        min_dist=min_dist,
        metric=metric,
        random_state=random_state,
        n_epochs=n_epochs,
        learning_rate=learning_rate,
        verbose=False,
    )

fit ¶

fit(X)

Fit UMAP on the data.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`UMAP`	Self (for method chaining)

Source code in src/manifold_genetics/embeddings/umap.py

def fit(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> "UMAP":
    """
    Fit UMAP on the data.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        Self (for method chaining)
    """
    X_array, self._sample_ids = self._load_input_data(X)

    logger.info(f"Fitting UMAP with n_neighbors={self.n_neighbors}...")
    self.model.fit(X_array)
    self._is_fitted = True

    logger.info(f"✓ UMAP fitted on {len(X_array)} samples")
    return self

transform ¶

transform(X)

Transform data using fitted UMAP.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and UMAP coordinates

Source code in src/manifold_genetics/embeddings/umap.py

def transform(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> pd.DataFrame:
    """
    Transform data using fitted UMAP.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        DataFrame with sample_id and UMAP coordinates
    """
    if not self._is_fitted:
        raise RuntimeError("UMAP not fitted. Call fit() first.")

    X_array, sample_ids = self._load_input_data(X)

    logger.info(f"Transforming {len(X_array)} samples with UMAP...")
    embedding = self.model.transform(X_array)

    return self._format_output(embedding, sample_ids)

fit_transform ¶

fit_transform(X, output_path=None)

Fit UMAP and transform data in one step.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save output CSV	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and UMAP coordinates

Source code in src/manifold_genetics/embeddings/umap.py

def fit_transform(
    self,
    X: Union[np.ndarray, pd.DataFrame, str, Path],
    output_path: Optional[Union[str, Path]] = None,
) -> pd.DataFrame:
    """
    Fit UMAP and transform data in one step.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)
        output_path: Optional path to save output CSV

    Returns:
        DataFrame with sample_id and UMAP coordinates
    """
    X_array, sample_ids = self._load_input_data(X)

    logger.info(f"Running UMAP (n_neighbors={self.n_neighbors}) on {len(X_array)} samples...")
    embedding = self.model.fit_transform(X_array)
    self._is_fitted = True
    self._sample_ids = sample_ids

    logger.info(f"✓ UMAP embedding complete: {embedding.shape}")

    # Format output with sample_id column
    return self._format_output(embedding, sample_ids, output_path)

manifold_genetics.embeddings.tsne.TSNE ¶

TSNE(n_components=2, perplexity=30.0, learning_rate='auto', n_iter=1000, metric='euclidean', random_state=42, n_jobs=-1)

Bases: EmbeddingBase

t-SNE embedding for manifold learning.

Examples:

>>> # Basic usage
>>> tsne_model = TSNE(n_components=2, perplexity=30)
>>> embedding = tsne_model.fit_transform("pca_50.csv")
>>>
>>> # With custom parameters
>>> tsne_model = TSNE(n_components=2, perplexity=50, learning_rate=200)
>>> embedding = tsne_model.fit_transform("pca_50.csv", output_path="tsne_2d.csv")

Initialize t-SNE embedding.

Parameters:

Name	Type	Description	Default
`n_components`	`int`	Number of embedding dimensions	`2`
`perplexity`	`float`	Perplexity parameter (balance local vs global structure)	`30.0`
`learning_rate`	`Union[float, str]`	Learning rate for optimization ('auto' or float)	`'auto'`
`n_iter`	`int`	Number of optimization iterations	`1000`
`metric`	`str`	Distance metric to use	`'euclidean'`
`random_state`	`Optional[int]`	Random seed	`42`
`n_jobs`	`int`	Number of parallel jobs (-1 for all cores)	`-1`

Source code in src/manifold_genetics/embeddings/tsne.py

def __init__(
    self,
    n_components: int = 2,
    perplexity: float = 30.0,
    learning_rate: Union[float, str] = "auto",
    n_iter: int = 1000,
    metric: str = "euclidean",
    random_state: Optional[int] = 42,
    n_jobs: int = -1,
):
    """
    Initialize t-SNE embedding.

    Args:
        n_components: Number of embedding dimensions
        perplexity: Perplexity parameter (balance local vs global structure)
        learning_rate: Learning rate for optimization ('auto' or float)
        n_iter: Number of optimization iterations
        metric: Distance metric to use
        random_state: Random seed
        n_jobs: Number of parallel jobs (-1 for all cores)
    """
    super().__init__(n_components=n_components, random_state=random_state)

    self.perplexity = perplexity
    self.learning_rate = learning_rate
    self.n_iter = n_iter
    self.metric = metric
    self.n_jobs = n_jobs

    # Initialize t-SNE model
    self.model = SklearnTSNE(
        n_components=n_components,
        perplexity=perplexity,
        learning_rate=learning_rate,
        max_iter=n_iter,  # sklearn uses max_iter, not n_iter
        metric=metric,
        random_state=random_state,
        n_jobs=n_jobs,
        verbose=0,
    )

fit ¶

fit(X)

Fit t-SNE on the data.

Note: t-SNE does not support separate fit/transform. This method will store the data for later use in transform().

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`TSNE`	Self (for method chaining)

Source code in src/manifold_genetics/embeddings/tsne.py

def fit(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> "TSNE":
    """
    Fit t-SNE on the data.

    Note: t-SNE does not support separate fit/transform. This method
    will store the data for later use in transform().

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        Self (for method chaining)
    """
    X_array, self._sample_ids = self._load_input_data(X)
    self._fit_data = X_array  # Store for transform
    self._is_fitted = True

    logger.info(f"✓ t-SNE prepared for {len(X_array)} samples")
    return self

transform ¶

transform(X)

Transform data using t-SNE.

Note: t-SNE does not support out-of-sample extension. This will refit the model on the new data.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and t-SNE coordinates

Source code in src/manifold_genetics/embeddings/tsne.py

def transform(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> pd.DataFrame:
    """
    Transform data using t-SNE.

    Note: t-SNE does not support out-of-sample extension. This will
    refit the model on the new data.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        DataFrame with sample_id and t-SNE coordinates
    """
    logger.warning(
        "t-SNE does not support out-of-sample projection. " "Refitting on new data..."
    )
    return self.fit_transform(X)

fit_transform ¶

fit_transform(X, output_path=None)

Fit t-SNE and transform data in one step.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save output CSV	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and t-SNE coordinates

Source code in src/manifold_genetics/embeddings/tsne.py

def fit_transform(
    self,
    X: Union[np.ndarray, pd.DataFrame, str, Path],
    output_path: Optional[Union[str, Path]] = None,
) -> pd.DataFrame:
    """
    Fit t-SNE and transform data in one step.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)
        output_path: Optional path to save output CSV

    Returns:
        DataFrame with sample_id and t-SNE coordinates
    """
    X_array, sample_ids = self._load_input_data(X)

    logger.info(f"Running t-SNE (perplexity={self.perplexity}) on {len(X_array)} samples...")
    embedding = self.model.fit_transform(X_array)
    self._is_fitted = True
    self._sample_ids = sample_ids

    logger.info(f"✓ t-SNE embedding complete: {embedding.shape}")

    # Format output with sample_id column
    return self._format_output(embedding, sample_ids, output_path)

manifold_genetics.embeddings.diffusion_map.DiffusionMap ¶

DiffusionMap(n_components=2, knn=5, alpha=1.0, t=1, random_state=42)

Bases: EmbeddingBase

Diffusion Maps embedding for manifold learning.

Examples:

>>> # Basic usage
>>> dm_model = DiffusionMap(n_components=2, knn=25)
>>> embedding = dm_model.fit_transform("pca_50.csv")
>>>
>>> # With custom parameters
>>> dm_model = DiffusionMap(n_components=2, knn=50, alpha=1.0)
>>> embedding = dm_model.fit_transform("pca_50.csv", output_path="dm_2d.csv")

Initialize Diffusion Maps embedding.

Parameters:

Name	Type	Description	Default
`n_components`	`int`	Number of embedding dimensions	`2`
`knn`	`int`	Number of nearest neighbors for kernel	`5`
`alpha`	`float`	Normalization parameter (0=no normalization, 1=Laplacian normalization)	`1.0`
`t`	`int`	Diffusion time (powers of diffusion operator)	`1`
`random_state`	`Optional[int]`	Random seed	`42`

Source code in src/manifold_genetics/embeddings/diffusion_map.py

def __init__(
    self,
    n_components: int = 2,
    knn: int = 5,
    alpha: float = 1.0,
    t: int = 1,
    random_state: Optional[int] = 42,
):
    """
    Initialize Diffusion Maps embedding.

    Args:
        n_components: Number of embedding dimensions
        knn: Number of nearest neighbors for kernel
        alpha: Normalization parameter (0=no normalization, 1=Laplacian normalization)
        t: Diffusion time (powers of diffusion operator)
        random_state: Random seed
    """
    super().__init__(n_components=n_components, random_state=random_state)

    self.knn = knn
    self.alpha = alpha
    self.t = t

    # Storage for fitted model
    self._eigenvectors = None
    self._eigenvalues = None

fit ¶

fit(X)

Fit Diffusion Maps on the data.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`DiffusionMap`	Self (for method chaining)

Source code in src/manifold_genetics/embeddings/diffusion_map.py

def fit(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> "DiffusionMap":
    """
    Fit Diffusion Maps on the data.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        Self (for method chaining)
    """
    X_array, self._sample_ids = self._load_input_data(X)

    logger.info(f"Fitting Diffusion Maps with knn={self.knn}, alpha={self.alpha}...")

    # Compute diffusion map
    self._compute_diffusion_map(X_array)
    self._is_fitted = True

    logger.info(f"✓ Diffusion Maps fitted on {len(X_array)} samples")
    return self

transform ¶

transform(X)

Transform data using fitted Diffusion Maps.

Note: Diffusion maps does not naturally support out-of-sample extension. This will use Nyström extension method.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and diffusion coordinates

Source code in src/manifold_genetics/embeddings/diffusion_map.py

def transform(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> pd.DataFrame:
    """
    Transform data using fitted Diffusion Maps.

    Note: Diffusion maps does not naturally support out-of-sample extension.
    This will use Nyström extension method.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        DataFrame with sample_id and diffusion coordinates
    """
    if not self._is_fitted:
        raise RuntimeError("Diffusion Maps not fitted. Call fit() first.")

    logger.warning(
        "Diffusion Maps out-of-sample extension is experimental. "
        "Consider refitting on combined data."
    )

    # For now, just refit (proper Nyström extension is complex)
    return self.fit_transform(X)

fit_transform ¶

fit_transform(X, output_path=None)

Fit Diffusion Maps and transform data in one step.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save output CSV	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and diffusion coordinates

Source code in src/manifold_genetics/embeddings/diffusion_map.py

def fit_transform(
    self,
    X: Union[np.ndarray, pd.DataFrame, str, Path],
    output_path: Optional[Union[str, Path]] = None,
) -> pd.DataFrame:
    """
    Fit Diffusion Maps and transform data in one step.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)
        output_path: Optional path to save output CSV

    Returns:
        DataFrame with sample_id and diffusion coordinates
    """
    X_array, sample_ids = self._load_input_data(X)

    logger.info(
        f"Running Diffusion Maps (knn={self.knn}, alpha={self.alpha}) "
        f"on {len(X_array)} samples..."
    )

    # Compute diffusion map
    embedding = self._compute_diffusion_map(X_array)
    self._is_fitted = True
    self._sample_ids = sample_ids

    logger.info(f"✓ Diffusion Maps embedding complete: {embedding.shape}")

    # Format output with sample_id column
    return self._format_output(embedding, sample_ids, output_path)

manifold_genetics.embeddings.base.EmbeddingBase ¶

EmbeddingBase(n_components=2, random_state=42)

Bases: ABC

Abstract base class for dimensionality reduction methods.

All embedding methods should inherit from this class and implement the fit(), transform(), and fit_transform() methods.

Initialize embedding method.

Parameters:

Name	Type	Description	Default
`n_components`	`int`	Number of dimensions for the embedding	`2`
`random_state`	`Optional[int]`	Random seed for reproducibility	`42`

Source code in src/manifold_genetics/embeddings/base.py

def __init__(self, n_components: int = 2, random_state: Optional[int] = 42):
    """
    Initialize embedding method.

    Args:
        n_components: Number of dimensions for the embedding
        random_state: Random seed for reproducibility
    """
    self.n_components = n_components
    self.random_state = random_state
    self._is_fitted = False

fit `abstractmethod` ¶

fit(X)

Fit the embedding method on the data.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`EmbeddingBase`	Self (for method chaining)

Source code in src/manifold_genetics/embeddings/base.py

@abstractmethod
def fit(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> "EmbeddingBase":
    """
    Fit the embedding method on the data.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        Self (for method chaining)
    """

transform `abstractmethod` ¶

transform(X)

Transform data using the fitted embedding.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and embedding coordinates

Source code in src/manifold_genetics/embeddings/base.py

@abstractmethod
def transform(self, X: Union[np.ndarray, pd.DataFrame, str, Path]) -> pd.DataFrame:
    """
    Transform data using the fitted embedding.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)

    Returns:
        DataFrame with sample_id and embedding coordinates
    """

fit_transform `abstractmethod` ¶

fit_transform(X, output_path=None)

Fit and transform data in one step.

Parameters:

Name	Type	Description	Default
`X`	`Union[ndarray, DataFrame, str, Path]`	Input data (numpy array, DataFrame, or path to CSV)	required
`output_path`	`Optional[Union[str, Path]]`	Optional path to save output CSV	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with sample_id and embedding coordinates

Source code in src/manifold_genetics/embeddings/base.py

@abstractmethod
def fit_transform(
    self,
    X: Union[np.ndarray, pd.DataFrame, str, Path],
    output_path: Optional[Union[str, Path]] = None,
) -> pd.DataFrame:
    """
    Fit and transform data in one step.

    Args:
        X: Input data (numpy array, DataFrame, or path to CSV)
        output_path: Optional path to save output CSV

    Returns:
        DataFrame with sample_id and embedding coordinates
    """

Admixture¶

manifold_genetics.admixture.neural.NeuralAdmixture ¶

NeuralAdmixture(k_min=2, k_max=10, neural_admixture_path=None, force=False, threads=None, num_gpus=None, batch_size=None, backend=None)

Neural admixture wrapper for ancestry inference.

This class provides a user-friendly API and delegates the actual computation to a backend. By default, it uses NeuralAdmixtureBackend for real computation, but you can inject different backends for testing.

Examples:

>>> # Default: real neural-admixture computation
>>> admix = NeuralAdmixture(k_min=2, k_max=5)
>>> admix.fit("data/hgdp.plink", output_dir="admixture/")
>>> q_files = admix.transform("data/ukbb.plink", output_prefix="admixture/ukbb")
>>>
>>> # For testing: use precomputed backend
>>> from manifold_genetics.admixture.backends import PrecomputedAdmixtureBackend
>>> backend = PrecomputedAdmixtureBackend(k_min=2, k_max=3)
>>> admix = NeuralAdmixture(backend=backend)
>>> q_files = admix.fit_transform("data/test.plink", output_prefix="test/output")

Initialize Neural Admixture analyzer.

Parameters:

Name	Type	Description	Default
`k_min`	`int`	Minimum number of ancestral populations	`2`
`k_max`	`int`	Maximum number of ancestral populations	`10`
`neural_admixture_path`	`Optional[str]`	Path to executable (None = auto-detect)	`None`
`force`	`bool`	If True, retrain even if models exist	`False`
`threads`	`Optional[int]`	Number of threads to use. If None, attempts to detect available CPUs (respecting SLURM/HPC limits).	`None`
`num_gpus`	`Optional[int]`	Number of GPUs to use. If None, uses 1 when CUDA is available and 0 otherwise.	`None`
`batch_size`	`Optional[int]`	Batch size for training and inference. If None, uses neural-admixture defaults.	`None`
`backend`	`Optional[AdmixtureBackend]`	Optional backend to use for computation. If None, uses NeuralAdmixtureBackend (real computation). For testing, you can inject PrecomputedAdmixtureBackend or FakeAdmixtureBackend.	`None`

Source code in src/manifold_genetics/admixture/neural.py

def __init__(
    self,
    k_min: int = 2,
    k_max: int = 10,
    neural_admixture_path: Optional[str] = None,
    force: bool = False,
    threads: Optional[int] = None,
    num_gpus: Optional[int] = None,
    batch_size: Optional[int] = None,
    backend: Optional[AdmixtureBackend] = None,
):
    """
    Initialize Neural Admixture analyzer.

    Args:
        k_min: Minimum number of ancestral populations
        k_max: Maximum number of ancestral populations
        neural_admixture_path: Path to executable (None = auto-detect)
        force: If True, retrain even if models exist
        threads: Number of threads to use. If None, attempts to detect
                 available CPUs (respecting SLURM/HPC limits).
        num_gpus: Number of GPUs to use. If None, uses 1 when CUDA is
                  available and 0 otherwise.
        batch_size: Batch size for training and inference. If None, uses
                    neural-admixture defaults.
        backend: Optional backend to use for computation. If None, uses
                 NeuralAdmixtureBackend (real computation). For testing,
                 you can inject PrecomputedAdmixtureBackend or FakeAdmixtureBackend.
    """
    self.k_min = k_min
    self.k_max = k_max
    self.force = force

    # Use provided backend or create default NeuralAdmixtureBackend
    if backend is None:
        backend = NeuralAdmixtureBackend(
            k_min=k_min,
            k_max=k_max,
            force=force,
            neural_admixture_path=neural_admixture_path,
            threads=threads,
            num_gpus=num_gpus,
            batch_size=batch_size,
        )

    self.backend = backend
    logger.debug(f"Using backend: {type(backend).__name__}")

fit ¶

fit(plink_prefix, output_dir=None, model_name='fit')

Train neural admixture models on reference data.

Parameters:

Name	Type	Description	Default
`plink_prefix`	`Union[str, Path]`	Path to PLINK file prefix (without extension)	required
`output_dir`	`Optional[Union[str, Path]]`	Directory to save model outputs (default: ./admixture_outputs)	`None`
`model_name`	`str`	Name for the model (used in output filenames)	`'fit'`

Returns:

Type	Description
`NeuralAdmixture`	Self (for method chaining)

Source code in src/manifold_genetics/admixture/neural.py

def fit(
    self,
    plink_prefix: Union[str, Path],
    output_dir: Optional[Union[str, Path]] = None,
    model_name: str = "fit",
) -> "NeuralAdmixture":
    """
    Train neural admixture models on reference data.

    Args:
        plink_prefix: Path to PLINK file prefix (without extension)
        output_dir: Directory to save model outputs (default: ./admixture_outputs)
        model_name: Name for the model (used in output filenames)

    Returns:
        Self (for method chaining)
    """
    if output_dir is None:
        output_dir = Path.cwd() / "admixture_outputs"

    # Delegate to backend
    self.backend.fit(plink_prefix, output_dir, model_name)

    logger.info(f"Neural admixture fitted for K={self.k_min} to {self.k_max}")
    return self

transform ¶

transform(plink_prefix, output_prefix)

Infer ancestry proportions on new samples.

Parameters:

Name	Type	Description	Default
`plink_prefix`	`Union[str, Path]`	Path to PLINK file prefix	required
`output_prefix`	`Union[str, Path]`	Prefix for output CSV files (will create .K.csv)	required

Returns:

Name	Type	Description
	`Dict[int, Path]`	Dictionary mapping K values to CSV file paths
`Example`	`Dict[int, Path]`	{2: Path("output.2.csv"), 3: Path("output.3.csv")}

Source code in src/manifold_genetics/admixture/neural.py

def transform(
    self,
    plink_prefix: Union[str, Path],
    output_prefix: Union[str, Path],
) -> Dict[int, Path]:
    """
    Infer ancestry proportions on new samples.

    Args:
        plink_prefix: Path to PLINK file prefix
        output_prefix: Prefix for output CSV files (will create <prefix>.K.csv)

    Returns:
        Dictionary mapping K values to CSV file paths
        Example: {2: Path("output.2.csv"), 3: Path("output.3.csv")}
    """
    # Delegate to backend
    return self.backend.transform(plink_prefix, output_prefix)

fit_transform ¶

fit_transform(plink_prefix, output_prefix)

Train models and infer on the same data.

Parameters:

Name	Type	Description	Default
`plink_prefix`	`Union[str, Path]`	Path to PLINK file prefix	required
`output_prefix`	`Union[str, Path]`	Prefix for output CSV files	required

Returns:

Type	Description
`Dict[int, Path]`	Dictionary mapping K values to CSV file paths

Source code in src/manifold_genetics/admixture/neural.py

def fit_transform(
    self,
    plink_prefix: Union[str, Path],
    output_prefix: Union[str, Path],
) -> Dict[int, Path]:
    """
    Train models and infer on the same data.

    Args:
        plink_prefix: Path to PLINK file prefix
        output_prefix: Prefix for output CSV files

    Returns:
        Dictionary mapping K values to CSV file paths
    """
    # Delegate to backend
    return self.backend.fit_transform(plink_prefix, output_prefix)

Visualisation¶

manifold_genetics.visualization.plotting.visualize ¶

visualize(embedding, labels, colormap, output_dir=None, output_prefix='embedding', dataset_prefix='')

Create all standard visualization plots.

Generates a plot for each label column in the colormap.

Parameters:

Name	Type	Description	Default
`embedding`	`Union[DataFrame, str, Path]`	DataFrame or path to embedding CSV	required
`labels`	`Union[DataFrame, str, Path]`	DataFrame or path to labels CSV	required
`colormap`	`Union[Dict, str, Path]`	Dict or path to colormap JSON	required
`output_dir`	`Optional[Union[str, Path]]`	Directory to save plots (default: current directory)	`None`
`output_prefix`	`str`	Prefix for output filenames	`'embedding'`
`dataset_prefix`	`str`	Prefix for dataset type (e.g., "fit_" or "project_")	`''`

Returns:

Type	Description
`List[Path]`	List of paths to saved figures

Source code in src/manifold_genetics/visualization/plotting.py

def visualize(
    embedding: Union[pd.DataFrame, str, Path],
    labels: Union[pd.DataFrame, str, Path],
    colormap: Union[Dict, str, Path],
    output_dir: Optional[Union[str, Path]] = None,
    output_prefix: str = "embedding",
    dataset_prefix: str = "",
) -> List[Path]:
    """
    Create all standard visualization plots.

    Generates a plot for each label column in the colormap.

    Args:
        embedding: DataFrame or path to embedding CSV
        labels: DataFrame or path to labels CSV
        colormap: Dict or path to colormap JSON
        output_dir: Directory to save plots (default: current directory)
        output_prefix: Prefix for output filenames
        dataset_prefix: Prefix for dataset type (e.g., "fit_" or "project_")

    Returns:
        List of paths to saved figures
    """
    if output_dir is None:
        output_dir = Path.cwd()
    else:
        output_dir = Path(output_dir)

    output_dir.mkdir(parents=True, exist_ok=True)

    # Load colormap to get label columns
    if isinstance(colormap, (str, Path)):
        colormap_dict = read_colormap(colormap)
    else:
        colormap_dict = colormap

    # Generate plots for each label column
    output_paths = []
    for label_col in colormap_dict.keys():
        output_path = output_dir / f"{dataset_prefix}{output_prefix}_by_{label_col}.png"

        plot_embedding(
            embedding=embedding,
            labels=labels,
            colormap={label_col: colormap_dict[label_col]},
            output_path=output_path,
            title=f"Embedding colored by {label_col}",
        )

        output_paths.append(output_path)

    logger.info(f"Generated {len(output_paths)} visualization plots")
    return output_paths

manifold_genetics.visualization.plotting.plot_embedding ¶

plot_embedding(embedding, labels, colormap, output_path, title=None, figsize=(6, 4), point_size=4.0, alpha=0.6, show_legend=True)

Plot 2D embedding colored by labels.

Parameters:

Name	Type	Description	Default
`embedding`	`Union[DataFrame, str, Path]`	DataFrame or path to embedding CSV (sample_id, dim_1, dim_2)	required
`labels`	`Union[DataFrame, str, Path]`	DataFrame or path to labels CSV (sample_id, label_columns)	required
`colormap`	`Union[Dict, str, Path]`	Dict or path to colormap JSON {label_col: {value: color}}	required
`output_path`	`Union[str, Path]`	Path to save figure	required
`title`	`Optional[str]`	Optional plot title	`None`
`figsize`	`tuple`	Figure size (width, height)	`(6, 4)`
`point_size`	`float`	Size of scatter plot points	`4.0`
`alpha`	`float`	Transparency of points	`0.6`
`show_legend`	`bool`	Whether to show legend	`True`

Returns:

Type	Description
`Path`	Path to saved figure

Source code in src/manifold_genetics/visualization/plotting.py

def plot_embedding(
    embedding: Union[pd.DataFrame, str, Path],
    labels: Union[pd.DataFrame, str, Path],
    colormap: Union[Dict, str, Path],
    output_path: Union[str, Path],
    title: Optional[str] = None,
    figsize: tuple = (6, 4),
    point_size: float = 4.0,
    alpha: float = 0.6,
    show_legend: bool = True,
) -> Path:
    """
    Plot 2D embedding colored by labels.

    Args:
        embedding: DataFrame or path to embedding CSV (sample_id, dim_1, dim_2)
        labels: DataFrame or path to labels CSV (sample_id, label_columns)
        colormap: Dict or path to colormap JSON {label_col: {value: color}}
        output_path: Path to save figure
        title: Optional plot title
        figsize: Figure size (width, height)
        point_size: Size of scatter plot points
        alpha: Transparency of points
        show_legend: Whether to show legend

    Returns:
        Path to saved figure
    """
    # Load data
    if isinstance(embedding, (str, Path)):
        embedding_df = read_embedding_csv(embedding)
    else:
        embedding_df = embedding

    if isinstance(labels, (str, Path)):
        labels_df = read_labels_csv(labels)
    else:
        labels_df = labels

    # Reset index if sample_id is the index (from read_labels_csv)
    if labels_df.index.name == "sample_id":
        labels_df = labels_df.reset_index()

    if isinstance(colormap, (str, Path)):
        colormap_dict = read_colormap(colormap)
    else:
        colormap_dict = colormap

    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    # Merge embedding with labels on sample_id to ensure correct alignment
    merged_df = embedding_df.merge(labels_df, on="sample_id", how="inner")

    # Create figure
    fig, axes = plt.subplots(
        1, len(colormap_dict), figsize=(figsize[0] * len(colormap_dict), figsize[1])
    )
    if len(colormap_dict) == 1:
        axes = [axes]

    # Plot for each label column in colormap
    for ax, (label_col, color_dict) in zip(axes, colormap_dict.items()):
        if label_col not in merged_df.columns:
            logger.warning(f"Column '{label_col}' not found in labels, skipping")
            continue

        # FIRST: Plot samples with missing data in gray (background layer)
        missing_mask = merged_df[label_col].isna()
        if missing_mask.sum() > 0:
            ax.scatter(
                merged_df.loc[missing_mask, "dim_1"],
                merged_df.loc[missing_mask, "dim_2"],
                s=point_size,
                alpha=alpha * 0.5,  # More transparent for background
                color="lightgray",
                edgecolors="none",
                label="Unknown",
                rasterized=True,
                zorder=1,  # Low z-order for background
            )

        # SECOND: Plot each color group separately (foreground layer)
        # Use the ordering from the color_dict (Python 3.7+ preserves insertion order)
        # Plot in REVERSE order so that the first items in colormap appear on top
        warn_about_unmatched_labels(merged_df[label_col], color_dict, label_col)
        color_groups = [k for k in color_dict.keys() if k in merged_df[label_col].values]
        for label in reversed(color_groups):
            mask = merged_df[label_col] == label
            label_data = merged_df[mask]
            color = color_dict.get(label, "#D3D3D3")  # Default to gray

            ax.scatter(
                label_data["dim_1"],
                label_data["dim_2"],
                s=point_size,
                alpha=alpha,
                color=color,
                edgecolors="none",
                label=label,
                rasterized=True,
                zorder=2,  # Higher z-order for foreground
            )

        # THIRD: Create legend using Patches instead of scatter handles
        # This ensures legend follows the color_dict order
        legend_elements = [
            Patch(facecolor=color_dict[g], label=g)
            for g in color_dict.keys()
            if g in merged_df[label_col].values
        ]
        if merged_df[label_col].isna().any():
            legend_elements.append(Patch(facecolor="lightgray", label="Unknown"))

        # Remove ticks, tick labels, axis labels, and titles
        ax.set_xticks([])
        ax.set_yticks([])
        ax.set_xlabel("")
        ax.set_ylabel("")
        ax.set_title("")

        # Add legend if requested and reasonable number of labels
        if show_legend and len(legend_elements) <= 50:
            ax.legend(
                handles=legend_elements,
                fontsize=8,
                framealpha=0.9,
                loc="center left",
                bbox_to_anchor=(1.02, 0.5),
            )

    # No titles needed

    plt.tight_layout()
    plt.savefig(output_path, dpi=300, bbox_inches="tight")
    plt.close()

    logger.info(f"Saved embedding plot: {output_path}")
    return output_path

Python API¶

Pipeline¶

manifold_genetics.pipeline.runner.run_pipeline ¶

manifold_genetics.pipeline.configfile.load_config ¶

manifold_genetics.pipeline.orchestrator.Pipeline ¶

run ¶

manifold_genetics.pipeline.result.PipelineResult dataclass ¶

figures property ¶

metrics property ¶

PCA¶

manifold_genetics.pca.flashpca.PCA ¶

fit ¶

project ¶

fit_transform ¶

Embeddings¶

manifold_genetics.embeddings.phate.PHATE ¶

fit ¶

transform ¶

fit_transform ¶

manifold_genetics.embeddings.umap.UMAP ¶

fit ¶

transform ¶

fit_transform ¶

manifold_genetics.embeddings.tsne.TSNE ¶

fit ¶

transform ¶

fit_transform ¶

manifold_genetics.embeddings.diffusion_map.DiffusionMap ¶

fit ¶

transform ¶

fit_transform ¶

manifold_genetics.embeddings.base.EmbeddingBase ¶

fit abstractmethod ¶

transform abstractmethod ¶

fit_transform abstractmethod ¶

Admixture¶

manifold_genetics.admixture.neural.NeuralAdmixture ¶

fit ¶

transform ¶

fit_transform ¶

Visualisation¶

manifold_genetics.visualization.plotting.visualize ¶

manifold_genetics.visualization.plotting.plot_embedding ¶

manifold_genetics.pipeline.result.PipelineResult `dataclass` ¶

figures `property` ¶

metrics `property` ¶

fit `abstractmethod` ¶

transform `abstractmethod` ¶

fit_transform `abstractmethod` ¶