klassify 0.1.6

Classify chimeric reads based on unique kmer contents
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import shutil
import subprocess as sp

from pathlib import Path

from jcvi.apps.base import logger


def check_tool(name: str):
    """
    Ensure a tool is in the PATH.
    """
    if shutil.which(name) is None:
        raise RuntimeError(f"Required tool '{name}' not found in PATH")


def run(cmd, *, cwd=None, stdout=None, stdin=None):
    """
    Run a command.
    """
    logger.info("RUN: %s", " ".join(map(str, cmd)))
    return sp.run(cmd, cwd=cwd, stdin=stdin, stdout=stdout, check=True)


def run_pipe(cmd1, cmd2, *, cwd=None, stdout_path: Path | None = None):
    """
    Run a pipeline of commands.
    """
    logger.info("PIPE: %s | %s", " ".join(map(str, cmd1)), " ".join(map(str, cmd2)))
    p1 = sp.Popen(cmd1, cwd=cwd, stdout=sp.PIPE)
    with open(stdout_path, "wb") as fh:
        p2 = sp.Popen(cmd2, cwd=cwd, stdin=p1.stdout, stdout=fh)
        p1.stdout.close()
        rc1 = p1.wait()
        rc2 = p2.wait()
        if rc1 != 0 or rc2 != 0:
            raise RuntimeError(f"Pipeline failed (rc1={rc1}, rc2={rc2})")