src/utilities/rna_velocity/run_velocyto_star.py
def main(logger):
    """ Download reference genome, run velocyto jobs, and upload results to S3.

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

    parser = get_parser()

    args = parser.parse_args()

    if os.environ.get("AWS_BATCH_JOB_ID"):
        root_dir = os.path.join("/mnt", os.environ["AWS_BATCH_JOB_ID"])
    else:
        root_dir = "/mnt"

    run_dir = os.path.join(root_dir, "data")
    os.makedirs(run_dir)

    os.mkdir(os.path.join(run_dir, "reference"))
    os.mkdir(os.path.join(run_dir, "input"))

    if args.taxon == "hg38-plus":
        gtf_file = f"{reference_genomes[args.taxon]}.gtf"
        mask_file = "hg38_rmsk.gtf"
    elif args.taxon == "mm10-plus":
        gtf_file = f"{reference_genomes[args.taxon]}.gtf"
        mask_file = "mm10_rmsk.gtf"
    else:
        raise ValueError("Invalid taxon {}".format(args.taxon))

    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}
                    gtf_file:\t{gtf_file}
                    mask_file:\t{mask_file}
                        taxon:\t{args.taxon}
                s3_input_path:\t{args.s3_input_path}
                input_dirs:\t{', '.join(args.input_dirs)}"""
    )

    gtf_path = os.path.join(run_dir, "reference", gtf_file)
    mask_path = os.path.join(run_dir, "reference", mask_file)
    s3u.download_files(
        [f"velocyto/{gtf_file}", f"velocyto/{mask_file}"],
        [gtf_path, mask_path],
        bucket="czbiohub-reference",
        n_proc=2,
    )

    sample_re = re.compile(f"([^/]+).{args.taxon}.Aligned.out.sorted.bam$")
    plate_set = set(args.plates)

    s3_output_bucket, s3_output_prefix = s3u.s3_bucket_and_key(args.s3_output_path)

    for input_dir in args.input_dirs:
        logger.info(
            "Running partition {} of {} for {}".format(
                args.partition_id, args.num_partitions, input_dir
            )
        )

        # Check the output folder for existing runs
        if not args.force_redo:
            output = s3u.prefix_gen(
                s3_output_bucket,
                s3_output_prefix,
                lambda r: (r["LastModified"], r["Key"]),
            )
        else:
            output = []

        output_files = {
            os.path.basename(fn).split(".")[0]
            for dt, fn in output
            if fn.endswith(".loom") and dt > CURR_MIN_VER
        }

        # STAR alignment result files are either stored directly under the s3 input folder, or in sample sub-folders under the s3 input foloder
        if list(s3u.get_files(s3_input_bucket, s3_input_prefix)):
            print(
                "plain", list(s3u.get_files(s3_input_bucket, s3_input_prefix))
            )  # testing
            sample_files = [
                fn
                for fn in s3u.get_files(
                    s3_input_bucket, os.path.join(s3_input_prefix, input_dir)
                )
                if fn.endswith(f"{args.taxon}.Aligned.out.sorted.bam")
            ]
        else:
            print(
                "sub-folders", list(s3u.get_files(s3_input_bucket, s3_input_prefix))
            )  # testing
            sample_folder_paths = s3u.get_folders(
                s3_input_bucket, os.path.join(s3_input_prefix, input_dir)
            )
            for folder in sample_folder_paths:
                if not folder.endswith("/"):
                    folder += "/"
            sample_files = []
            for sample in sample_folder_paths:
                sample_files = [
                    fn
                    for fn in s3u.get_files(
                        s3_input_bucket, os.path.join(s3_input_prefix, input_dir)
                    )
                    if fn.endswith(f"{args.taxon}.Aligned.out.sorted.bam")
                ]
                sample_files += files

        # Run velocyto on the alignment results of specific plates if specified. Otherwise run velocyto on all input alignment results
        plate_samples = []

        for fn in sample_files:
            matched = sample_re.search(os.path.basename(fn))
            if matched.group(1) not in output_files:
                if len(plate_set) == 0 or matched.group(1).split("_")[1] in plate_set:
                    plate_samples.append(fn)

        logger.info(f"number of bam files: {len(plate_samples)}")

        for sample_name in sorted(plate_samples)[
            args.partition_id :: args.num_partitions
        ]:
            run_sample(
                sample_name,
                mask_path,
                gtf_path,
                s3_input_bucket,
                s3_output_bucket,
                os.path.join(s3_output_prefix, input_dir),
                run_dir,
                logger,
            )
            time.sleep(30)

        logger.info("Job completed")
 Download reference genome, run velocyto jobs, and upload results to S3.

        logger - Logger object that exposes the interface the code directly uses
    
Download reference genome, run velocyto jobs, and upload results to S3.
logger - Logger object that exposes the interface the code directly uses
Download reference genome run velocyto jobs and upload results to S3 . logger - Logger object that exposes the interface the code directly uses
def main(logger):
    parser = get_parser()
    args = parser.parse_args()
    if os.environ.get("AWS_BATCH_JOB_ID"):
        root_dir = os.path.join("/mnt", os.environ["AWS_BATCH_JOB_ID"])
    else:
        root_dir = "/mnt"
    run_dir = os.path.join(root_dir, "data")
    os.makedirs(run_dir)
    os.mkdir(os.path.join(run_dir, "reference"))
    os.mkdir(os.path.join(run_dir, "input"))
    if args.taxon == "hg38-plus":
        gtf_file = f"{reference_genomes[args.taxon]}.gtf"
        mask_file = "hg38_rmsk.gtf"
    elif args.taxon == "mm10-plus":
        gtf_file = f"{reference_genomes[args.taxon]}.gtf"
        mask_file = "mm10_rmsk.gtf"
    else:
        raise ValueError("Invalid taxon {}".format(args.taxon))
    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}
                    gtf_file:\t{gtf_file}
                    mask_file:\t{mask_file}
                        taxon:\t{args.taxon}
                s3_input_path:\t{args.s3_input_path}
                input_dirs:\t{', '.join(args.input_dirs)}"""
    )
    gtf_path = os.path.join(run_dir, "reference", gtf_file)
    mask_path = os.path.join(run_dir, "reference", mask_file)
    s3u.download_files(
        [f"velocyto/{gtf_file}", f"velocyto/{mask_file}"],
        [gtf_path, mask_path],
        bucket="czbiohub-reference",
        n_proc=2,
    )
    sample_re = re.compile(f"([^/]+).{args.taxon}.Aligned.out.sorted.bam$")
    plate_set = set(args.plates)
    s3_output_bucket, s3_output_prefix = s3u.s3_bucket_and_key(args.s3_output_path)
    for input_dir in args.input_dirs:
        logger.info(
            "Running partition {} of {} for {}".format(
                args.partition_id, args.num_partitions, input_dir
            )
        )
        if not args.force_redo:
            output = s3u.prefix_gen(
                s3_output_bucket,
                s3_output_prefix,
                lambda r: (r["LastModified"], r["Key"]),
            )
        else:
            output = []
        output_files = {
            os.path.basename(fn).split(".")[0]
            for dt, fn in output
            if fn.endswith(".loom") and dt > CURR_MIN_VER
        }
        if list(s3u.get_files(s3_input_bucket, s3_input_prefix)):
            print(
                "plain", list(s3u.get_files(s3_input_bucket, s3_input_prefix))
            )  
            sample_files = [
                fn
                for fn in s3u.get_files(
                    s3_input_bucket, os.path.join(s3_input_prefix, input_dir)
                )
                if fn.endswith(f"{args.taxon}.Aligned.out.sorted.bam")
            ]
        else:
            print(
                "sub-folders", list(s3u.get_files(s3_input_bucket, s3_input_prefix))
            )  
            sample_folder_paths = s3u.get_folders(
                s3_input_bucket, os.path.join(s3_input_prefix, input_dir)
            )
            for folder in sample_folder_paths:
                if not folder.endswith("/"):
                    folder += "/"
            sample_files = []
            for sample in sample_folder_paths:
                sample_files = [
                    fn
                    for fn in s3u.get_files(
                        s3_input_bucket, os.path.join(s3_input_prefix, input_dir)
                    )
                    if fn.endswith(f"{args.taxon}.Aligned.out.sorted.bam")
                ]
                sample_files += files
        plate_samples = []
        for fn in sample_files:
            matched = sample_re.search(os.path.basename(fn))
            if matched.group(1) not in output_files:
                if len(plate_set) == 0 or matched.group(1).split("_")[1] in plate_set:
                    plate_samples.append(fn)
        logger.info(f"number of bam files: {len(plate_samples)}")
        for sample_name in sorted(plate_samples)[
            args.partition_id :: args.num_partitions
        ]:
            run_sample(
                sample_name,
                mask_path,
                gtf_path,
                s3_input_bucket,
                s3_output_bucket,
                os.path.join(s3_output_prefix, input_dir),
                run_dir,
                logger,
            )
            time.sleep(30)
        logger.info("Job completed")
def main ( logger ) : parser = get_parser ( ) args = parser . parse_args ( ) if os . environ . get ( "AWS_BATCH_JOB_ID" ) : root_dir = os . path . join ( "/mnt" , os . environ [ "AWS_BATCH_JOB_ID" ] ) else : root_dir = "/mnt" run_dir = os . path . join ( root_dir , "data" ) os . makedirs ( run_dir ) os . mkdir ( os . path . join ( run_dir , "reference" ) ) os . mkdir ( os . path . join ( run_dir , "input" ) ) if args . taxon == "hg38-plus" : gtf_file = f"{reference_genomes[args.taxon]}.gtf" mask_file = "hg38_rmsk.gtf" elif args . taxon == "mm10-plus" : gtf_file = f"{reference_genomes[args.taxon]}.gtf" mask_file = "mm10_rmsk.gtf" else : raise ValueError ( "Invalid taxon {}" . format ( args . taxon ) ) 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}
                    gtf_file:\t{gtf_file}
                    mask_file:\t{mask_file}
                        taxon:\t{args.taxon}
                s3_input_path:\t{args.s3_input_path}
                input_dirs:\t{', '.join(args.input_dirs)}""" ) gtf_path = os . path . join ( run_dir , "reference" , gtf_file ) mask_path = os . path . join ( run_dir , "reference" , mask_file ) s3u . download_files ( [ f"velocyto/{gtf_file}" , f"velocyto/{mask_file}" ] , [ gtf_path , mask_path ] , bucket = "czbiohub-reference" , n_proc = 2 , ) sample_re = re . compile ( f"([^/]+).{args.taxon}.Aligned.out.sorted.bam$" ) plate_set = set ( args . plates ) s3_output_bucket , s3_output_prefix = s3u . s3_bucket_and_key ( args . s3_output_path ) for input_dir in args . input_dirs : logger . info ( "Running partition {} of {} for {}" . format ( args . partition_id , args . num_partitions , input_dir ) ) if not args . force_redo : output = s3u . prefix_gen ( s3_output_bucket , s3_output_prefix , lambda r : ( r [ "LastModified" ] , r [ "Key" ] ) , ) else : output = [ ] output_files = { os . path . basename ( fn ) . split ( "." ) [ 0 ] for dt , fn in output if fn . endswith ( ".loom" ) and dt > CURR_MIN_VER } if list ( s3u . get_files ( s3_input_bucket , s3_input_prefix ) ) : print ( "plain" , list ( s3u . get_files ( s3_input_bucket , s3_input_prefix ) ) ) sample_files = [ fn for fn in s3u . get_files ( s3_input_bucket , os . path . join ( s3_input_prefix , input_dir ) ) if fn . endswith ( f"{args.taxon}.Aligned.out.sorted.bam" ) ] else : print ( "sub-folders" , list ( s3u . get_files ( s3_input_bucket , s3_input_prefix ) ) ) sample_folder_paths = s3u . get_folders ( s3_input_bucket , os . path . join ( s3_input_prefix , input_dir ) ) for folder in sample_folder_paths : if not folder . endswith ( "/" ) : folder += "/" sample_files = [ ] for sample in sample_folder_paths : sample_files = [ fn for fn in s3u . get_files ( s3_input_bucket , os . path . join ( s3_input_prefix , input_dir ) ) if fn . endswith ( f"{args.taxon}.Aligned.out.sorted.bam" ) ] sample_files += files plate_samples = [ ] for fn in sample_files : matched = sample_re . search ( os . path . basename ( fn ) ) if matched . group ( 1 ) not in output_files : if len ( plate_set ) == 0 or matched . group ( 1 ) . split ( "_" ) [ 1 ] in plate_set : plate_samples . append ( fn ) logger . info ( f"number of bam files: {len(plate_samples)}" ) for sample_name in sorted ( plate_samples ) [ args . partition_id : : args . num_partitions ] : run_sample ( sample_name , mask_path , gtf_path , s3_input_bucket , s3_output_bucket , os . path . join ( s3_output_prefix , input_dir ) , run_dir , logger , ) time . sleep ( 30 ) logger . info ( "Job completed" )
Download reference genome, run velocyto jobs, and upload results to S3.
Download reference genome run velocyto jobs and upload results to S3 .
""" Download reference genome, run velocyto jobs, and upload results to S3.

        logger - Logger object that exposes the interface the code directly uses
    """ # Check the output folder for existing runs # STAR alignment result files are either stored directly under the s3 input folder, or in sample sub-folders under the s3 input foloder # testing # testing # Run velocyto on the alignment results of specific plates if specified. Otherwise run velocyto on all input alignment results
logger