Adding an analysis step

This guide walks through one complete contribution: a new analysis that the pipeline can run, configure, validate, and document. Follow it in order. It aims to cover everything needed to write an analysis in one place. Browse the rest of the documentation when you want more depth on a particular concept.

What you are building

An analysis contribution is always two classes:

An analysis class holding the scientific calculation. It knows nothing about files, the command line, or configuration. You give it the data it needs; it does the calculation and returns results.
A step class describing how that calculation enters the pipeline: which files it reads, which parameters it accepts, and where its results are written.

The split keeps the science separate from the plumbing. Your analysis class deals only with biological objects and the values you calculate from them; everything about files, paths, parameters, scheduling, and validation lives in the step class, where ASOkai already knows how to handle it. In practice that means you can develop and test the calculation on its own, and that changes to how the pipeline runs do not affect your calculations.

You write the calculation and the declarations that describe it. Argument parsing, file loading and writing, path construction, dependency resolution, and the step’s reference documentation page are all handled by ASOkai’s own code, driven by those declarations.

Before you start

These are the files you create or change for a built-in contribution:

Path	What it is for
`src/ASOkai/Analysis/_<name>.py`	Your analysis class.
`src/ASOkai/Analysis/__init__.py`	Where you export the analysis.
`src/ASOkai/_pipeline/steps/analysis/<name>.py`	Your step class.
`src/ASOkai/_pipeline/catalog.py`	Where you add the step to the registry.
`src/ASOkai/_pipeline/config.schema.yaml`	Where the schema for your step’s parameters is defined.
`examples/config.example.yaml`	A realistic example value for each new field your analysis adds to the schema.
`docs/source/steps/<step-name>.rst`	A three-line page whose contents are autogenerated. Add extra information if needed.

A copyable starting point is kept at examples/contributor_steps/analysis_step.py. It is reproduced at the end of this guide. Copy it once you understand the choices below.

An independently installed plugin follows the same design but imports its contracts from ASOkai.Plugins and registers through entry points instead of the catalog. Choose a plugin when the analysis should ship as its own installable package, for example work that is maintained outside the ASOkai repository or released on its own schedule, rather than being merged into ASOkai itself. See Publishing a plugin instead.

Step 1: choose your two base classes

Two decisions have to be made before you write the code.

Decision 1: how your calculation consumes units

First decide whether your calculation looks at one unit at a time or needs the whole collection together. That choice fixes the base class of your analysis and what you implement in it.

An analysis unit is one biological object to analyze: a target site, a therapeutic, or a therapeutic-site binding. The pipeline hands your analysis a collection of them.

Note

These three unit types are the ones ASOkai currently supports. If your analysis needs a different kind of unit, raise it with Arash Ayat before you start, so it can be added rather than worked around.

	`PerUnitAnalysis`	`BatchAnalysis`
Choose when	Each unit is analyzed independently of the others.	The whole collection is needed at once: one external command over all units, a calculation that compares units with each other, or a calculation that is much faster in one batch than per unit.
Function you implement	`analyze_unit(unit)`	`analyze_batch(units, *, n_workers)`
Called	Once per unit.	Once in total.
Parallelism	Handled by ASOkai.	Yours to implement, using the worker budget you are given.
Result order	Input order, guaranteed.	Whatever you return.

Warning

Most analyses are per-unit. Choose BatchAnalysis only when a per-unit implementation would be wrong or wasteful, not merely because you want control of the loop. No built-in analysis currently uses it; if yours is the first, inform Arash Ayat.

Decision 2: what your step writes

Then decide what the step writes to disk. That choice fixes the base class of your step and whether you write any output code at all.

	`TableAnalysisStep`	`AnalysisStep`
Choose when	Your results are a table: one row per result, the same columns every time.	Your results are anything else: FASTA, JSON, a format a downstream tool requires, or several files of which none is a result table.
Writing	Done for you, from the rows your `analyze_unit` or `analyze_batch` returns.	You implement `write_results`.
Your results are	`dict` rows, one key per declared column.	Any shape you need for writing them in `write_results`.

TableAnalysisStep is an AnalysisStep — it adds the CSV convention on top. If a table fits, use it: you get column validation, type checking, and duplicate-index detection at no cost. Use the plain AnalysisStep when the result is not tabular.

Both base classes can write more than one file. A TableAnalysisStep declares its extra outputs after the primary CSV and returns their values from the additional_output_values function, so several outputs alone are not a reason to avoid it. What decides the choice is whether the first output is a result table.

The two decisions are independent. A BatchAnalysis can feed a TableAnalysisStep, and a PerUnitAnalysis can feed a plain AnalysisStep.

What each base class gives you

Members marked required are the ones you implement or define. The others already work as described; override one only when your analysis needs different behavior. The one exception is run: never override or reimplement it on either analysis class, without consulting with Arash Ayat beforehand.

The analysis classes

Shared by both Analysis classes:

Member	Purpose
`analysis_unit_type` required	Class attribute naming the unit kind: `"site"`, `"therapeutic"`, or `"binding"`.
`__init__`	Receives the scientific context your calculation needs. Validate its arguments here.
`run(units, , n_workers)` Don’t override*	Provided. Collects results and applies the worker budget. Never override it. Everything you need is available through the hooks below; overriding `run` bypasses the parallel execution the base class sets up.

Only on PerUnitAnalysis:

Member	Purpose
`analyze_unit(unit)` required	Returns zero or more results for one unit. Returning a list of one is normal; returning several is how one unit produces several rows, and returning an empty list drops the unit. The framework passes the unit positionally, so your parameter may be named `site`, `therapeutic`, or another name that fits the calculation.
`uses_worker_processes`	Set `True` when the calculation holds the interpreter lock. See Running in parallel.
`worker_chunk_size`, `target_tasks_per_worker`	Tune how units are grouped into worker tasks. The defaults are usually right.

Only on BatchAnalysis:

Member	Purpose
`analyze_batch(units, , n_workers)` required*	Returns all results for the whole collection, using the worker budget however your backend needs it.

The step classes

Shared by both step base classes:

Member	Purpose
`name`, `description` required	The registered step name and its one-line summary. Both appear in the CLI and in the generated documentation page.
`analysis_cls` required	Your analysis class.
`spec` required	The `StepSpec` collecting parameters, inputs, and outputs.
`analysis_units(args)` required	Returns the units to analyze, taken from already-parsed inputs.
`analysis_kwargs(args)`	Returns a mapping passed to your analysis constructor. Omit it when the constructor takes nothing.
`validate_parameters(values)`	Only for relationships between parameters that the schema cannot express.
`run_from_args(args)` Don’t override	Provided and marked final. Constructs the analysis, supplies its units and worker count, and calls the result-writing hook.
`listed`	Set `False` for an internal step used by a task. It is omitted from the default step and analysis listings, remains selectable by name, and appears in verbose listings.

Only on AnalysisStep:

Member

Purpose

write_results(args, analysis, results) required

Writes the results through your declared outputs. After a PerUnitAnalysis, results is an iterator that arrives while the workers are still running and yields each result once. Pass it straight to a writer that takes an iterable, such as SeqIO.write or csv.writer.writerows, and an analysis returning far more results than units never collects them all in memory. Call list(results) when you need them twice or need their number up front, and only when the results are small enough to hold.

Only on TableAnalysisStep:

Member	Purpose
`additional_output_values(analysis, results)`	Only when you declare outputs beyond the primary CSV. Returns a mapping from each extra `FileOutput` to its value. The primary CSV is written for you, so `write_results` is already implemented.

Step 2: write the analysis class

Put the class in src/ASOkai/Analysis/_<name>.py and export it from ASOkai.Analysis. Start from ExampleAnalysis in the template at the end of this page and replace its calculation.

The built-in IntrinsicFeaturesAnalysis shows the same shape filled in:

class IntrinsicFeaturesAnalysis(PerUnitAnalysis):
    """
    Analyzes intrinsic features for each target site.

    Available features:
        - **GC_content**: The proportion of Guanine (G) and Cytosine (C)
          bases (between 0 and 1).
        - **T_count**: The count of Thymine (T) bases.
    """

    analysis_unit_type: ClassVar[AnalysisUnitType] = "site"

    def __init__(self, features: Iterable[str] | None = None) -> None:
        selected = list(self.supported_features if features is None else features)
        unknown = sorted(set(selected) - set(self.supported_features))
        if unknown:
            raise ValueError(f"Unknown intrinsic features: {unknown}.")
        self.features = tuple(selected)

    def analyze_unit(self, site: Site) -> list[dict[str, Scalar]]:
        sequence = str(site.sequence).upper()
        return [{
            "site_id": site.id,
            "GC_content": ...,
            "T_count": ...,
        }]

Four conventions are visible there, and each matters.

Declare the unit type. analysis_unit_type tells the catalog what your analysis consumes.

Take context through the constructor, not through globals. Anything the calculation needs beyond the unit itself — a target gene, a threshold, a selection of columns — becomes a constructor argument. The step supplies it from analysis_kwargs. Validate it here, so a direct library user gets the same error the pipeline would give.

Return complete results. For a table step, each result is a dictionary containing every declared column, including the index columns. Because a list is returned, one unit may produce one row, several rows, or no rows at all. When every unit yields exactly one row, as IntrinsicFeaturesAnalysis does above with one row per site, that is the typical shape of a table analysis.

Document each column with a named bullet. The generated reference reads your class docstring and matches each - **column_name**: description bullet to a declared CSV column. A column without a bullet still works but arrives in the documentation with no meaning attached.

Running in parallel

The person running the pipeline requests a worker count through resources.<step-name>.n_workers. For example:

resources:
  my-analysis:
    n_workers: 1

For PerUnitAnalysis, one worker runs directly without a worker pool. A BatchAnalysis receives the same budget and manages its backend’s parallelism. The setting configures the budget for that step and can be changed for another pipeline invocation. Analysis steps expose the standard --n-workers argument automatically; do not declare it as a ScalarParam or disable supports_multiple_workers on an analysis step.

PerUnitAnalysis.run gives you parallelism without any work on your part, but the right kind depends on your calculation:

Your calculation	What to do
Pure Python, or anything holding the interpreter lock	Set `uses_worker_processes = True`. Threads cannot overlap this work and may run it slower than one worker.
Waits on input or output, or calls a C library that releases the lock	Leave the default. Threads are cheaper than processes.

With uses_worker_processes, workers are forked on Linux when the parent has a single Python thread and is not itself a worker; anything else falls back to threads automatically. Forked workers inherit your analysis and its units through copy-on-write memory, so nothing is copied or serialized on the way in — but results travel back to the parent, so they must be picklable.

While workers run, treat the analysis and its units as read-only. Changes made in a worker do not reach the parent. Open database connections and other stateful handles inside the worker rather than reusing an inherited one.

Step 3: write the step class

Put the step in src/ASOkai/_pipeline/steps/analysis/<name>.py. Start from ExampleAnalysisStep in the template at the end of this page and replace its names and declarations.

Here is the step for the analysis above, trimmed to its structure:

class IntrinsicFeaturesStep(TableAnalysisStep):
    name = "compute-intrinsic-features"
    description = "Computes nucleotide composition and CpG features for each target site."
    analysis_cls = IntrinsicFeaturesAnalysis

    features_param = ScalarParam(
        "features",
        str,
        config="parameters.analysis.intrinsic_features.features",
        default="all",
        doc="Comma-separated intrinsic feature columns; 'all' selects every feature.",
    )
    target_gene_input = FileInput(
        "target_gene",
        source=CreateTargetGeneStep.target_gene_output,
        doc="Serialized target gene object from create-target-gene.",
    )
    intrinsic_features_output = FileOutput(
        "intrinsic_features",
        file_type=CsvFile(
            index={"site_id": str},
            columns={"GC_content": float, "T_count": int},
        ),
        destination=OutputPathTemplate(...),
        doc="Intrinsic features per target site.",
    )
    spec = StepSpec(
        params=[features_param],
        inputs=[target_gene_input],
        outputs=[intrinsic_features_output],
    )

    def analysis_kwargs(self, args):
        return {"features": self._features(args.features)}

    def analysis_units(self, args):
        return args.target_gene.sites

Each part of that class has a job.

Name and describe the step. name is how the step is selected on the command line and in configuration; use lowercase words joined by hyphens. description is the one-line summary shown by ASOkai list steps and at the top of the generated page, so write it for someone deciding whether this step is the one they want.

Point at the analysis. analysis_cls names the class from Step 2. The step constructs it for you, so never instantiate it yourself.

Give each parameter, input, and output its own attribute. Write the declaration once, give it a name on the class, and then list that same name in spec. The attribute is the declaration, not a description of one, so writing it out a second time would create a second, unrelated declaration.

Naming them this way is also how steps connect to each other. This step reads its input by pointing at an attribute on another step’s class, which is what source=CreateTargetGeneStep.target_gene_output above does. A later step can read this step’s result the same way, as IntrinsicFeaturesStep.intrinsic_features_output. Step 4 covers what goes inside these declarations.

Collect them in spec. StepSpec is the step’s whole interface. From it ASOkai derives the command-line options, the configuration keys, the dependencies, the output paths, and the generated reference page. A declaration that is not listed here does not exist as far as the pipeline is concerned.

Supply constructor arguments in analysis_kwargs. It returns a mapping that is passed straight to your analysis constructor, which is where parsed parameters such as args.features are translated into whatever the calculation expects. Omit the method when the constructor takes nothing.

Return the units in analysis_units. Note what it does not do: it does not open a file or parse JSON. args.target_gene is already a TargetGene object, because the input declaration said so. Reading and validating it happens before your code runs.

Nothing in this step writes a file. That is because it is a TableAnalysisStep: the rows returned by the analysis become the CSV without any output code. If you chose the plain AnalysisStep in Decision 2, you implement write_results as well; see Worked example: an output that is not a CSV for one written out in full.

Once you have both classes, this is the order their hooks run in, once per step run:

the step parses and validates every declared input file
analysis_kwargs(args)                -> constructor arguments
analysis_cls(**kwargs)               -> your analysis instance
analysis_units(args)                 -> the units
analysis.run(units, n_workers=...)   -> results
write_results(args, analysis, results)
the step validates every declared output file

Step 4: connect inputs and outputs

This is the part that wires your step into the rest of the pipeline.

Taking a file from another step

Point a FileInput at another step’s FileOutput object:

target_gene_input = FileInput(
    "target_gene",
    source=CreateTargetGeneStep.target_gene_output,
    doc="Serialized target gene object from create-target-gene.",
)

That single reference does four things at once:

it declares a dependency — ASOkai now knows create-target-gene must run first, and will schedule it for you. There is no separate dependency list to maintain;
it inherits the file type, so args.target_gene arrives as a parsed TargetGene rather than a path;
it gives the step a default producer, so a user who selects only your step still gets a working pipeline; and
it supplies the generated documentation with the link between the two steps.

Taking a file from anywhere

When your input is not tied to one producer, declare a file type instead of a source:

reference_input = FileInput(
    "reference",
    file_type=FastaFile(),
    doc="Plain FASTA reference to search.",
)

This is a free input: it has no default producer, so it must be supplied every time, either on the command line or by a task that links it to a producing run.

Either way, a source is a default rather than a restriction: a user can still point any input at an existing file or at another run’s output. See Configuring task runs for how that selection works.

Declaring what you write

A FileOutput names the result, gives it a file type, and says where it belongs under the data directory:

destination=OutputPathTemplate(
    "{assembly}/{source}/{release}/targets/{target}/"
    "analysis/intrinsic/{target}_k{k}_{region}_intrinsic_{features}.csv",
    fields={
        "assembly": TemplateField("parameters.genome.assembly"),
        "release": TemplateField("parameters.genome.release"),
        "target": TemplateField.first_of(
            "parameters.target.target_id",
            "parameters.target.target_name",
        ),
        "k": TemplateField("parameters.target.k"),
        "region": TemplateField("parameters.target.region"),
        "features": TemplateField(
            "parameters.analysis.intrinsic_features.features",
            default="all",
        ),
    },
)

Each {placeholder} is filled from configuration:

Field rule	Reads
`TemplateField("a.b.c")`	One configuration path. Add `default=` when the value may be absent.
`TemplateField.first_of("a", "b")`	The first of several paths that has a value.
`InputConfigField("input_name", "a.b.c")`	A value from the configuration of that particular input’s producer. Use it when two inputs of the same type come from differently configured runs, and the filename must tell them apart.

Put values that change the result into the filename. Two runs with different parameters must not write to the same path. The {features} placeholder above exists for exactly that reason. A value that cannot change the result, such as a worker count, does not belong there.

Choosing file types

Every input and output declares a file type. The available types are CsvFile, JsonFile, FastaFile, SamFile, KffFile, and Bowtie2IndexDirectory, from ASOkai._pipeline.contracts.file_types (or ASOkai.Plugins for a plugin). The type decides how an input is parsed, how an output is validated, and what the documentation says about the format.

Note

If your analysis reads or writes a format that is not listed, raise it with Arash Ayat, or add the type yourself alongside the existing ones. The type is what gives you parsing on the way in and validation on the way out, so it is worth adding rather than working around.

Step 5: parameters and configuration

A ScalarParam declares one configurable value. It becomes a command-line option, a configuration key, and a documented parameter at once:

ScalarParam(
    "features",
    str,
    config="parameters.analysis.intrinsic_features.features",
    default="all",
    doc="Comma-separated intrinsic feature columns; 'all' selects every feature.",
)

For a built-in step, every config path must exist in src/ASOkai/_pipeline/config.schema.yaml, and should have a realistic value in examples/config.example.yaml — the generated page uses that file for its examples.

Where to put a rule. Follow the Validation boundaries:

Kind of rule	Where it belongs
Types, ranges, allowed values, required fields	The YAML schema.
A relationship between two parameters that the schema cannot express clearly	`validate_parameters(values)` on the step.
A rule about the calculation’s own arguments	Your analysis constructor.

The parameter resolver applies the schema at each ScalarParam.config path to shared configuration, run overrides, defaults, and direct ASOkai step arguments. Keep scalar type, choice, and range constraints in the schema. Full configuration validation also checks the configuration structure and relationships between groups.

validate_parameters receives the resolved values and handles relationships that validation of individual fields cannot establish. For example, create-target-gene checks that exactly one of target_id and target_name is supplied, including when invoked directly. A relationship may also appear in the full configuration schema; it should not require duplicating the validation of individual scalar fields in the step.

Use StepSpec.compatible_with only for a real biological restriction, such as an analysis that is meaningful only for one mechanism:

compatible_with={"parameters.antisense.modality.mechanism": ["rnase-h1"]}

Omit it when the analysis applies generally. The modality selector is parameters.antisense.modality.type; each modality owns its mechanism at parameters.antisense.modality.mechanism and its design settings below parameters.antisense.modality.design.

Step 6: register and document

For a built-in step:

Export the analysis class from src/ASOkai/Analysis/__init__.py.
Add the step to _BUILTIN_STEPS in src/ASOkai/_pipeline/catalog.py.
Add an autoclass entry for the analysis to the API page matching its execution model: docs/source/api/modules/Analysis/per-unit.rst or batch.rst.
Create docs/source/steps/<step-name>.rst with a title and one directive:
```
Compute intrinsic features
==========================

.. asokai-step-reference:: compute-intrinsic-features
```
The directive generates the page body from your declarations. Add prose below it only for things the declarations cannot express.

Step 7: write tests

A contribution is expected to bring its own tests. Built-in analyses use two files under tests/unit, mirroring the two classes you wrote:

File	What it covers
`test_<analysis>.py`	The calculation on its own. Construct the analysis directly and check its results for inputs whose answers you know, the edge cases that matter for your science, and the errors your constructor raises for invalid arguments.
`test_<analysis>_step.py`	The step in the pipeline. Check the declared contract, that the output path changes when a parameter that changes the result changes, and at least one end-to-end `run_cli` call that writes a file and reads it back.

The analysis tests are where the science is pinned down, so they matter most. Test the calculation through its own class rather than through the step: it is faster, and a failure points at the calculation instead of the plumbing.

tests/unit/test_intrinsic_features.py and tests/unit/test_intrinsic_features_step.py are a small, complete pair to model yours on.

The shared contract checks in tests/contributor already cover registration, configuration paths, and generated interfaces for every step. Documentation generation is checked in tests/docs and by the Sphinx build, so you do not need to repeat those shared checks in your analysis tests.

Step 8: check your work

While developing, run your own tests together with the shared contract checks:

pytest -q tests/unit/test_<analysis>.py tests/unit/test_<analysis>_step.py
pytest -q tests/contributor/test_analysis_steps.py
pytest -q tests/docs/test_generated_references.py
pre-commit run validate-asokai-contributor-steps --all-files

The contract checks:

pytest -q tests/contributor/test_analysis_steps.py
pre-commit run validate-asokai-contributor-steps --all-files

verify that your step is registered once, that its configuration paths exist in the schema, and that its contract builds from the example configuration. The documentation tests check the generated reference pages.

Before submitting:

pytest
python -m pyright
python -m sphinx -W --keep-going -b html docs/source docs/build/html

Pyright checks method-override compatibility, including overrides of the final execution methods. The check-python-types pre-commit hook runs the same project type check.

The documentation build is run with warnings as errors, so a broken cross-reference fails it.

Finally, run the step itself. A step that passes every contract check can still produce the wrong file:

ASOkai run --step <your-step> --dry-run
ASOkai run --step <your-step>

Worked example: an output that is not a CSV

identify-potential-repeated-sites writes FASTA, so it inherits from AnalysisStep and implements write_results itself. Its analysis returns (identifier, sequence) pairs rather than row dictionaries — with a plain AnalysisStep, the result type is entirely yours, because only your code reads it:

class IdentifyPotentialRepeatedSitesStep(AnalysisStep):
    name = "identify-potential-repeated-sites"
    listed = False
    analysis_cls = SimilarSequenceAnalysis

    potential_sites_output = FileOutput(
        "potential_repeated_sites",
        file_type=FastaFile(),
        destination=OutputPathTemplate(...),
        doc="One FASTA record per potential repeated site.",
    )

    def analysis_units(self, args):
        return args.target_gene.sites

    def write_results(self, args, analysis, results) -> None:
        with args.potential_repeated_sites_output.open("w", encoding="utf-8") as stream:
            SeqIO.write(
                (
                    SeqRecord(Seq(sequence), id=identifier, description="")
                    for identifier, sequence in results
                ),
                stream,
                "fasta",
            )

Two details are worth copying. The output path argument is named after the output: an output called potential_repeated_sites arrives as args.potential_repeated_sites_output, with its parent directory already created. And the step still declares its output with a file type, so the written file is validated afterwards exactly as a CSV would be.

This step is a child of the find-repeated-sites task, which is the public entry in the Analyses catalog. listed = False keeps the child out of the default lists; use asokai describe step identify-potential-repeated-sites to inspect it or asokai list -v steps to include internal steps.

The template

examples/contributor_steps/analysis_step.py

#!/usr/bin/env python
"""
Filename: examples/contributor_steps/analysis_step.py
Author: Arash Ayat
Copyright: 2026, Alexander Schliep
Description: Copyable starting point for an ASOkai analysis and pipeline step.
License: LGPL-3.0-or-later

To add a table analysis step:

1. Copy this file and replace the example analysis, names, and declarations.
2. Add any new configuration fields to
   ``src/ASOkai/_pipeline/config.schema.yaml`` and
   ``examples/config.example.yaml``.
3. Register the step in ``ASOkai._pipeline.catalog``.
4. Run the contributor contract test, or let pre-commit run it.

The declared ``StepSpec`` is the step's schema; no second contributor schema is
needed.

The first declared output is the primary analysis CSV. A table step may declare more
CSV or JSON outputs after it and return their values from
``additional_output_values``.

Use ``StepSpec.compatible_with`` only when the analysis is biologically
restricted, for example
``compatible_with={"parameters.antisense.modality.mechanism": ["rnase-h1"]}``. Omit it for an
analysis that applies regardless of modality, mechanism, or chemistry.

Most steps need no custom validation. Override ``validate_parameters`` only
for relationships between scalar parameters that the YAML schema cannot
express; the method receives values keyed by their ``ScalarParam`` names.
"""

from typing import ClassVar

from ASOkai.Types import AnalysisUnitType, Scalar
from ASOkai.Analysis import PerUnitAnalysis
from ASOkai.Sites import Site
from ASOkai.Targets import TargetGene
from ASOkai._pipeline.contracts.file_types import CsvFile
from ASOkai._pipeline.contracts.files import FileInput, FileOutput
from ASOkai._pipeline.contracts.paths import (
    InputConfigField,
    OutputPathTemplate,
    TemplateField,
)
from ASOkai._pipeline.contracts.step_spec import StepSpec
from ASOkai._pipeline.steps.analysis.table.base import TableAnalysisStep
from ASOkai._pipeline.steps.core.create_target_gene import CreateTargetGeneStep


class ExampleAnalysis(PerUnitAnalysis):
    """Replace this calculation with the new target-context analysis.

    - **target_matches**: Number of equal sequences in the selected target.
    """

    analysis_unit_type: ClassVar[AnalysisUnitType] = "site"

    def __init__(self, target: TargetGene) -> None:
        self.target = target

    def analyze_unit(self, site: Site) -> list[dict[str, Scalar]]:
        # Add per-site analysis code here.
        return [
            {
                "site_id": site.id,
                "target_matches": sum(
                    target_site.sequence == site.sequence
                    for target_site in self.target.sites
                ),
            }
        ]


class ExampleAnalysisStep(TableAnalysisStep):
    """Replace the example names and declarations for the new step."""

    name = "example-analysis"
    description = "Compares every main target site with a selected target."
    analysis_cls = ExampleAnalysis

    sites_input = FileInput(
        "sites",
        source=CreateTargetGeneStep.target_gene_output,
        doc="Main target containing the sites to analyze.",
    )
    target_input = FileInput(
        "target",
        source=CreateTargetGeneStep.target_gene_output,
        doc=(
            "Target used for the target-wide comparison. Link a named "
            "producer run to select a different region from the main sites."
        ),
    )
    analysis_output = FileOutput(
        "analysis",
        file_type=CsvFile(
            index={"site_id": str},
            columns={"target_matches": int},
        ),
        destination=OutputPathTemplate(
            template=(
                "{assembly}/{source}/{release}/targets/{target_name}/"
                "analysis/example/"
                "{target_name}_k{k}_{sites_region}_against_"
                "{target_region}_example.csv"
            ),
            fields={
                "assembly": TemplateField("parameters.genome.assembly"),
                "source": TemplateField(
                    "parameters.genome.source",
                    default="ensembl",
                ),
                "release": TemplateField("parameters.genome.release"),
                "target_name": TemplateField.first_of(
                    "parameters.target.target_id",
                    "parameters.target.target_name",
                ),
                "k": TemplateField("parameters.target.k"),
                "sites_region": InputConfigField(
                    "sites",
                    "parameters.target.region",
                ),
                "target_region": InputConfigField(
                    "target",
                    "parameters.target.region",
                ),
            },
        ),
    )
    spec = StepSpec(
        inputs=[sites_input, target_input],
        outputs=[analysis_output],
    )

    def analysis_kwargs(self, args) -> dict:
        return {"target": args.target}

    def analysis_units(self, args):
        return args.sites.sites

Generated reference documentation

The pipeline reference is generated from the same declarations that ASOkai uses at runtime. Contributors do not maintain a second documentation contract. The generated page reads the following sources:

Step.description

Introduces the step.

StepSpec.compatible_with

Creates the applicability section when the step has a biological restriction.

ScalarParam and config.schema.yaml

Select the configuration values used by the step and provide their types, requirements, constraints, and descriptions. Recognized string patterns are rendered as readable format rules, including run identifiers, named output references, and comma-separated selections of unique values. Other patterns retain their regular expression in the generated reference.

FileInput.doc and FileOutput.doc

Describe file dependencies and produced files. Their file types supply the format and link JSON outputs to their serializable classes.

OutputPathTemplate

Supplies the documented destination template and example output path.

CsvFile and the analysis class docstring

Declare output columns and their types. Describe a column in the analysis class docstring with one named bullet:

class ExampleAnalysis(PerUnitAnalysis):
    """Calculate one value for each target site.

    - **example_value**: Length of the target-site sequence.
    """

The bullet name must match a declared CSV column. Columns without a matching bullet remain valid but have no generated meaning beyond their type and role.

Conditional parameter alternatives

When one configuration value selects between parameter sets, express that relationship in the YAML schema. The reference renderer recognizes two or more sibling allOf branches in which each if selects one string const and each then declares the fields required, allowed, or forbidden for that value. It renders those alternatives as separate parameter subsections with separate configuration examples.

For example, this schema separates simple and bounded modes without adding documentation-only declarations:

mode:
  type: object
  additionalProperties: false
  required:
    - type
  properties:
    type:
      type: string
      description: Calculation mode.
    limit:
      type: integer
      minimum: 1
      description: Maximum value used by bounded mode.
  allOf:
    - if:
        required: [type]
        properties:
          type:
            const: bounded
      then:
        required: [limit]
    - if:
        required: [type]
        properties:
          type:
            const: simple
      then:
        not:
          anyOf:
            - required: [limit]

The selector and conditional fields must also be referenced by the step’s ScalarParam declarations; unused schema fields are not added to that step’s page. Other valid YAML-schema conditions still validate configuration, but the renderer does not turn conditions outside this documented form into separate subsections. An x-error-message on such a condition can provide an ordinary parameter rule instead.

Built-in pages use examples/config.example.yaml for rendered values and example paths, so every built-in configuration field should have a realistic example there.

Publishing a plugin instead

An independently installed plugin does not modify ASOkai’s registry. It publishes the step through the asokai.steps Python entry-point group. When it introduces configuration fields, it also publishes its YAML schema through asokai.config_schemas. ASOkai discovers and validates these entry points from the installed package metadata.

A plugin imports its contracts from ASOkai.Plugins rather than from the internal ASOkai._pipeline modules, and owns its own documentation: plugins are not inserted into ASOkai’s static documentation build.

See examples/plugins/example_analysis for a minimal installable plugin, including its package layout, entry points, and configuration schema.