src/utilities/rna_velocity/run_loompy.py
def main(logger):
    """ Download reference genome from the S3 bucket to an EC2 instance, run alignment jobs with kallisto, then generate loom files based on the alignment results, and upload the loom files to S3 bucket.

        logger - Logger object that exposes the interface the code directly uses
    """

    parser = get_parser()

    args = parser.parse_args()

    root_dir = pathlib.Path(args.root_dir)

    if os.environ.get("AWS_BATCH_JOB_ID"):
        root_dir = root_dir / os.environ["AWS_BATCH_JOB_ID"]

    # local directories on the EC2 instance
    if args.s3_input_path.endswith("/"):
        args.s3_input_path = args.s3_input_path[:-1]

    run_dir = root_dir / "data"
    run_dir.mkdir(parents=True)

    # extract sample name(s) and technology from the metadata tsv file
    metadata_name = os.path.basename(args.metadata)
    metadata_dir = run_dir / "metadata"
    metadata_dir.mkdir(parents=True)
    metadata_dir = metadata_dir / metadata_name
    s3_metadata_bucket, s3_metadata_prefix = s3u.s3_bucket_and_key(args.metadata)

    s3c.download_file(
        Bucket=s3_metadata_bucket,  # just always download this from us-west-2...
        Key=s3_metadata_prefix,
        Filename=str(metadata_dir),
    )

    technology, sample_name = "", ""
    with open(metadata_dir) as fd:
        rd = csv.reader(fd, delimiter="\t", quotechar='"')
        file_content = list()
        for row in rd:
            file_content.append(row)
        file_content = file_content[1:]
        sample_name = file_content[0][0]
        technology = file_content[0][
            1
        ]  # need to fix this later to fit tsv file with multiple samples

    # check if the input genome is valid
    if args.taxon in reference_genomes_indexes:
        genome_name = reference_genomes_indexes[args.taxon]
    else:
        raise ValueError(f"unknown taxon {args.taxon}")

    if "10x" in technology:
        genome_dir = root_dir / "genome" / "10X" / genome_name
    elif (
        "smartseq2" in technology
    ):  # may need to update these after confirming what technology name looks like for smartseq2 data
        genome_dir = (
            root_dir / "genome" / "smartseq2" / genome_name
        )  # necessary to separate the reference genome location path for 10x and smartseq2?
    genome_dir.mkdir(parents=True)

    s3_input_bucket, s3_input_prefix = s3u.s3_bucket_and_key(args.s3_input_path)

    logger.info(
        f"""Run Info: partition {args.partition_id} out of {args.num_partitions}
                   genome_dir:\t{genome_dir}
                        taxon:\t{args.taxon}
                s3_input_path:\t{args.s3_input_path}"""
    )

    s3 = boto3.resource("s3")

    # download the reference genome index data from s3 bucket to an EC2 instance
    logger.info("Downloading reference genome index files of {}".format(genome_name))

    if "10x" in technology:
        s3_genome_index = f"s3://{S3_REFERENCE['west']}/loompy/10X/{genome_name}"
    elif "smartseq2" in technology:
        s3_genome_index = f"s3://{S3_REFERENCE['west']}/loompy/smartseq2/{genome_name}"

    s3_genome_index_bucket, s3_genome_index_prefix = s3u.s3_bucket_and_key(
        s3_genome_index
    )
    s3_genome_files_prefix = list(
        s3u.get_files(s3_genome_index_bucket, s3_genome_index_prefix)
    )
    s3_genome_files_prefix = s3_genome_files_prefix[1:]
    file_names = list(
        os.path.basename(file_path) for file_path in s3_genome_files_prefix
    )
    genome_name_to_prefix = dict(zip(file_names, s3_genome_files_prefix))

    for file in genome_name_to_prefix.keys():
        s3c.download_file(
            Bucket=s3_genome_index_bucket,
            Key=genome_name_to_prefix[file],
            Filename=str(genome_dir / file),
        )

    # extract valid fastq files
    sample_re_smartseq2 = re.compile("([^/]+)_R\d(?:_\d+)?.fastq.gz$")
    sample_re_10x = re.compile("([^/]+)_L\d+_R\d(?:_\d+)?.fastq.gz$")
    s3_output_bucket, s3_output_prefix = s3u.s3_bucket_and_key(args.s3_output_path)

    logger.info(
        "Running partition {} of {}".format(args.partition_id, args.num_partitions)
    )

    # fastq files are either stored directly under the s3 input folder, or in sample sub-folders under the s3 input folder
    if list(s3u.get_files(s3_input_bucket, s3_input_prefix)):
        fastq_key_and_size = [
            (fn, s)
            for fn, s in s3u.get_size(s3_input_bucket, s3_input_prefix)
            if fn.endswith("fastq.gz")
        ]
    else:
        sample_folder_paths = s3u.get_folders(s3_input_bucket, s3_input_prefix + "/")
        fastq_key_and_size = []
        for sample in sample_folder_paths:
            files = [
                (fn, s)
                for fn, s in s3u.get_size(s3_input_bucket, s3_input_prefix)
                if fn.endswith("fastq.gz")
            ]
            fastq_key_and_size += files

    sample_name_to_fastq_keys = defaultdict(list)
    fastq_sizes = defaultdict(list)
    fastqs_key_to_name = dict()

    for fn, s in fastq_key_and_size:
        matched = False
        if "10x" in technology:
            matched = sample_re_10x.search(os.path.basename(fn))
        elif "smartseq2" in technology:
            matched = sample_re_smartseq2.search(os.path.basename(fn))
        if matched:
            sample_name_to_fastq_keys[matched.group(1)].append(fn)
            fastq_sizes[matched.group(1)].append(s)
            fastqs_key_to_name[fn] = os.path.basename(fn)

    logger.info(f"number of samples: {len(sample_name_to_fastq_keys)}")

    # download input fastqs from S3 to an EC2 instance
    fastq_dir = run_dir / "fastqs"
    fastq_dir.mkdir(parents=True)

    for key in fastqs_key_to_name.keys():
        s3c.download_file(
            Bucket=s3_input_bucket,
            Key=key,
            Filename=str(fastq_dir / fastqs_key_to_name[key]),
        )

    # run kallisto alignment and RNA velocity analysis on the valid fastq files
    for sample in sorted(sample_name_to_fastq_keys)[
        args.partition_id :: args.num_partitions
    ]:
        result_path = run_dir / "results"
        result_path.mkdir(parents=True)

        command = [
            "loompy",
            "fromfq",
            str(result_path / f"{sample_name}.loom"),
            sample_name,
            str(genome_dir),
            str(metadata_dir),
        ]
        fastq_names = [
            fastqs_key_to_name[key] for key in sample_name_to_fastq_keys[sample]
        ]
        fastq_dirs = [str(fastq_dir / fastq) for fastq in fastq_names]
        command += fastq_dirs
        print(command)  # for testing purpose

        failed = ut_log.log_command(
            logger,
            command,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            universal_newlines=True,
        )

        t_config = TransferConfig(use_threads=False)
        if failed:
            raise RuntimeError("loompy failed")
        else:
            logger.info(f"Uploading {sample_name}.loom")
            s3c.upload_file(
                Filename=str(result_path / f"{sample_name}.loom"),
                Bucket=s3_output_bucket,
                Key=os.path.join(s3_output_prefix, f"{sample_name}.loom"),
                Config=t_config,
            )

        command = ["rm", "-rf", str(result_path)]
        ut_log.log_command(logger, command, shell=True)

        time.sleep(30)

    logger.info("Job completed")
 Download reference genome from the S3 bucket to an EC2 instance, run alignment jobs with kallisto, then generate loom files based on the alignment results, and upload the loom files to S3 bucket.

        logger - Logger object that exposes the interface the code directly uses
    
Download reference genome from the S3 bucket to an EC2 instance, run alignment jobs with kallisto, then generate loom files based on the alignment results, and upload the loom files to S3 bucket.
logger - Logger object that exposes the interface the code directly uses
Download reference genome from the S3 bucket to an EC2 instance run alignment jobs with kallisto then generate loom files based on the alignment results and upload the loom files to S3 bucket . logger - Logger object that exposes the interface the code directly uses
def main(logger):
    parser = get_parser()
    args = parser.parse_args()
    root_dir = pathlib.Path(args.root_dir)
    if os.environ.get("AWS_BATCH_JOB_ID"):
        root_dir = root_dir / os.environ["AWS_BATCH_JOB_ID"]
    if args.s3_input_path.endswith("/"):
        args.s3_input_path = args.s3_input_path[:-1]
    run_dir = root_dir / "data"
    run_dir.mkdir(parents=True)
    metadata_name = os.path.basename(args.metadata)
    metadata_dir = run_dir / "metadata"
    metadata_dir.mkdir(parents=True)
    metadata_dir = metadata_dir / metadata_name
    s3_metadata_bucket, s3_metadata_prefix = s3u.s3_bucket_and_key(args.metadata)
    s3c.download_file(
        Bucket=s3_metadata_bucket,  
        Key=s3_metadata_prefix,
        Filename=str(metadata_dir),
    )
    technology, sample_name = "", ""
    with open(metadata_dir) as fd:
        rd = csv.reader(fd, delimiter="\t", quotechar='"')
        file_content = list()
        for row in rd:
            file_content.append(row)
        file_content = file_content[1:]
        sample_name = file_content[0][0]
        technology = file_content[0][
            1
        ]  
    if args.taxon in reference_genomes_indexes:
        genome_name = reference_genomes_indexes[args.taxon]
    else:
        raise ValueError(f"unknown taxon {args.taxon}")
    if "10x" in technology:
        genome_dir = root_dir / "genome" / "10X" / genome_name
    elif (
        "smartseq2" in technology
    ):  
        genome_dir = (
            root_dir / "genome" / "smartseq2" / genome_name
        )  
    genome_dir.mkdir(parents=True)
    s3_input_bucket, s3_input_prefix = s3u.s3_bucket_and_key(args.s3_input_path)
    logger.info(
        f"""Run Info: partition {args.partition_id} out of {args.num_partitions}
                   genome_dir:\t{genome_dir}
                        taxon:\t{args.taxon}
                s3_input_path:\t{args.s3_input_path}"""
    )
    s3 = boto3.resource("s3")
    logger.info("Downloading reference genome index files of {}".format(genome_name))
    if "10x" in technology:
        s3_genome_index = f"s3://{S3_REFERENCE['west']}/loompy/10X/{genome_name}"
    elif "smartseq2" in technology:
        s3_genome_index = f"s3://{S3_REFERENCE['west']}/loompy/smartseq2/{genome_name}"
    s3_genome_index_bucket, s3_genome_index_prefix = s3u.s3_bucket_and_key(
        s3_genome_index
    )
    s3_genome_files_prefix = list(
        s3u.get_files(s3_genome_index_bucket, s3_genome_index_prefix)
    )
    s3_genome_files_prefix = s3_genome_files_prefix[1:]
    file_names = list(
        os.path.basename(file_path) for file_path in s3_genome_files_prefix
    )
    genome_name_to_prefix = dict(zip(file_names, s3_genome_files_prefix))
    for file in genome_name_to_prefix.keys():
        s3c.download_file(
            Bucket=s3_genome_index_bucket,
            Key=genome_name_to_prefix[file],
            Filename=str(genome_dir / file),
        )
    sample_re_smartseq2 = re.compile("([^/]+)_R\d(?:_\d+)?.fastq.gz$")
    sample_re_10x = re.compile("([^/]+)_L\d+_R\d(?:_\d+)?.fastq.gz$")
    s3_output_bucket, s3_output_prefix = s3u.s3_bucket_and_key(args.s3_output_path)
    logger.info(
        "Running partition {} of {}".format(args.partition_id, args.num_partitions)
    )
    if list(s3u.get_files(s3_input_bucket, s3_input_prefix)):
        fastq_key_and_size = [
            (fn, s)
            for fn, s in s3u.get_size(s3_input_bucket, s3_input_prefix)
            if fn.endswith("fastq.gz")
        ]
    else:
        sample_folder_paths = s3u.get_folders(s3_input_bucket, s3_input_prefix + "/")
        fastq_key_and_size = []
        for sample in sample_folder_paths:
            files = [
                (fn, s)
                for fn, s in s3u.get_size(s3_input_bucket, s3_input_prefix)
                if fn.endswith("fastq.gz")
            ]
            fastq_key_and_size += files
    sample_name_to_fastq_keys = defaultdict(list)
    fastq_sizes = defaultdict(list)
    fastqs_key_to_name = dict()
    for fn, s in fastq_key_and_size:
        matched = False
        if "10x" in technology:
            matched = sample_re_10x.search(os.path.basename(fn))
        elif "smartseq2" in technology:
            matched = sample_re_smartseq2.search(os.path.basename(fn))
        if matched:
            sample_name_to_fastq_keys[matched.group(1)].append(fn)
            fastq_sizes[matched.group(1)].append(s)
            fastqs_key_to_name[fn] = os.path.basename(fn)
    logger.info(f"number of samples: {len(sample_name_to_fastq_keys)}")
    fastq_dir = run_dir / "fastqs"
    fastq_dir.mkdir(parents=True)
    for key in fastqs_key_to_name.keys():
        s3c.download_file(
            Bucket=s3_input_bucket,
            Key=key,
            Filename=str(fastq_dir / fastqs_key_to_name[key]),
        )
    for sample in sorted(sample_name_to_fastq_keys)[
        args.partition_id :: args.num_partitions
    ]:
        result_path = run_dir / "results"
        result_path.mkdir(parents=True)
        command = [
            "loompy",
            "fromfq",
            str(result_path / f"{sample_name}.loom"),
            sample_name,
            str(genome_dir),
            str(metadata_dir),
        ]
        fastq_names = [
            fastqs_key_to_name[key] for key in sample_name_to_fastq_keys[sample]
        ]
        fastq_dirs = [str(fastq_dir / fastq) for fastq in fastq_names]
        command += fastq_dirs
        print(command)  
        failed = ut_log.log_command(
            logger,
            command,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            universal_newlines=True,
        )
        t_config = TransferConfig(use_threads=False)
        if failed:
            raise RuntimeError("loompy failed")
        else:
            logger.info(f"Uploading {sample_name}.loom")
            s3c.upload_file(
                Filename=str(result_path / f"{sample_name}.loom"),
                Bucket=s3_output_bucket,
                Key=os.path.join(s3_output_prefix, f"{sample_name}.loom"),
                Config=t_config,
            )
        command = ["rm", "-rf", str(result_path)]
        ut_log.log_command(logger, command, shell=True)
        time.sleep(30)
    logger.info("Job completed")
def main ( logger ) : parser = get_parser ( ) args = parser . parse_args ( ) root_dir = pathlib . Path ( args . root_dir ) if os . environ . get ( "AWS_BATCH_JOB_ID" ) : root_dir = root_dir / os . environ [ "AWS_BATCH_JOB_ID" ] if args . s3_input_path . endswith ( "/" ) : args . s3_input_path = args . s3_input_path [ : - 1 ] run_dir = root_dir / "data" run_dir . mkdir ( parents = True ) metadata_name = os . path . basename ( args . metadata ) metadata_dir = run_dir / "metadata" metadata_dir . mkdir ( parents = True ) metadata_dir = metadata_dir / metadata_name s3_metadata_bucket , s3_metadata_prefix = s3u . s3_bucket_and_key ( args . metadata ) s3c . download_file ( Bucket = s3_metadata_bucket , Key = s3_metadata_prefix , Filename = str ( metadata_dir ) , ) technology , sample_name = "" , "" with open ( metadata_dir ) as fd : rd = csv . reader ( fd , delimiter = "\t" , quotechar = '"' ) file_content = list ( ) for row in rd : file_content . append ( row ) file_content = file_content [ 1 : ] sample_name = file_content [ 0 ] [ 0 ] technology = file_content [ 0 ] [ 1 ] if args . taxon in reference_genomes_indexes : genome_name = reference_genomes_indexes [ args . taxon ] else : raise ValueError ( f"unknown taxon {args.taxon}" ) if "10x" in technology : genome_dir = root_dir / "genome" / "10X" / genome_name elif ( "smartseq2" in technology ) : genome_dir = ( root_dir / "genome" / "smartseq2" / genome_name ) genome_dir . mkdir ( parents = True ) s3_input_bucket , s3_input_prefix = s3u . s3_bucket_and_key ( args . s3_input_path ) logger . info ( f"""Run Info: partition {args.partition_id} out of {args.num_partitions}
                   genome_dir:\t{genome_dir}
                        taxon:\t{args.taxon}
                s3_input_path:\t{args.s3_input_path}""" ) s3 = boto3 . resource ( "s3" ) logger . info ( "Downloading reference genome index files of {}" . format ( genome_name ) ) if "10x" in technology : s3_genome_index = f"s3://{S3_REFERENCE['west']}/loompy/10X/{genome_name}" elif "smartseq2" in technology : s3_genome_index = f"s3://{S3_REFERENCE['west']}/loompy/smartseq2/{genome_name}" s3_genome_index_bucket , s3_genome_index_prefix = s3u . s3_bucket_and_key ( s3_genome_index ) s3_genome_files_prefix = list ( s3u . get_files ( s3_genome_index_bucket , s3_genome_index_prefix ) ) s3_genome_files_prefix = s3_genome_files_prefix [ 1 : ] file_names = list ( os . path . basename ( file_path ) for file_path in s3_genome_files_prefix ) genome_name_to_prefix = dict ( zip ( file_names , s3_genome_files_prefix ) ) for file in genome_name_to_prefix . keys ( ) : s3c . download_file ( Bucket = s3_genome_index_bucket , Key = genome_name_to_prefix [ file ] , Filename = str ( genome_dir / file ) , ) sample_re_smartseq2 = re . compile ( "([^/]+)_R\d(?:_\d+)?.fastq.gz$" ) sample_re_10x = re . compile ( "([^/]+)_L\d+_R\d(?:_\d+)?.fastq.gz$" ) s3_output_bucket , s3_output_prefix = s3u . s3_bucket_and_key ( args . s3_output_path ) logger . info ( "Running partition {} of {}" . format ( args . partition_id , args . num_partitions ) ) if list ( s3u . get_files ( s3_input_bucket , s3_input_prefix ) ) : fastq_key_and_size = [ ( fn , s ) for fn , s in s3u . get_size ( s3_input_bucket , s3_input_prefix ) if fn . endswith ( "fastq.gz" ) ] else : sample_folder_paths = s3u . get_folders ( s3_input_bucket , s3_input_prefix + "/" ) fastq_key_and_size = [ ] for sample in sample_folder_paths : files = [ ( fn , s ) for fn , s in s3u . get_size ( s3_input_bucket , s3_input_prefix ) if fn . endswith ( "fastq.gz" ) ] fastq_key_and_size += files sample_name_to_fastq_keys = defaultdict ( list ) fastq_sizes = defaultdict ( list ) fastqs_key_to_name = dict ( ) for fn , s in fastq_key_and_size : matched = False if "10x" in technology : matched = sample_re_10x . search ( os . path . basename ( fn ) ) elif "smartseq2" in technology : matched = sample_re_smartseq2 . search ( os . path . basename ( fn ) ) if matched : sample_name_to_fastq_keys [ matched . group ( 1 ) ] . append ( fn ) fastq_sizes [ matched . group ( 1 ) ] . append ( s ) fastqs_key_to_name [ fn ] = os . path . basename ( fn ) logger . info ( f"number of samples: {len(sample_name_to_fastq_keys)}" ) fastq_dir = run_dir / "fastqs" fastq_dir . mkdir ( parents = True ) for key in fastqs_key_to_name . keys ( ) : s3c . download_file ( Bucket = s3_input_bucket , Key = key , Filename = str ( fastq_dir / fastqs_key_to_name [ key ] ) , ) for sample in sorted ( sample_name_to_fastq_keys ) [ args . partition_id : : args . num_partitions ] : result_path = run_dir / "results" result_path . mkdir ( parents = True ) command = [ "loompy" , "fromfq" , str ( result_path / f"{sample_name}.loom" ) , sample_name , str ( genome_dir ) , str ( metadata_dir ) , ] fastq_names = [ fastqs_key_to_name [ key ] for key in sample_name_to_fastq_keys [ sample ] ] fastq_dirs = [ str ( fastq_dir / fastq ) for fastq in fastq_names ] command += fastq_dirs print ( command ) failed = ut_log . log_command ( logger , command , shell = True , stdout = subprocess . PIPE , stderr = subprocess . STDOUT , universal_newlines = True , ) t_config = TransferConfig ( use_threads = False ) if failed : raise RuntimeError ( "loompy failed" ) else : logger . info ( f"Uploading {sample_name}.loom" ) s3c . upload_file ( Filename = str ( result_path / f"{sample_name}.loom" ) , Bucket = s3_output_bucket , Key = os . path . join ( s3_output_prefix , f"{sample_name}.loom" ) , Config = t_config , ) command = [ "rm" , "-rf" , str ( result_path ) ] ut_log . log_command ( logger , command , shell = True ) time . sleep ( 30 ) logger . info ( "Job completed" )
Download reference genome from the S3 bucket to an EC2 instance, run alignment jobs with kallisto, then generate loom files based on the alignment results, and upload the loom files to S3 bucket.
Download reference genome from the S3 bucket to an EC2 instance run alignment jobs with kallisto then generate loom files based on the alignment results and upload the loom files to S3 bucket .
""" Download reference genome from the S3 bucket to an EC2 instance, run alignment jobs with kallisto, then generate loom files based on the alignment results, and upload the loom files to S3 bucket.

        logger - Logger object that exposes the interface the code directly uses
    """ # local directories on the EC2 instance # extract sample name(s) and technology from the metadata tsv file # just always download this from us-west-2... # need to fix this later to fit tsv file with multiple samples # check if the input genome is valid # may need to update these after confirming what technology name looks like for smartseq2 data # necessary to separate the reference genome location path for 10x and smartseq2? # download the reference genome index data from s3 bucket to an EC2 instance # extract valid fastq files # fastq files are either stored directly under the s3 input folder, or in sample sub-folders under the s3 input folder # download input fastqs from S3 to an EC2 instance # run kallisto alignment and RNA velocity analysis on the valid fastq files # for testing purpose
logger