index_ops

pykmhelpers.pipeline.index_ops

Build and merge k-mer indexes from index definitions or span registries.

IndexOps is the entry point; the other modules are its components.

IndexOps

Orchestrates k-mer index build and merge operations against a kmindex registry.

Loads an index definition or a span registry (see sources), builds the missing sub-indexes, merges partial indexes into their target and deletes the merged parts. In plan/dry-run modes, commands are only generated and can be exported as a shell script via write_script().

Parameters:

Name	Type	Description	Default
config	IndexOpsConfig	Runtime configuration governing paths, build parameters, filtering, and execution behaviour. See IndexOpsConfig.	required

Attributes:

Name	Type	Description
config	IndexOpsConfig	The resolved configuration (paths are converted to absolute paths on construction).
work_dir	str	Absolute path to the working directory.
asset_dir	str	<work_dir>/assets - output location for generated shell scripts.
log_dir	str	<work_dir>/logs - log file destination.
kmindex_registry_dir	str	Path to the kmindex registry directory.
kmindex_data_dir	str	Path to the folder that holds index data.
timestamp	str	YYYYmmdd_HHMMSS string captured at construction.

Source code in pykmhelpers/pipeline/index_ops/ops.py

class IndexOps:
    """Orchestrates k-mer index build and merge operations against a kmindex registry.

    Loads an index definition or a span registry (see ``sources``), builds
    the missing sub-indexes, merges partial indexes into their target and
    deletes the merged parts. In plan/dry-run modes, commands are only
    generated and can be exported as a shell script via ``write_script()``.

    Args:
        config: Runtime configuration governing paths, build parameters,
            filtering, and execution behaviour.  See ``IndexOpsConfig``.

    Attributes:
        config (IndexOpsConfig): The resolved configuration (paths are
            converted to absolute paths on construction).
        work_dir (str): Absolute path to the working directory.
        asset_dir (str): ``<work_dir>/assets`` - output location for generated
            shell scripts.
        log_dir (str): ``<work_dir>/logs`` - log file destination.
        kmindex_registry_dir (str): Path to the kmindex registry directory.
        kmindex_data_dir (str): Path to the folder that holds index data.
        timestamp (str): ``YYYYmmdd_HHMMSS`` string captured at construction.
    """

    def __init__(self, config: IndexOpsConfig) -> None:
        self._config = config
        self._config.workdir = os.path.realpath(config.workdir)
        self._config.index_data_folder = os.path.realpath(config.index_data_folder)
        self._config.registry_dir = os.path.realpath(config.registry_dir)
        self._timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

        # Kept across runs: sample files are resolved relative to their definition file
        self._dbs = DbCache()
        self._script = ScriptRecorder(self.work_dir)
        self._session: Optional[str] = None

        logger.debug(f"Init {type(self).__name__}")
        logger.debug("workdir: " + self.work_dir)
        logger.debug("registry_dir: " + self.kmindex_registry_dir)
        logger.debug("asset_dir: " + self.asset_dir)
        logger.debug("data_dir: " + self.kmindex_data_dir)

        for d in (
            self.log_dir,
            self.asset_dir,
            self.kmindex_registry_dir,
            self.kmindex_data_dir,
        ):
            os.makedirs(d, exist_ok=True)

    # ---
    # PROPERTIES

    @property
    def config(self) -> IndexOpsConfig:
        return self._config

    @property
    def work_dir(self) -> str:
        return self.config.workdir

    @property
    def asset_dir(self) -> str:
        return os.path.join(self.work_dir, "assets")

    @property
    def log_dir(self) -> str:
        return os.path.join(self.work_dir, "logs")

    @property
    def kmindex_registry_dir(self) -> str:
        return self.config.registry_dir

    @property
    def kmindex_data_dir(self) -> str:
        return self.config.index_data_folder

    @property
    def timestamp(self) -> str:
        return self._timestamp

    # ---
    # PUBLIC METHODS

    def write_script(self) -> None:
        """Write the commands collected by the last ``run()`` to ``asset_dir``.

        One script per group of parts (``<db>_g<span>_<session>.sh``: builds,
        merge, then cleanup), plus ``kmhelpers_apply.sh`` running them in
        order. With ``config.session_assets``, they go to
        ``asset_dir/<session>/`` and ``asset_dir/kmhelpers_apply.sh`` runs
        that session. Existing files are backed up with a ``.bak`` suffix.
        """
        run_dir = self._run_asset_dir
        self._script.write(run_dir)
        if run_dir != self.asset_dir:
            runner = os.path.join(run_dir, ScriptRecorder.RUNNER_NAME)
            self._script.write_entry(self.asset_dir, runner)

    def run(
        self, path: str, mode: ApplyMode, fail_on_error: bool = False
    ) -> ApplyResult:
        """Apply an index definition or span registry file to the kmindex registry.

        Builds every selected sub-index missing from the registry, then
        performs the merges declared by a span registry.

        Args:
            path: Path to a YAML or JSON file containing either an
                ``IndexDefinition`` or a span registry.
            mode: Execution mode - controls whether to dry-run, plan, or apply.
            fail_on_error: Abort this run on the first build or merge error
                instead of continuing and returning ``PARTIAL``.

        Returns:
            An ``ApplyResult`` with the overall status and a per-index details
            dict. Status is ``SUCCESS`` when all operations complete without
            error, ``PARTIAL`` when some fail (only when
            ``fail_on_error=False``), or ``FAILED`` when a fatal error occurs.
        """
        path = os.path.realpath(path)
        self._script = ScriptRecorder(self.work_dir)
        self._session = (
            os.path.basename(os.path.dirname(path))
            if self.config.session_assets
            else None
        )
        report = RunReport(path, mode)
        idt = IndexDefinitionTools()

        input_type, data = detect_input(path, idt)
        report.result.input_type = input_type
        if data is None or input_type is ApplyInputType.UNKNOWN:
            logger.error("Could not retrieve data type.")
            report.result.status = ApplyStatus.FAILED
            return report.result

        os.makedirs(self._run_asset_dir, exist_ok=True)
        builder = IndexBuilder(
            workdir=self.work_dir,
            registry_name=self.kmindex_registry_dir,
            data_folder=self.kmindex_data_dir,
            log_folder=self.log_dir,
            assets_folder=os.path.relpath(self._run_asset_dir, self.work_dir),
        )
        try:
            plan = SOURCES[input_type](self.config, self._dbs).load(path, idt, data)
        except Exception as e:
            Log.handle_exception(logger, e, f"Failed to load definition file '{path}'")
            report.result.status = ApplyStatus.FAILED
            return report.result

        layout = LayoutStore.next_to(path)
        executor = IndexExecutor(builder, mode, self.config, self._script)
        samples = SampleResolver(
            self.config.sample_rootpath, mode.checks_files, self._dbs
        )

        for i in plan.definitions:
            assert i.name  # WorkPlan only holds named definitions
            logger.info(f"► Processing index definition '{i.name}'...")
            if executor.has_index(i.name):
                logger.info(f"  └── {i.name} found in registry: skip")
                report.record(i.name, ApplyStatus.NONE)
                continue
            try:
                self._script.select(plan.script_of(i.name))
                self._build(
                    report, executor, samples, i, layout, plan.script_of(i.name)
                )
            except Exception as e:
                msg = f"   Failed to build index '{i.name}'"
                if report.fail(e, msg, i.name, fail_on_error):
                    layout.save()
                    return report.result

        for target, parts in plan.merges.items():
            try:
                self._merge(report, executor, target, parts)
            except Exception as e:
                msg = f"Failed to merge index '{target}'"
                if report.fail(e, msg, target, fail_on_error):
                    layout.save()
                    return report.result

        layout.save()
        return report.finish()

    # ---
    # PRIVATE METHODS

    @property
    def _run_asset_dir(self) -> str:
        """Asset directory of the current run: per session, or flat."""
        if self._session:
            return os.path.join(self.asset_dir, self._session)
        return self.asset_dir

    def _store_params(
        self, layout: LayoutStore, i: IndexDefinition, params: BuildParams
    ) -> None:
        """Keep the layout in line with what is actually built."""
        layout.set_partition_count(i.span, params.partitions)
        layout.set_minim_size(params.minim_size)
        i.partition_count = params.partitions

    def _resolve_params(
        self, i: IndexDefinition, sample_count: int, layout: LayoutStore
    ) -> BuildParams:
        params = resolve_build_params(i, sample_count, self.config, layout)
        # Keeps size estimates of i in line with what is built
        i.partition_count = params.partitions
        return params

    def _build(
        self,
        report: RunReport,
        executor: IndexExecutor,
        samples: SampleResolver,
        i: IndexDefinition,
        layout: LayoutStore,
        target: str,
    ) -> None:
        """Build index ``i`` and record its outcome in ``report``."""
        # Partitions must be known before estimating the size
        params = self._resolve_params(i, i.sample_count, layout)
        size = i.get_stored_size()
        report.add_span_stats(i.span, i.sample_count, size.byte_count)
        logger.info(f"  └── Sample count: {i.sample_count}")
        logger.info(f"  └── Estimated build size: {size}")

        validate_definition(i)
        assert i.name is not None
        if not executor.claim(i.name):
            report.record(i.name, ApplyStatus.NONE)
            return

        fof, issues = samples.build_fof(i)
        sample_count = fof.get_sample_count()
        result = None
        if sample_count == 0:
            logger.warning(
                f"{indent_prefix(logger)}Skipping index '{i.name}' as no sample was added to it"
            )
        else:
            # Samples may have been skipped since the estimate: resolve for the actual count
            if params is None or sample_count != i.sample_count:
                params = self._resolve_params(i, sample_count, layout)
            params = executor.align_with(target, params)
            self._store_params(layout, i, params)
            result = executor.build(i, fof, params)

        if not result:
            status = ApplyStatus.FAILED if issues else ApplyStatus.NONE
            report.record(i.name, status, issues=issues)
        elif not executor.executes or result.get("return_code", -1) == 0:
            report.record(i.name, ApplyStatus.SUCCESS, issues=issues)
        else:
            error = f"error_code={result['return_code']}"
            report.record(i.name, ApplyStatus.FAILED, error=error, issues=issues)

    def _merge(
        self, report: RunReport, executor: IndexExecutor, target: str, parts: list[str]
    ) -> None:
        """Merge ``parts`` into ``target``, record the outcome and delete the parts."""
        missing = executor.missing_parts(parts)
        if missing:
            logger.warning(
                f"Cannot merge '{target}' due to some sub-indexes missing: "
                f"{summarize_names(missing)}"
            )
            report.record(
                target, ApplyStatus.FAILED, error=f"Missing sub-indexes: {missing}"
            )
            return

        threads = self.config.kmindex_threads or os.cpu_count() or 1
        self._script.select(target)
        result = executor.merge(target, parts, threads)
        if not result or "command" not in result:
            raise Exception("Malformed result")
        report.fold_merge(target, parts)
        executor.finalize_merge(target, parts)

write_script()

Write the commands collected by the last run() to asset_dir.

One script per group of parts (<db>_g<span>_<session>.sh: builds, merge, then cleanup), plus kmhelpers_apply.sh running them in order. With config.session_assets, they go to asset_dir/<session>/ and asset_dir/kmhelpers_apply.sh runs that session. Existing files are backed up with a .bak suffix.

Source code in pykmhelpers/pipeline/index_ops/ops.py

def write_script(self) -> None:
    """Write the commands collected by the last ``run()`` to ``asset_dir``.

    One script per group of parts (``<db>_g<span>_<session>.sh``: builds,
    merge, then cleanup), plus ``kmhelpers_apply.sh`` running them in
    order. With ``config.session_assets``, they go to
    ``asset_dir/<session>/`` and ``asset_dir/kmhelpers_apply.sh`` runs
    that session. Existing files are backed up with a ``.bak`` suffix.
    """
    run_dir = self._run_asset_dir
    self._script.write(run_dir)
    if run_dir != self.asset_dir:
        runner = os.path.join(run_dir, ScriptRecorder.RUNNER_NAME)
        self._script.write_entry(self.asset_dir, runner)

run(path, mode, fail_on_error=False)

Apply an index definition or span registry file to the kmindex registry.

Builds every selected sub-index missing from the registry, then performs the merges declared by a span registry.

Parameters:

Name	Type	Description	Default
path	str	Path to a YAML or JSON file containing either an IndexDefinition or a span registry.	required
mode	ApplyMode	Execution mode - controls whether to dry-run, plan, or apply.	required
fail_on_error	bool	Abort this run on the first build or merge error instead of continuing and returning PARTIAL.	False

Returns:

Type	Description
ApplyResult	An ApplyResult with the overall status and a per-index details
ApplyResult	dict. Status is SUCCESS when all operations complete without
ApplyResult	error, PARTIAL when some fail (only when
ApplyResult	fail_on_error=False), or FAILED when a fatal error occurs.

Source code in pykmhelpers/pipeline/index_ops/ops.py

def run(
    self, path: str, mode: ApplyMode, fail_on_error: bool = False
) -> ApplyResult:
    """Apply an index definition or span registry file to the kmindex registry.

    Builds every selected sub-index missing from the registry, then
    performs the merges declared by a span registry.

    Args:
        path: Path to a YAML or JSON file containing either an
            ``IndexDefinition`` or a span registry.
        mode: Execution mode - controls whether to dry-run, plan, or apply.
        fail_on_error: Abort this run on the first build or merge error
            instead of continuing and returning ``PARTIAL``.

    Returns:
        An ``ApplyResult`` with the overall status and a per-index details
        dict. Status is ``SUCCESS`` when all operations complete without
        error, ``PARTIAL`` when some fail (only when
        ``fail_on_error=False``), or ``FAILED`` when a fatal error occurs.
    """
    path = os.path.realpath(path)
    self._script = ScriptRecorder(self.work_dir)
    self._session = (
        os.path.basename(os.path.dirname(path))
        if self.config.session_assets
        else None
    )
    report = RunReport(path, mode)
    idt = IndexDefinitionTools()

    input_type, data = detect_input(path, idt)
    report.result.input_type = input_type
    if data is None or input_type is ApplyInputType.UNKNOWN:
        logger.error("Could not retrieve data type.")
        report.result.status = ApplyStatus.FAILED
        return report.result

    os.makedirs(self._run_asset_dir, exist_ok=True)
    builder = IndexBuilder(
        workdir=self.work_dir,
        registry_name=self.kmindex_registry_dir,
        data_folder=self.kmindex_data_dir,
        log_folder=self.log_dir,
        assets_folder=os.path.relpath(self._run_asset_dir, self.work_dir),
    )
    try:
        plan = SOURCES[input_type](self.config, self._dbs).load(path, idt, data)
    except Exception as e:
        Log.handle_exception(logger, e, f"Failed to load definition file '{path}'")
        report.result.status = ApplyStatus.FAILED
        return report.result

    layout = LayoutStore.next_to(path)
    executor = IndexExecutor(builder, mode, self.config, self._script)
    samples = SampleResolver(
        self.config.sample_rootpath, mode.checks_files, self._dbs
    )

    for i in plan.definitions:
        assert i.name  # WorkPlan only holds named definitions
        logger.info(f"► Processing index definition '{i.name}'...")
        if executor.has_index(i.name):
            logger.info(f"  └── {i.name} found in registry: skip")
            report.record(i.name, ApplyStatus.NONE)
            continue
        try:
            self._script.select(plan.script_of(i.name))
            self._build(
                report, executor, samples, i, layout, plan.script_of(i.name)
            )
        except Exception as e:
            msg = f"   Failed to build index '{i.name}'"
            if report.fail(e, msg, i.name, fail_on_error):
                layout.save()
                return report.result

    for target, parts in plan.merges.items():
        try:
            self._merge(report, executor, target, parts)
        except Exception as e:
            msg = f"Failed to merge index '{target}'"
            if report.fail(e, msg, target, fail_on_error):
                layout.save()
                return report.result

    layout.save()
    return report.finish()

BuildParams dataclass

Resolved kmindex build parameters for one index.

partitions and minim_size are index invariants: indexes are only mergeable when they share them, so they come from the layout once resolved. chunk_size is set when the open-files ceiling can't fit every sample in one kmtricks build; the build is then split into chunks of at most that many samples, which are merged.

Source code in pykmhelpers/pipeline/index_ops/sizing.py

@dataclass(frozen=True)
class BuildParams:
    """Resolved kmindex build parameters for one index.

    ``partitions`` and ``minim_size`` are index invariants: indexes are only
    mergeable when they share them, so they come from the layout once
    resolved. ``chunk_size`` is set when the open-files ceiling can't fit
    every sample in one kmtricks build; the build is then split into chunks
    of at most that many samples, which are merged.
    """

    threads: int
    partitions: int
    chunk_size: Optional[int] = None
    minim_size: int = 10

ApplyInputType

Bases: str, Enum

Type of the input file passed to an apply operation.

Values

UNKNOWN: File type could not be determined. SPAN_REGISTRY: Input is a span registry file describing how partial indexes should be merged. INDEX_DEFINITION: Input is an index definition file describing one or more sub-indexes to build. NONE: No input file has been inspected yet.

Source code in pykmhelpers/pipeline/index_ops/types.py

class ApplyInputType(str, Enum):
    """Type of the input file passed to an apply operation.

    Values:
        UNKNOWN: File type could not be determined.
        SPAN_REGISTRY: Input is a span registry file describing how partial
            indexes should be merged.
        INDEX_DEFINITION: Input is an index definition file describing one or
            more sub-indexes to build.
        NONE: No input file has been inspected yet.
    """

    UNKNOWN = "unknown"
    SPAN_REGISTRY = "span registry"
    INDEX_DEFINITION = "index definition"
    NONE = "none"

ApplyMode

Bases: int, Enum

Source code in pykmhelpers/pipeline/index_ops/types.py

class ApplyMode(int, Enum):
    DRY_RUN = 0
    PLAN = 1
    APPLY = 2
    APPLY_SHOW_PROGRESS = 3

    @property
    def executes(self) -> bool:
        """True when commands are actually run (not just planned)."""
        return self >= ApplyMode.APPLY

    @property
    def checks_files(self) -> bool:
        """True when sample files must exist on this machine."""
        return self > ApplyMode.DRY_RUN

executes property

True when commands are actually run (not just planned).

checks_files property

True when sample files must exist on this machine.

ApplyResult dataclass

Result of an apply operation.

Attributes:

Name	Type	Description
status	ApplyStatus	Overall outcome of the operation.
input_type	ApplyInputType	Detected type of the input file that was processed.
details	dict	Per-index outcome strings keyed by index name, plus an "input_file" entry with the resolved path of the source file.

Source code in pykmhelpers/pipeline/index_ops/types.py

@dataclass
class ApplyResult:
    """Result of an apply operation.

    Attributes:
        status: Overall outcome of the operation.
        input_type: Detected type of the input file that was processed.
        details: Per-index outcome strings keyed by index name, plus an
            ``"input_file"`` entry with the resolved path of the source file.
    """

    status: ApplyStatus = ApplyStatus.NONE
    input_type: ApplyInputType = ApplyInputType.NONE
    mode: ApplyMode = ApplyMode.APPLY
    details: dict = field(default_factory=dict)

ApplyStatus

Bases: str, Enum

Status of an apply operation.

Source code in pykmhelpers/pipeline/index_ops/types.py

class ApplyStatus(str, Enum):
    """Status of an apply operation."""

    SUCCESS = "SUCCESS"
    PARTIAL = "PARTIAL"
    FAILED = "FAILED"
    NONE = "NONE"

IndexOpsConfig dataclass

Configuration for an IndexOps instance.

Attributes:

Name	Type	Description
workdir	str	Root working directory. Created automatically if absent.
index_data_folder	str	Directory that holds the raw index data files.
registry_dir	str	Path to the kmindex registry used to track sub-indexes.
minimizer_length	int	K-mer minimizer size passed to the builder. Defaults to 10.
sample_rootpath	Optional[str]	Optional prefix prepended to every sample file path in an index definition. Useful when paths are stored relative to a root that differs from the current working directory.
kmindex_threads	Optional[int]	Number of threads passed to kmindex build commands. None (or 0) auto-sizes threads per index from limits (system RAM/ulimit by default, scaled by safety_margin) via auto_params. Defaults to None.
kmindex_skip_compression	bool	When True, intermediate files are not compressed during the build. Defaults to False.
kmindex_build_from	Optional[str]	Override the parent index for all build operations, replacing the value declared in each index definition.
filter_spans	Optional[list[int]]	If set, only process index definitions whose span value is in this list.
filter_names	Optional[list[str]]	If set, only process index definitions whose name is in this list.
on_existing	str	Behaviour when a sub-index folder already exists on disk but is not registered. Passed directly to the builder (e.g. "fail", "register"). Defaults to "fail".
limits	Optional[str]	JSON line of resource limits (ram, files, threads, focus) forwarded to auto_params when kmindex_threads is unset. Any key omitted is auto-detected from the system. Defaults to None (all keys auto-detected).
safety_margin	float	Fraction of a detected system limit to use for any key missing from limits. Defaults to 0.9.
max_chunks	int	Cap on the number of chunks a single index is split into. A bigger chunk needs more open files in the merge stage, so the cap is paid for with threads. 0 disables it, leaving the chunk size to the open-files limit alone. Defaults to 200.
session_assets	bool	When True, generated assets and scripts go to assets/<session>/, where the session is the folder name of the input file, and assets/kmhelpers_apply.sh runs the last session. Defaults to False (flat assets/).

Source code in pykmhelpers/pipeline/index_ops/types.py

@dataclass
class IndexOpsConfig:
    """Configuration for an ``IndexOps`` instance.

    Attributes:
        workdir: Root working directory.  Created automatically if absent.
        index_data_folder: Directory that holds the raw index data files.
        registry_dir: Path to the kmindex registry used to track sub-indexes.
        minimizer_length: K-mer minimizer size passed to the builder.
            Defaults to ``10``.
        sample_rootpath: Optional prefix prepended to every sample file path
            in an index definition.  Useful when paths are stored relative to
            a root that differs from the current working directory.
        kmindex_threads: Number of threads passed to kmindex build commands.
            ``None`` (or ``0``) auto-sizes threads per index from ``limits``
            (system RAM/ulimit by default, scaled by ``safety_margin``) via
            ``auto_params``. Defaults to ``None``.
        kmindex_skip_compression: When ``True``, intermediate files are not
            compressed during the build.  Defaults to ``False``.
        kmindex_build_from: Override the parent index for all build operations,
            replacing the value declared in each index definition.
        filter_spans: If set, only process index definitions whose span value is
            in this list.
        filter_names: If set, only process index definitions whose name is in
            this list.
        on_existing: Behaviour when a sub-index folder already exists on disk but is not registered.
            Passed directly to the builder (e.g. ``"fail"``, ``"register"``).
            Defaults to ``"fail"``.
        limits: JSON line of resource limits (``ram``, ``files``, ``threads``,
            ``focus``) forwarded to ``auto_params`` when
            ``kmindex_threads`` is unset. Any key omitted is auto-detected
            from the system. Defaults to ``None`` (all keys auto-detected).
        safety_margin: Fraction of a detected system limit to use for any
            key missing from ``limits``. Defaults to ``0.9``.
        max_chunks: Cap on the number of chunks a single index is split into.
            A bigger chunk needs more open files in the merge stage, so the
            cap is paid for with threads. ``0`` disables it, leaving the chunk
            size to the open-files limit alone. Defaults to ``200``.
        session_assets: When ``True``, generated assets and scripts go to
            ``assets/<session>/``, where the session is the folder name of the
            input file, and ``assets/kmhelpers_apply.sh`` runs the last
            session. Defaults to ``False`` (flat ``assets/``).
    """

    workdir: str
    index_data_folder: str
    registry_dir: str
    minimizer_length: int = 10
    sample_rootpath: Optional[str] = None
    kmindex_threads: Optional[int] = None
    kmindex_skip_compression: bool = False
    kmindex_build_from: Optional[str] = None
    filter_spans: Optional[list[int]] = None
    filter_names: Optional[list[str]] = None
    on_existing: str = "fail"
    partition_count: Optional[int] = None
    limits: Optional[str] = None
    safety_margin: float = 0.9
    max_chunks: int = 200
    session_assets: bool = False

clamp_partitions(count)

Apply the kmtricks partition floor; 0 (auto) is left untouched.

Source code in pykmhelpers/pipeline/index_ops/sizing.py

def clamp_partitions(count: int) -> int:
    """Apply the kmtricks partition floor; 0 (auto) is left untouched."""
    return max(count, KMTRICKS_MIN_PARTITIONS) if count > 0 else count