crispr_analysis_utils.guide_annotation.pipeline¶

crispr_analysis_utils.guide_annotation.pipeline ¶

The gene annotation: read the aligned sites, place them in genes, write the table.

outdir gets three folders and nothing else: annotations/ with the one results table and its guide_gene_annotation_columns.yaml, run_metadata/ with this invocation's record, and logs/.

RESULTS_DIR `module-attribute` ¶

RESULTS_DIR = 'annotations'

Where the step's results go, so nothing of its own sits beside the inputs.

METADATA_DIR `module-attribute` ¶

METADATA_DIR = 'run_metadata'

Where each invocation leaves its metadata, one file per invocation.

STEP `module-attribute` ¶

STEP = 'guide_annotation'

The step those metadata files are named for.

PROMOTER_WINDOW `module-attribute` ¶

PROMOTER_WINDOW = 1000

How far either side of a TSS counts as its promoter, by default.

RunResult `dataclass` ¶

What run produced.

Attributes:

outdir (Path) –

The output folder.
annotated (DataFrame) –

The annotation table: one row per site per overlapping gene.
store (Store) –

The annotation store the run used.
metadata (dict) –

The record written under run_metadata/.

Source code in src/crispr_analysis_utils/guide_annotation/pipeline.py

@dataclass
class RunResult:
    """What `run` produced.

    Attributes
    ----------
    outdir
        The output folder.
    annotated
        The annotation table: one row per site per overlapping gene.
    store
        The annotation store the run used.
    metadata
        The record written under ``run_metadata/``.
    """

    outdir: Path
    annotated: pd.DataFrame
    store: Store
    metadata: dict = field(default_factory=dict)

run ¶

run(alignment: str | PathLike[str] | DataFrame, store: str | PathLike[str] | Store, outdir: str | PathLike[str], *, promoters: str | PathLike[str] | None = None, promoter_window: int = PROMOTER_WINDOW, distance_from: str = MIDPOINT) -> RunResult

Annotate every aligned site against the genes of the store.

Parameters:

alignment (str | PathLike[str] | DataFrame) –

guide_genome_annotation.tsv from guide_alignment.run, or the same table as a DataFrame. Only the sites with a PAM are annotated, which is what it means for a guide to map.
store (str | PathLike[str] | Store) –

The annotation store, as build_store wrote it.
outdir (str | PathLike[str]) –

Where the table goes. Created if it does not exist; nothing is ever written outside it.
promoters (str | PathLike[str] | None, default: None ) –

A TSS BED. Without one, a site per transcript is taken from the store.
promoter_window (int, default: PROMOTER_WINDOW ) –

How far either side of a TSS counts as its promoter.
distance_from (str, default: MIDPOINT ) –

Where distances are measured from: the protospacer midpoint, or the target site 3 bases 5' of the PAM.

Returns:

RunResult –

The table, the store and the record.

Raises:

AnnotationError –

The sites and the store name contigs differently, or the alignment table holds no mapped site.

Source code in src/crispr_analysis_utils/guide_annotation/pipeline.py

def run(
    alignment: str | os.PathLike[str] | pd.DataFrame,
    store: str | os.PathLike[str] | Store,
    outdir: str | os.PathLike[str],
    *,
    promoters: str | os.PathLike[str] | None = None,
    promoter_window: int = PROMOTER_WINDOW,
    distance_from: str = MIDPOINT,
) -> RunResult:
    """Annotate every aligned site against the genes of the store.

    Parameters
    ----------
    alignment
        ``guide_genome_annotation.tsv`` from `guide_alignment.run`, or the
        same table as a DataFrame. Only the sites with a PAM are annotated,
        which is what it means for a guide to map.
    store
        The annotation store, as `build_store` wrote it.
    outdir
        Where the table goes. Created if it does not exist; nothing is ever
        written outside it.
    promoters
        A TSS BED. Without one, a site per transcript is taken from the store.
    promoter_window
        How far either side of a TSS counts as its promoter.
    distance_from
        Where distances are measured from: the protospacer midpoint, or the
        target site 3 bases 5' of the PAM.

    Returns
    -------
    RunResult
        The table, the store and the record.

    Raises
    ------
    AnnotationError
        The sites and the store name contigs differently, or the alignment
        table holds no mapped site.
    """
    outdir = Path(outdir)
    outdir.mkdir(parents=True, exist_ok=True)
    (outdir / "logs").mkdir(exist_ok=True)
    (outdir / METADATA_DIR).mkdir(exist_ok=True)
    (outdir / RESULTS_DIR).mkdir(exist_ok=True)
    record = outdir / METADATA_DIR / f"{STEP}_{_stamp()}.json"

    parameters = {
        "promoters": None if promoters is None else str(promoters),
        "promoter_window": promoter_window,
        "distance_from": distance_from,
    }
    state: dict = {
        "step": STEP,
        "complete": False,
        "error": None,
        "started": _now(),
        "finished": None,
        "cau_version": __version__,
        "inputs": {},
        "parameters": parameters,
        "store": {},
        "counts": {},
        "seconds": {},
    }
    started = time.perf_counter()

    with _log_to(outdir / "logs" / LOG_FILE):
        logger.info("Gene annotation into %s, with %s", outdir, parameters)
        try:
            return _run(
                alignment, store, outdir, promoters, parameters, state, record, started
            )
        except Exception as error:
            state["error"] = f"{type(error).__name__}: {error}"
            state["finished"] = _now()
            state["seconds"]["total"] = round(time.perf_counter() - started, 3)
            write_json(record, state)
            logger.error("Gene annotation failed: %s", state["error"])
            raise

crispr_analysis_utils.guide_annotation.pipeline¶