Experiment scheduling script

simple_schedule.py

import argparse
import json
import subprocess
from datetime import datetime
from itertools import starmap


def get_args() -> argparse.Namespace:
    """Parse command-line arguments"""
    parser = argparse.ArgumentParser(prog="simple_schedule",
                                     description="Simple experiment scheduler")

    parser.add_argument("--json", type=str, required=True,
                        help="JSON config filename")
    parser.add_argument("--output", type=str, required=True,
                        help="Output filename")

    return parser.parse_args()


def check_cmd(cmd: dict) -> None:
    """Check that command dict includes required keys"""
    required_keys = ["run_name", "protocol"]
    for k in required_keys:
        if k not in cmd.keys():
            raise KeyError(f"Argument missing in arguments dict: {k}")


def build_cmd_str(args_dict: dict) -> str:
    """Build bash nohup command for an experiment

    Keyword arguments:
    args_dict -- dictionary in the format "arg:value" for command-line args
                 to be used in the command
    """
    cmd_prefix = "python3 main.py"
    args_strings = starmap(lambda a, v: f"--{a} {v}", args_dict.items())
    cmd_as_list = [cmd_prefix, " ".join(args_strings)]
    return " ".join(cmd_as_list)


def log_line(*args: str) -> None:
    """Append line to output file"""
    time = datetime.now().strftime("%d/%m/%y %H:%M:: ")
    with open(opt.output, 'a') as f:
        f.write(time + " ".join(args) + "\n")


opt = get_args()
cmds_json = json.load(open(opt.json, 'r'))

for cmd in cmds_json:
    check_cmd(cmd)
    now_as_hexdec = hex(int(datetime.now().timestamp()))
    cmd["run_name"] = cmd["run_name"] + ' ' + now_as_hexdec
    log_line("+experiment:", cmd["run_name"])
    cmd_str = build_cmd_str(cmd)
    log_line("    ", cmd_str)
    subprocess.run(build_cmd_str(cmd), shell=True, check=True)
    log_line("%completed\n")