transcribe Namespace Reference

Functions
def	parse (List[str] argv)
def	create_new_dag (str script, str outputDir, bool verbose=False)
def	main ()

Function Documentation

create_new_dag()

def transcribe.create_new_dag	(	str	script,
		str	outputDir,
		bool	verbose = False
	)

Actual script transcription and preparation of dag directory

def create_new_dag(script: str, outputDir: str, verbose: bool = False):
    """Actual script transcription and preparation of dag directory"""
 
    outputDir.mkdir(parents=True)
 
    # copy the script to the output directory with an additional alias
    # to replace `submit` with `job`
    lines = []
    with open(script, "r") as f:
        lines = f.readlines()
    alias = f"alias submit='job -D {outputDir}'"
    if verbose:
        alias += " -v"
    lines = [alias + "\n"] + lines
    if lines[1][:2] == "#!":
        lines[0], lines[1] = lines[1], lines[0]
    script = outputDir / "script"
    with open(script, "w") as f:
        for line in lines:
            f.write(line)
    os.chmod(script, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
 
    # run the modified script to make the DAG
    subprocess.run(str(script), shell=True, check=True)
    # \todo in case, capture environment from the script (`/proc/<pid>/environ`)
 
    # add the dot file to visualise the dependencies
    with open(outputDir / "dag", "a") as f:
        print(f"DOT {outputDir}/jobs.dot", file=f)
 
    # copy the libraries and dictionaries in a safe place
    for ext in ["so", "pcm"]:
        for lib in glob(os.environ["DARWIN_FIRE_AND_FORGET"] + "/*" + ext):
            if not Path(lib).exists():
                copy2(lib, outputDir)
 
    # \todo copy any config file found in the input script
 
 

main()

def transcribe.main

(

)

Perform the transcription of a script with submit prefix commands
to a DAG format, which can then be submitted on HTCondor.

def main():
    """Perform the transcription of a script with submit prefix commands
    to a DAG format, which can then be submitted on HTCondor."""
 
    # retrieve and define a few variables
    args = parse(sys.argv)
    script = Path(args.script)
    batchname = script.name
 
    # prepare output directory
    outputDir = Path(args.dag)
    outputDir_preexists = outputDir.exists()
    rescue_preexists = len(list(outputDir.glob("dag.rescue*"))) > 0
    if outputDir_preexists:
        print("(Re)submitting existing dag.")
    else:
        create_new_dag(script, outputDir, args.verbose)
 
    # prepare and run dag
    if outputDir_preexists and not rescue_preexists:
        cmd = f"condor_submit -batch-name {batchname}"
    else:
        cmd = f"condor_submit_dag -import_env -batch-name {batchname}"
 
    if args.verbose:
        cmd += " -verbose"
    if args.dry_run:
        cmd += " -no_submit"
 
    if outputDir_preexists and not rescue_preexists:
        cmd += f" dag.condor.sub"
    else:
        cmd += f" dag"
 
    # prepare the environment
    path = os.environ["PATH"]
    ld_library_path = os.environ["LD_LIBRARY_PATH"]
    env = os.environ.copy()
    env["PATH"] = f"{outputDir}:{path}"
    env["LD_LIBRARY_PATH"] = f"{outputDir}:{ld_library_path}"
    if args.verbose:
        print(cmd)
    if which("condor_submit_dag") is None or which("condor_submit") is None:
        print("HTCondor shell commands cannot be found.")
        sys.exit(not args.dry_run)
    subprocess.run(cmd, shell=True, env=env, check=True, cwd=outputDir)
    if args.dry_run:
        sys.exit()
 
    # prepare the babysitting
    cmd = "condor_q -dag -nobatch"  # \todo constraint
    if args.background:
        print(f"Check the status of your jobs with `{cmd}`.")
        sys.exit()
 
    # prepare killing jobs on Ctrl+C
    def handler(signum, frame):
        subprocess.run(
            f"condor_rm -constraint 'JobBatchName == \"{batchname}\"'", shell=True
        )
        sys.exit()
 
    signal.signal(signal.SIGINT, handler)
 
    # watch jobs
    cmd = f"watch -t -d {cmd}"
    subprocess.run(cmd, check=True, shell=True)
 
 

parse()

def transcribe.parse

(

List[str]

argv

)

def parse(argv: List[str]):
    parser = ArgumentParser()
    parser.add_argument(
        "-b",
        "--background",
        action="store_true",
        help="exit right after submitting the job",
    )
    parser.add_argument(
        "-d", "--dry-run", action="store_true", help="exit right before submitting"
    )
    parser.add_argument(
        "-D",
        "--dag",
        help="indicate the DAG directory",
        type=str,
        nargs='?',
        default="dag"
    )
    parser.add_argument(
        "-t", "--tutorial", action="store_true", help="show a brief tutorial"
    )
    parser.add_argument(
        "-v", "--verbose", action="store_true", help="increase verbosity"
    )
    parser.add_argument(
        "script",
        nargs="?",
        type=str,
        help="shell script including a series of `submit` statements",
    )
 
    args = parser.parse_args(argv[1:])
    if args.tutorial:
        print(
            "Replaces `submit` prefix commands with `job` prefix commands in the input script "
            "to produce and submit a DAG file, with the fire-and-forget strategy. The full "
            "power of HTCondor DAGMans may then be used, in particular to resubmit failed jobs."
        )
        sys.exit()
 
    if len(args.script) == 0:
        raise ValueError("Missing input script")
 
    return args