src/utilities/rna_velocity/run_kb_python.py
def main(logger):
    """Command-line entrypoint. Download fastqs from the S3 bucket to an EC2 instance, build kallisto index, run kallisto peusoalignment, generate count matrices, and upload the data quantification results back to the S3 bucket.

    Keyword Argument:    
    logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
    """
    # Parse input arguments
    parser = get_parser()
    args = parser.parse_args()
    print(f"testing purpose - what is sys.argv: {sys.argv}")
    print(f"testing purpose - what is args: {args}")

    # Root direcotry path on the EC2 instance
    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"]

    # Directory on the EC2 instance where relevant files for the run are stored
    run_dir = root_dir / "data"
    run_dir.mkdir(parents=True)

    # Run `kb --info` to see kb_python package details, or `kb --list` to see supported technologies list
    if "info" in sys.argv:
        display_info(logger)
    elif "--list" in sys.argv:
        display_technologies(logger)

    # Use the updated input format for non-standard workflows
    if any(arg in sys.argv for arg in {"--lamanno", "--nucleus"}):
        logger.warning(
            (
                "The `--lamanno` and `--nucleus` flags are deprecated. "
                "These options will be removed in a future release. "
                "Please use `--workflow lamanno` or `--workflow nucleus` instead when the --workflow argument is available."
            )
        )

    # Create tmp directory on the EC2 instance to store temporary files
    logger.debug("Creating tmp directory")
    tmp_dir = run_dir / "tmp"
    tmp_dir.mkdir(parents=True)

    # Run the command, and delete the temporary directory unless otherwise noted in the input
    try:
        logger.debug(args)
        print(
            f"testing purpose - see if args.command returns `ref` or `count`: {args.command}"
        )  # testing purpose
        COMMAND_TO_FUNCTION[args.command](args, run_dir, logger)
    finally:
        # Always clean temp dir
        if not args.keep_tmp:
            logger.debug("Removing tmp directory")
            shutil.rmtree(tmp_dir, ignore_errors=True)

    logger.info("Job completed")
Command-line entrypoint. Download fastqs from the S3 bucket to an EC2 instance, build kallisto index, run kallisto peusoalignment, generate count matrices, and upload the data quantification results back to the S3 bucket.

    Keyword Argument:    
    logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
    
Command-line entrypoint. Download fastqs from the S3 bucket to an EC2 instance, build kallisto index, run kallisto peusoalignment, generate count matrices, and upload the data quantification results back to the S3 bucket.
Keyword Argument:
logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
Command - line entrypoint . Download fastqs from the S3 bucket to an EC2 instance build kallisto index run kallisto peusoalignment generate count matrices and upload the data quantification results back to the S3 bucket . Keyword Argument : logger -- Logger object that exposes the interface the code directly uses ( type : logging . Logger )
def main(logger):
    parser = get_parser()
    args = parser.parse_args()
    print(f"testing purpose - what is sys.argv: {sys.argv}")
    print(f"testing purpose - what is args: {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"]
    run_dir = root_dir / "data"
    run_dir.mkdir(parents=True)
    if "info" in sys.argv:
        display_info(logger)
    elif "--list" in sys.argv:
        display_technologies(logger)
    if any(arg in sys.argv for arg in {"--lamanno", "--nucleus"}):
        logger.warning(
            (
                "The `--lamanno` and `--nucleus` flags are deprecated. "
                "These options will be removed in a future release. "
                "Please use `--workflow lamanno` or `--workflow nucleus` instead when the --workflow argument is available."
            )
        )
    logger.debug("Creating tmp directory")
    tmp_dir = run_dir / "tmp"
    tmp_dir.mkdir(parents=True)
    try:
        logger.debug(args)
        print(
            f"testing purpose - see if args.command returns `ref` or `count`: {args.command}"
        )  
        COMMAND_TO_FUNCTION[args.command](args, run_dir, logger)
    finally:
        if not args.keep_tmp:
            logger.debug("Removing tmp directory")
            shutil.rmtree(tmp_dir, ignore_errors=True)
    logger.info("Job completed")
def main ( logger ) : parser = get_parser ( ) args = parser . parse_args ( ) print ( f"testing purpose - what is sys.argv: {sys.argv}" ) print ( f"testing purpose - what is args: {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" ] run_dir = root_dir / "data" run_dir . mkdir ( parents = True ) if "info" in sys . argv : display_info ( logger ) elif "--list" in sys . argv : display_technologies ( logger ) if any ( arg in sys . argv for arg in { "--lamanno" , "--nucleus" } ) : logger . warning ( ( "The `--lamanno` and `--nucleus` flags are deprecated. " "These options will be removed in a future release. " "Please use `--workflow lamanno` or `--workflow nucleus` instead when the --workflow argument is available." ) ) logger . debug ( "Creating tmp directory" ) tmp_dir = run_dir / "tmp" tmp_dir . mkdir ( parents = True ) try : logger . debug ( args ) print ( f"testing purpose - see if args.command returns `ref` or `count`: {args.command}" ) COMMAND_TO_FUNCTION [ args . command ] ( args , run_dir , logger ) finally : if not args . keep_tmp : logger . debug ( "Removing tmp directory" ) shutil . rmtree ( tmp_dir , ignore_errors = True ) logger . info ( "Job completed" )
Command-line entrypoint.
Command - line entrypoint .
"""Command-line entrypoint. Download fastqs from the S3 bucket to an EC2 instance, build kallisto index, run kallisto peusoalignment, generate count matrices, and upload the data quantification results back to the S3 bucket.

    Keyword Argument:    
    logger -- Logger object that exposes the interface the code directly uses (type: logging.Logger)
    """ # Parse input arguments # Root direcotry path on the EC2 instance # Directory on the EC2 instance where relevant files for the run are stored # Run `kb --info` to see kb_python package details, or `kb --list` to see supported technologies list # Use the updated input format for non-standard workflows # Create tmp directory on the EC2 instance to store temporary files # Run the command, and delete the temporary directory unless otherwise noted in the input # testing purpose # Always clean temp dir
logger