src/utilities/rna_velocity/run_kb_python.py
def parse_count(args, run_dir, logger):
    """Parser for the `count` command. Data quantification with kallisto and bustools.
    
    Keyword Arguments:
    args -- Command-line arguments dictionary, as parsed by argparse (type: dict)
    run_dir -- Path on the EC2 instance under which jobs are run (type: pathlib.Path)
    logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
    """
    # Build paths on the EC2 instance to store inputs and outputs of kb data quantification results.
    kb_count_dir = run_dir / "kb_count"
    kb_fastqs = kb_count_dir / "fastqs"
    kb_count_inputs = kb_count_dir / "inputs"
    kb_count_outputs = kb_count_dir / "outputs"
    kb_fastqs.mkdir(parents=True)
    kb_count_inputs.mkdir(parents=True)
    kb_count_outputs.mkdir(parents=True)
    kb_count_paths = dict()
    s3_kb_count = dict()

    for arg in ["--tmp", "-o", "-w", "-i", "-g", "-c1", "-c2"]:
        if arg in sys.argv:
            arg = arg[2:] if arg == "--tmp" else arg[1:]
            print(
                f"testing purpose - see if getattr(args, arg) returns the correct values for `kb count` inputs: {getattr(args, arg)}"
            )  # testing purpose
            if arg == "tmp":
                kb_count_paths[arg + "_path"] = kb_count_dir / "alter_tmp"
                s3_kb_count["s3_" + arg + "_prefix"] = posixpath.join(
                    s3u.s3_bucket_and_key(getattr(args, arg))[1], "alter_tmp"
                )
            elif arg == "o":
                kb_count_paths[arg + "_path"] = kb_count_outputs
                s3_kb_count["s3_" + arg + "_prefix"] = posixpath.join(
                    s3u.s3_bucket_and_key(getattr(args, arg))[1], "outputs"
                )
            else:
                kb_count_paths[arg + "_path"] = kb_count_inputs / os.path.basename(
                    getattr(args, arg)
                )
                (
                    s3_kb_count["s3_" + arg + "_bucket"],
                    s3_kb_count["s3_" + arg + "_prefix"],
                ) = s3u.s3_bucket_and_key(getattr(args, arg))
                s3c.download_file(
                    Bucket=s3_kb_count["s3_" + arg + "_bucket"],
                    Key=s3_kb_count["s3_" + arg + "_prefix"],
                    Filename=str(kb_count_paths[arg + "_path"]),
                )
            s3_kb_count["s3_" + arg + "_bucket"] = s3u.s3_bucket_and_key(
                getattr(args, arg)
            )[0]

    # Download fastq files from the AWS S3 bucket to the EC2 instance.
    (
        kb_count_paths["fastqs_path"],
        s3_kb_count["s3_fastqs_bucket"],
        s3_kb_count["s3_fastqs_prefix"],
    ) = (dict(), dict(), dict())
    kb_count_fastq_files_paths, s3_kb_count_fastqs_bucket, s3_kb_count_fastqs_prefix = (
        kb_count_paths["fastqs_path"],
        s3_kb_count["s3_fastqs_bucket"],
        s3_kb_count["s3_fastqs_prefix"],
    )

    s3_fastq_folder_bucket, s3_fastq_folder_prefix = s3u.s3_bucket_and_key(
        args.fastq_folder
    )
    s3_fastq_files_prefix = list(
        s3u.get_files(bucket=s3_fastq_folder_bucket, prefix=s3_fastq_folder_prefix)
    )[1:]
    print(
        "testing purpose - see if all fastq files prefix are extracted: {s3_fastq_files_prefix}"
    )  # testing purpose
    fastq_format = re.compile("([^/]+)_R\d(?:_\d+)?.fastq.gz$")

    for fastq_prefix in s3_fastq_files_prefix:
        if not fastq_format.search(os.path.basename(fastq_prefix)):
            continue
        kb_count_fastq_files_paths[
            os.path.basename(fastq_prefix)
        ] = kb_fastqs / os.path.basename(fastq_prefix)
        s3_kb_count_fastqs_bucket[
            os.path.basename(fastq_prefix)
        ] = s3_fastq_folder_bucket
        s3_kb_count_fastqs_prefix[os.path.basename(fastq_prefix)] = fastq_prefix
        fastq_t_config = TransferConfig(
            use_threads=False
        )  # testing purpose: comment this line if this runs into error.
        s3c.download_file(
            Bucket=s3_kb_count_fastqs_bucket[os.path.basename(fastq_prefix)],
            Key=s3_kb_count_fastqs_prefix[os.path.basename(fastq_prefix)],
            Filename=str(kb_count_fastq_files_paths[os.path.basename(fastq_prefix)]),
            Config=fastq_t_config,
        )

    # Build the command of running `kb count` to generate count matrices
    count_input_boolean = [
        "--keep-tmp",
        "--verbose",
        "--mm",
        "--tcc",
        "--overwrite",
        "--lamanno",
        "--nucleus",
        "--loom",
        "--h5ad",
    ]
    count_input_file_transfer_required = ["--tmp", "-o", "-w"]
    count_input_kb_indices = ["-i", "-g", "-c1", "-c2"]
    count_input_left_args = ["-t", "-m", "--workflow", "--filter", "-x"]

    kb_count_command = ["kb", "count"]
    for fastq_path in kb_count_fastq_files_paths.values():
        print(
            f"testing purpose - view the paths of individual fastqs on the EC2 instance, i.e. values of dictionary `kb_count_fastq_files_paths`: {kb_count_fastq_files_paths.values()}"
        )  # testing purpose
        kb_count_command += [str(fastq_path)]
    for input in count_input_boolean:
        if input in sys.argv:
            kb_count_command += [input]
    for input in count_input_file_transfer_required:
        if input in sys.argv:
            kb_count_command += [
                input,
                str(kb_count_paths[input[2:] + "_path"])
                if input == "--tmp"
                else str(kb_count_paths[input[1:] + "_path"]),
            ]
    for input in count_input_kb_indices:
        if input in sys.argv:
            kb_count_command += [input, str(kb_count_paths[input[1:] + "_path"])]
    for input in count_input_left_args:
        if input in sys.argv:
            kb_count_command += [
                input,
                getattr(args, input[2:])
                if input == ("--workflow" or "--filter")
                else getattr(args, input[1:]),
            ]
    print(
        f"testing purpose - view kb count command: {kb_count_command}"
    )  # testing purpose

    # Run the command to generate count matrices, and upload the output files on the EC2 instance back to AWS S3
    kb_count_failed = ut_log.log_command(
        logger,
        kb_count_command,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        universal_newlines=True,
    )
    kb_count_t_config = TransferConfig(
        use_threads=False
    )  # testing purpose: comment this line if this runs into error. count matrices are not pretty big and don't necessarily need a transfer config
    if kb_count_failed:
        raise RuntimeError("kb data quantification failed")
    else:
        logger.info(f"Uploading kb quantification data from the EC2 instance to AWS S3")
        for root, dirs, files in os.walk(str(kb_count_outputs)):
            for file in files:
                s3c.upload_file(
                    Filename=os.path.join(root, file),
                    Bucket=s3_kb_count["s3_o_bucket"],
                    Key=posixpath.join(s3_kb_count["s3_o_prefix"], file),
                    Config=kb_count_t_config,
                )
                time.sleep(30)
                print(
                    f"testing purpose - see if kb data quantification outputs are uploaded to the correct s3 paths: "
                    + s3_kb_count["s3_o_bucket"]
                    + ", "
                    + posixpath.join(s3_kb_count["s3_o_prefix"], file)
                )  # testing purpose
    return
Parser for the `count` command. Data quantification with kallisto and bustools.
    
    Keyword Arguments:
    args -- Command-line arguments dictionary, as parsed by argparse (type: dict)
    run_dir -- Path on the EC2 instance under which jobs are run (type: pathlib.Path)
    logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
    
Parser for the `count` command. Data quantification with kallisto and bustools.
Parser for the ` count ` command . Data quantification with kallisto and bustools .
def parse_count(args, run_dir, logger):
    kb_count_dir = run_dir / "kb_count"
    kb_fastqs = kb_count_dir / "fastqs"
    kb_count_inputs = kb_count_dir / "inputs"
    kb_count_outputs = kb_count_dir / "outputs"
    kb_fastqs.mkdir(parents=True)
    kb_count_inputs.mkdir(parents=True)
    kb_count_outputs.mkdir(parents=True)
    kb_count_paths = dict()
    s3_kb_count = dict()
    for arg in ["--tmp", "-o", "-w", "-i", "-g", "-c1", "-c2"]:
        if arg in sys.argv:
            arg = arg[2:] if arg == "--tmp" else arg[1:]
            print(
                f"testing purpose - see if getattr(args, arg) returns the correct values for `kb count` inputs: {getattr(args, arg)}"
            )  
            if arg == "tmp":
                kb_count_paths[arg + "_path"] = kb_count_dir / "alter_tmp"
                s3_kb_count["s3_" + arg + "_prefix"] = posixpath.join(
                    s3u.s3_bucket_and_key(getattr(args, arg))[1], "alter_tmp"
                )
            elif arg == "o":
                kb_count_paths[arg + "_path"] = kb_count_outputs
                s3_kb_count["s3_" + arg + "_prefix"] = posixpath.join(
                    s3u.s3_bucket_and_key(getattr(args, arg))[1], "outputs"
                )
            else:
                kb_count_paths[arg + "_path"] = kb_count_inputs / os.path.basename(
                    getattr(args, arg)
                )
                (
                    s3_kb_count["s3_" + arg + "_bucket"],
                    s3_kb_count["s3_" + arg + "_prefix"],
                ) = s3u.s3_bucket_and_key(getattr(args, arg))
                s3c.download_file(
                    Bucket=s3_kb_count["s3_" + arg + "_bucket"],
                    Key=s3_kb_count["s3_" + arg + "_prefix"],
                    Filename=str(kb_count_paths[arg + "_path"]),
                )
            s3_kb_count["s3_" + arg + "_bucket"] = s3u.s3_bucket_and_key(
                getattr(args, arg)
            )[0]
    (
        kb_count_paths["fastqs_path"],
        s3_kb_count["s3_fastqs_bucket"],
        s3_kb_count["s3_fastqs_prefix"],
    ) = (dict(), dict(), dict())
    kb_count_fastq_files_paths, s3_kb_count_fastqs_bucket, s3_kb_count_fastqs_prefix = (
        kb_count_paths["fastqs_path"],
        s3_kb_count["s3_fastqs_bucket"],
        s3_kb_count["s3_fastqs_prefix"],
    )
    s3_fastq_folder_bucket, s3_fastq_folder_prefix = s3u.s3_bucket_and_key(
        args.fastq_folder
    )
    s3_fastq_files_prefix = list(
        s3u.get_files(bucket=s3_fastq_folder_bucket, prefix=s3_fastq_folder_prefix)
    )[1:]
    print(
        "testing purpose - see if all fastq files prefix are extracted: {s3_fastq_files_prefix}"
    )  
    fastq_format = re.compile("([^/]+)_R\d(?:_\d+)?.fastq.gz$")
    for fastq_prefix in s3_fastq_files_prefix:
        if not fastq_format.search(os.path.basename(fastq_prefix)):
            continue
        kb_count_fastq_files_paths[
            os.path.basename(fastq_prefix)
        ] = kb_fastqs / os.path.basename(fastq_prefix)
        s3_kb_count_fastqs_bucket[
            os.path.basename(fastq_prefix)
        ] = s3_fastq_folder_bucket
        s3_kb_count_fastqs_prefix[os.path.basename(fastq_prefix)] = fastq_prefix
        fastq_t_config = TransferConfig(
            use_threads=False
        )  : comment this line if this runs into error.
        s3c.download_file(
            Bucket=s3_kb_count_fastqs_bucket[os.path.basename(fastq_prefix)],
            Key=s3_kb_count_fastqs_prefix[os.path.basename(fastq_prefix)],
            Filename=str(kb_count_fastq_files_paths[os.path.basename(fastq_prefix)]),
            Config=fastq_t_config,
        )
    count_input_boolean = [
        "--keep-tmp",
        "--verbose",
        "--mm",
        "--tcc",
        "--overwrite",
        "--lamanno",
        "--nucleus",
        "--loom",
        "--h5ad",
    ]
    count_input_file_transfer_required = ["--tmp", "-o", "-w"]
    count_input_kb_indices = ["-i", "-g", "-c1", "-c2"]
    count_input_left_args = ["-t", "-m", "--workflow", "--filter", "-x"]
    kb_count_command = ["kb", "count"]
    for fastq_path in kb_count_fastq_files_paths.values():
        print(
            f"testing purpose - view the paths of individual fastqs on the EC2 instance, i.e. values of dictionary `kb_count_fastq_files_paths`: {kb_count_fastq_files_paths.values()}"
        )  
        kb_count_command += [str(fastq_path)]
    for input in count_input_boolean:
        if input in sys.argv:
            kb_count_command += [input]
    for input in count_input_file_transfer_required:
        if input in sys.argv:
            kb_count_command += [
                input,
                str(kb_count_paths[input[2:] + "_path"])
                if input == "--tmp"
                else str(kb_count_paths[input[1:] + "_path"]),
            ]
    for input in count_input_kb_indices:
        if input in sys.argv:
            kb_count_command += [input, str(kb_count_paths[input[1:] + "_path"])]
    for input in count_input_left_args:
        if input in sys.argv:
            kb_count_command += [
                input,
                getattr(args, input[2:])
                if input == ("--workflow" or "--filter")
                else getattr(args, input[1:]),
            ]
    print(
        f"testing purpose - view kb count command: {kb_count_command}"
    )  
    kb_count_failed = ut_log.log_command(
        logger,
        kb_count_command,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        universal_newlines=True,
    )
    kb_count_t_config = TransferConfig(
        use_threads=False
    )  : comment this line if this runs into error. count matrices are not pretty big and don't necessarily need a transfer config
    if kb_count_failed:
        raise RuntimeError("kb data quantification failed")
    else:
        logger.info(f"Uploading kb quantification data from the EC2 instance to AWS S3")
        for root, dirs, files in os.walk(str(kb_count_outputs)):
            for file in files:
                s3c.upload_file(
                    Filename=os.path.join(root, file),
                    Bucket=s3_kb_count["s3_o_bucket"],
                    Key=posixpath.join(s3_kb_count["s3_o_prefix"], file),
                    Config=kb_count_t_config,
                )
                time.sleep(30)
                print(
                    f"testing purpose - see if kb data quantification outputs are uploaded to the correct s3 paths: "
                    + s3_kb_count["s3_o_bucket"]
                    + ", "
                    + posixpath.join(s3_kb_count["s3_o_prefix"], file)
                )  
    return
def parse_count ( args , run_dir , logger ) : kb_count_dir = run_dir / "kb_count" kb_fastqs = kb_count_dir / "fastqs" kb_count_inputs = kb_count_dir / "inputs" kb_count_outputs = kb_count_dir / "outputs" kb_fastqs . mkdir ( parents = True ) kb_count_inputs . mkdir ( parents = True ) kb_count_outputs . mkdir ( parents = True ) kb_count_paths = dict ( ) s3_kb_count = dict ( ) for arg in [ "--tmp" , "-o" , "-w" , "-i" , "-g" , "-c1" , "-c2" ] : if arg in sys . argv : arg = arg [ 2 : ] if arg == "--tmp" else arg [ 1 : ] print ( f"testing purpose - see if getattr(args, arg) returns the correct values for `kb count` inputs: {getattr(args, arg)}" ) if arg == "tmp" : kb_count_paths [ arg + "_path" ] = kb_count_dir / "alter_tmp" s3_kb_count [ "s3_" + arg + "_prefix" ] = posixpath . join ( s3u . s3_bucket_and_key ( getattr ( args , arg ) ) [ 1 ] , "alter_tmp" ) elif arg == "o" : kb_count_paths [ arg + "_path" ] = kb_count_outputs s3_kb_count [ "s3_" + arg + "_prefix" ] = posixpath . join ( s3u . s3_bucket_and_key ( getattr ( args , arg ) ) [ 1 ] , "outputs" ) else : kb_count_paths [ arg + "_path" ] = kb_count_inputs / os . path . basename ( getattr ( args , arg ) ) ( s3_kb_count [ "s3_" + arg + "_bucket" ] , s3_kb_count [ "s3_" + arg + "_prefix" ] , ) = s3u . s3_bucket_and_key ( getattr ( args , arg ) ) s3c . download_file ( Bucket = s3_kb_count [ "s3_" + arg + "_bucket" ] , Key = s3_kb_count [ "s3_" + arg + "_prefix" ] , Filename = str ( kb_count_paths [ arg + "_path" ] ) , ) s3_kb_count [ "s3_" + arg + "_bucket" ] = s3u . s3_bucket_and_key ( getattr ( args , arg ) ) [ 0 ] ( kb_count_paths [ "fastqs_path" ] , s3_kb_count [ "s3_fastqs_bucket" ] , s3_kb_count [ "s3_fastqs_prefix" ] , ) = ( dict ( ) , dict ( ) , dict ( ) ) kb_count_fastq_files_paths , s3_kb_count_fastqs_bucket , s3_kb_count_fastqs_prefix = ( kb_count_paths [ "fastqs_path" ] , s3_kb_count [ "s3_fastqs_bucket" ] , s3_kb_count [ "s3_fastqs_prefix" ] , ) s3_fastq_folder_bucket , s3_fastq_folder_prefix = s3u . s3_bucket_and_key ( args . fastq_folder ) s3_fastq_files_prefix = list ( s3u . get_files ( bucket = s3_fastq_folder_bucket , prefix = s3_fastq_folder_prefix ) ) [ 1 : ] print ( "testing purpose - see if all fastq files prefix are extracted: {s3_fastq_files_prefix}" ) fastq_format = re . compile ( "([^/]+)_R\d(?:_\d+)?.fastq.gz$" ) for fastq_prefix in s3_fastq_files_prefix : if not fastq_format . search ( os . path . basename ( fastq_prefix ) ) : continue kb_count_fastq_files_paths [ os . path . basename ( fastq_prefix ) ] = kb_fastqs / os . path . basename ( fastq_prefix ) s3_kb_count_fastqs_bucket [ os . path . basename ( fastq_prefix ) ] = s3_fastq_folder_bucket s3_kb_count_fastqs_prefix [ os . path . basename ( fastq_prefix ) ] = fastq_prefix fastq_t_config = TransferConfig ( use_threads = False ) s3c . download_file ( Bucket = s3_kb_count_fastqs_bucket [ os . path . basename ( fastq_prefix ) ] , Key = s3_kb_count_fastqs_prefix [ os . path . basename ( fastq_prefix ) ] , Filename = str ( kb_count_fastq_files_paths [ os . path . basename ( fastq_prefix ) ] ) , Config = fastq_t_config , ) count_input_boolean = [ "--keep-tmp" , "--verbose" , "--mm" , "--tcc" , "--overwrite" , "--lamanno" , "--nucleus" , "--loom" , "--h5ad" , ] count_input_file_transfer_required = [ "--tmp" , "-o" , "-w" ] count_input_kb_indices = [ "-i" , "-g" , "-c1" , "-c2" ] count_input_left_args = [ "-t" , "-m" , "--workflow" , "--filter" , "-x" ] kb_count_command = [ "kb" , "count" ] for fastq_path in kb_count_fastq_files_paths . values ( ) : print ( f"testing purpose - view the paths of individual fastqs on the EC2 instance, i.e. values of dictionary `kb_count_fastq_files_paths`: {kb_count_fastq_files_paths.values()}" ) kb_count_command += [ str ( fastq_path ) ] for input in count_input_boolean : if input in sys . argv : kb_count_command += [ input ] for input in count_input_file_transfer_required : if input in sys . argv : kb_count_command += [ input , str ( kb_count_paths [ input [ 2 : ] + "_path" ] ) if input == "--tmp" else str ( kb_count_paths [ input [ 1 : ] + "_path" ] ) , ] for input in count_input_kb_indices : if input in sys . argv : kb_count_command += [ input , str ( kb_count_paths [ input [ 1 : ] + "_path" ] ) ] for input in count_input_left_args : if input in sys . argv : kb_count_command += [ input , getattr ( args , input [ 2 : ] ) if input == ( "--workflow" or "--filter" ) else getattr ( args , input [ 1 : ] ) , ] print ( f"testing purpose - view kb count command: {kb_count_command}" ) kb_count_failed = ut_log . log_command ( logger , kb_count_command , shell = True , stdout = subprocess . PIPE , stderr = subprocess . STDOUT , universal_newlines = True , ) kb_count_t_config = TransferConfig ( use_threads = False ) if kb_count_failed : raise RuntimeError ( "kb data quantification failed" ) else : logger . info ( f"Uploading kb quantification data from the EC2 instance to AWS S3" ) for root , dirs , files in os . walk ( str ( kb_count_outputs ) ) : for file in files : s3c . upload_file ( Filename = os . path . join ( root , file ) , Bucket = s3_kb_count [ "s3_o_bucket" ] , Key = posixpath . join ( s3_kb_count [ "s3_o_prefix" ] , file ) , Config = kb_count_t_config , ) time . sleep ( 30 ) print ( f"testing purpose - see if kb data quantification outputs are uploaded to the correct s3 paths: " + s3_kb_count [ "s3_o_bucket" ] + ", " + posixpath . join ( s3_kb_count [ "s3_o_prefix" ] , file ) ) return
Parser for the `count` command.
Parser for the ` count ` command .
"""Parser for the `count` command. Data quantification with kallisto and bustools.
    
    Keyword Arguments:
    args -- Command-line arguments dictionary, as parsed by argparse (type: dict)
    run_dir -- Path on the EC2 instance under which jobs are run (type: pathlib.Path)
    logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
    """ # Build paths on the EC2 instance to store inputs and outputs of kb data quantification results. # testing purpose # Download fastq files from the AWS S3 bucket to the EC2 instance. # testing purpose # testing purpose: comment this line if this runs into error. # Build the command of running `kb count` to generate count matrices # testing purpose # testing purpose # Run the command to generate count matrices, and upload the output files on the EC2 instance back to AWS S3 # testing purpose: comment this line if this runs into error. count matrices are not pretty big and don't necessarily need a transfer config # testing purpose
args
run_dir
logger