Skip to content

crispr_analysis_utils.gem_mapper.map_guides_with_gem

crispr_analysis_utils.gem_mapper.map_guides_with_gem

map_guides_with_gem(gem_index: str | Path, input_fastq: str | Path, output_sam: str | Path, *, threads: int = 8, mapping_mode: str = 'sensitive', sam_compact: bool = False, gem_mapper_bin: str = 'gem-mapper', log_path: str | Path | None = None) -> str

Run GEM mapper for single-end FASTQ input.

If log_path is not provided, command output is redirected to a .log file next to output_sam.

Source code in src/crispr_analysis_utils/gem_mapper.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def map_guides_with_gem(
    gem_index: str | Path,
    input_fastq: str | Path,
    output_sam: str | Path,
    *,
    threads: int = 8,
    mapping_mode: str = "sensitive",
    sam_compact: bool = False,
    gem_mapper_bin: str = "gem-mapper",
    log_path: str | Path | None = None,
) -> str:
    """Run GEM mapper for single-end FASTQ input.

    If `log_path` is not provided, command output is redirected to a `.log` file
    next to `output_sam`.
    """
    output_sam_path = Path(output_sam)
    if log_path is None:
        log_path = output_sam_path.with_suffix(".log")

    cmd = [
        gem_mapper_bin,
        "-I",
        str(gem_index),
        "-i",
        str(input_fastq),
        "-o",
        str(output_sam_path),
        f"--sam-compact={'true' if sam_compact else 'false'}",
        "--mapping-mode",
        mapping_mode,
        "--threads",
        str(threads),
    ]
    cmd_str = " ".join(shlex.quote(token) for token in cmd)
    cmd_str = f"{cmd_str} > {shlex.quote(str(log_path))} 2>&1"
    return run_shell_cmd(cmd_str)