gen.py

#!/usr/bin/python3

from os import path, listdir, replace, remove
from subprocess import run, PIPE, TimeoutExpired, CalledProcessError
from datetime import datetime

from common import *

from check import verify_test

def generate_input(test_name, generator_command):
    input_file_name = input_path(test_name)

    generator_args = generator_command.split()
    generator_name = generator_args[0]

    generator_path = path.join(GENS_DIR, generator_name)

    print(f"Generating {test_name}.in ... ", end="", flush=True)
    try:
        called = run([generator_path, *generator_args[1:]], stdout=PIPE, stderr=PIPE)
        called.check_returncode()
    except CalledProcessError as err:
        print("Error!!")
        print(called.stderr.decode())
    else:
        print("Done.")
        print(f"Writing to {test_name}.in ... ", end="", flush=True)
        with open(input_file_name, "wb") as input_file:
            input_file.write(called.stdout)
        print("Done.")
    return called.returncode == 0

def generate_output(test_name, solution_name, timeout=None, keep_old=True):

    input_file_name = input_path(test_name)
    output_file_name = path.join(OUT_DIR, f"{test_name}.out")

    solution_path = path.join(SOLS_DIR, solution_name)

    print(f"Solving case {test_name}.in ... ", end="", flush=True)
    success=True
    with open(input_file_name) as input_file:
        try:
            start_time = datetime.now()
            comp = run(solution_path, stdin=input_file, stdout=PIPE, stderr=PIPE, timeout=timeout)
            end_time = datetime.now()
            print((end_time-start_time).total_seconds(),end=" ")
            comp.check_returncode()

        except CalledProcessError as err:
            print("Error!!")
            print(err)
            print(comp.stderr.decode())
            success=False
        except TimeoutExpired as err:
            print("Timeout!!")
            print(err)
            success=False
        else:
            print("Done.")
            if path.isfile(output_file_name):
                if keep_old:
                    replace(output_file_name, f"{output_file_name}.old")
                else:
                    print(f"Overwriting {test_name}.out.")
            print(f"Writing to {test_name}.out ...", end="", flush=True)
            with open(output_file_name, "wb") as output_file:
                output_file.write(comp.stdout)
            print("Done.")

        if path.isfile(f"{output_file_name}.old"):
            try:
                run(["diff", output_file_name, f"{output_file_name}.old"], check=True)
            except CalledProcessError:
                print(f"Warning: Conflicting output files!! ({test_name})")
            else:
                remove(f"{output_file_name}.old")
        
    return success
    
def generate_input_output(test_name, generator_command=None, verifier_name=None, solution_name=None, timeout=None):
    input_file_name = input_path(test_name)
    output_file_name  = path.join(OUT_DIR, f"{test_name}.out")

    generator_success, verifier_success, output_success = False, False, False

    if generator_command is not None:
        generator_success = generate_input(test_name, generator_command)
        if not generator_success:
            return generator_success, verifier_success, output_success
        if path.isfile(output_file_name):
            print(f"Deleting old {test_name}.out")
            remove(output_file_name)
    else:
        if not path.isfile(input_file_name):
            print(f"{test_name}.in does not exist!")
            return generator_success, verifier_success, output_success
    
    if verifier_name is not None:
        print(f"Verifying {test_name}.in ... ", end="", flush=True)
        verifier_success, description = verify_test(test_name, verifier_name)
        if not verifier_success:
            print("Error!")
            print(description.stderr.decode())
            # print(f"Renaming invalid {test_name}.in")
            # replace(input_file_name, f"{input_file_name}.invalid")
            return generator_success, verifier_success, output_success
        else:
            print("Done.")
    
    if solution_name is not None:
        output_success = generate_output(test_name, solution_name, timeout=timeout, keep_old=(generator_command is None))
    
    return generator_success, verifier_success, output_success



if __name__ == "__main__":
    from getopt import getopt, GetoptError
    import sys

    short_opts = "g:s:v:t:yn"
    try:
        opts, args = getopt(sys.argv[1:], short_opts)
    except GetoptError as err:
        print(err)
        sys.exit(2)
    
    opt_dict = {opt:value for opt, value in opts}

    generator_command = opt_dict.get("-g")
    verifier_name = opt_dict.get("-v")
    solution_name = opt_dict.get("-s")
    timeout_str = opt_dict.get("-t")

    failure = False

    if generator_command is not None:
        generator_args = generator_command.split()
        if len(generator_args) == 0:
            generator_name=""
        else:
            generator_name = generator_args[0]
        if not path.isfile(path.join(GENS_DIR, generator_name)):
            print(f"No such generator ({generator_name})!")
            failure = True

    if verifier_name is not None:
        if not path.isfile(path.join(VERS_DIR, verifier_name)):
            print(f"No such verifier ({verifier_name})!")
            failure = True

    if solution_name is not None:
        if not path.isfile(path.join(SOLS_DIR, solution_name)):
            print(f"No such solution ({solution_name})!")
            failure = True
    
    if timeout_str is not None:
        try:
            timeout = float(timeout_str)
            assert timeout > 0
        except (ValueError, AssertionError):
            print(f"Invalid Timeout ({timeout_str})")
            failure=True
    else:
        timeout=None
    
    if "-y" in opt_dict and "-n" in opt_dict:
        print("Options -y and -n are mutually exclusive!")
        failure = True
    
    if len(args) < 1:
        print("Missing argument <test_name>")
        failure = True
    
    if len(args) > 1:
        print("Too many arguments!")
        failure = True
    
    if failure:
        print("Aborting.")
        sys.exit(2)
    
    test_name = args[0]
    
    input_file_name = input_path(test_name)
    if path.isfile(input_file_name):
        print(f"File {test_name}.in already exists")
        if "-n" in opt_dict:
            generator_command=None
        elif "-y" not in opt_dict:
            yes_or_no = input("Replace exisiting case? (y/n) ").strip().lower()
            if yes_or_no != "y":
                generator_command=None
    
    generate_input_output(test_name, generator_command=generator_command, verifier_name=verifier_name, solution_name=solution_name, timeout=timeout)