Input and output file generation command (#63)

* Simple generation

* Fix problems

* Add support for shell ingens

* Fix executing shell ingens

* Add comments for struct

* Add package with shell ingen

* Add unit tests for gen command

* Bash parallel input generation

* Apply suggestions from code review

Co-authored-by: Tomasz Nowak <[email protected]>

* Rename variable

* Change compile_ingen function

* Change compile_correct_solution function

* Add lambda function for compiling solutions

* Refactor ingen execution

* Use weak-compilation-flags flag

* Change to use imap instead of imap_unordered

* Fix tests

* Refactor

* Rename files and add correct solution in gen test package

* Fix typo

* Add integration tests for `gen` command

* Add execution permission to ingen.sh when running

* Small refactor

* Change ingen.sh in package

* Refactor code

* Return less Nones in gen

* Fix tests

* Add function for calling function and exiting

* Refactor

* Refactor `util.exit_with_error`

* Use python for giving execution permission and cleanup imports

* Apply suggestions from code review

Co-authored-by: Tomasz Nowak <[email protected]>

* Fix tests

* Bump version for release

* Add function for fixing line endings

* Refactor .md5sums parsing

---------

Co-authored-by: Tomasz Nowak <[email protected]>

Author: MasloMaslane

src/sinol_make/__init__.py

@@ -6,7 +6,7 @@
 
 from sinol_make import util
 
-__version__ = "1.3.2"
+__version__ = "1.4.0"
 
 def configure_parsers():
     parser = argparse.ArgumentParser(

src/sinol_make/commands/gen/__init__.py

@@ -0,0 +1,119 @@
+import argparse
+import glob
+import os
+import hashlib
+import yaml
+
+import multiprocessing as mp
+
+from sinol_make import util
+from sinol_make.commands.gen import gen_util
+from sinol_make.commands.gen.structs import OutputGenerationArguments
+from sinol_make.helpers import parsers, package_util, compile
+from sinol_make.interfaces.BaseCommand import BaseCommand
+
+
+class Command(BaseCommand):
+    """
+    Class for `gen` command.
+    """
+
+    def get_name(self):
+        return "gen"
+
+    def configure_subparser(self, subparser):
+        parser = subparser.add_parser(
+            self.get_name(),
+            help='Generate input and output files',
+            description='Generate input files using ingen program '
+                        '(for example prog/abcingen.cpp for abc task). Whenever '
+                        'the new input differs from the previous one, '
+                        'the model solution will be used to generate the new output '
+                        'file. You can also specify your ingen source '
+                        'file which will be used.'
+        )
+
+        parser.add_argument('ingen_path', type=str, nargs='?',
+                            help='path to ingen source file, for example prog/abcingen.cpp')
+        parser.add_argument('-c', '--cpus', type=int,
+                            help=f'number of cpus to use, by default {mp.cpu_count()} (all available). '
+                                 f'Used when generating output files.', default=mp.cpu_count())
+        parsers.add_compilation_arguments(parser)
+
+    def generate_outputs(self, outputs_to_generate):
+        print(f'Generating output files for {len(outputs_to_generate)} tests on {self.args.cpus} cpus.')
+        arguments = []
+        for output in outputs_to_generate:
+            output_basename = os.path.basename(output)
+            input = os.path.join(os.getcwd(), 'in', os.path.splitext(output_basename)[0] + '.in')
+            arguments.append(OutputGenerationArguments(self.correct_solution_exe, input, output))
+
+        with mp.Pool(self.args.cpus) as pool:
+            results = []
+            for i, result in enumerate(pool.imap(gen_util.generate_output, arguments)):
+                results.append(result)
+                if result:
+                    print(util.info(f'Successfully generated output file {os.path.basename(arguments[i].output_test)}'))
+                else:
+                    print(util.error(f'Failed to generate output file {os.path.basename(arguments[i].output_test)}'))
+
+            if not all(results):
+                util.exit_with_error('Failed to generate some output files.')
+            else:
+                print(util.info('Successfully generated all output files.'))
+
+    def calculate_md5_sums(self):
+        """
+        Calculates md5 sums for each test.
+        :return: Tuple (dictionary of md5 sums, list of outputs tests that need to be generated)
+        """
+        old_md5_sums = None
+        try:
+            with open(os.path.join(os.getcwd(), 'in', '.md5sums'), 'r') as f:
+                file_content = yaml.load(f, Loader=yaml.FullLoader)
+                if isinstance(file_content, dict):
+                    old_md5_sums = file_content
+        except (yaml.YAMLError, OSError):
+            pass
+
+        md5_sums = {}
+        outputs_to_generate = []
+        for file in glob.glob(os.path.join(os.getcwd(), 'in', '*.in')):
+            basename = os.path.basename(file)
+            md5_sums[basename] = hashlib.md5(open(file, 'rb').read()).hexdigest()
+            if old_md5_sums is None or old_md5_sums.get(basename, '') != md5_sums[basename]:
+                output_basename = os.path.splitext(os.path.basename(basename))[0] + '.out'
+                outputs_to_generate.append(os.path.join(os.getcwd(), "out", output_basename))
+
+        return md5_sums, outputs_to_generate
+
+    def run(self, args: argparse.Namespace):
+        if not util.check_if_project():
+            util.exit_with_error('You are not in a project directory (couldn\'t find config.yml in current directory).')
+
+        self.args = args
+        self.task_id = package_util.get_task_id()
+        self.ingen = gen_util.get_ingen(self.task_id, args.ingen_path)
+        print(util.info(f'Using ingen file {os.path.basename(self.ingen)}'))
+
+        self.correct_solution = gen_util.get_correct_solution(self.task_id)
+
+        if os.path.splitext(self.ingen)[1] != '.sh':
+            self.ingen_exe = gen_util.compile_ingen(self.ingen, self.args, self.args.weak_compilation_flags)
+        else:
+            self.ingen_exe = self.ingen
+
+        if gen_util.run_ingen(self.ingen_exe):
+            print(util.info('Successfully generated input files.'))
+        else:
+            util.exit_with_error('Failed to generate input files.')
+        md5_sums, outputs_to_generate = self.calculate_md5_sums()
+        if len(outputs_to_generate) == 0:
+            print(util.info('All output files are up to date.'))
+        else:
+            self.correct_solution_exe = gen_util.compile_correct_solution(self.correct_solution, self.args,
+                                                                          self.args.weak_compilation_flags)
+            self.generate_outputs(outputs_to_generate)
+
+        with open(os.path.join(os.getcwd(), 'in', '.md5sums'), 'w') as f:
+            yaml.dump(md5_sums, f)

src/sinol_make/commands/gen/gen_util.py

@@ -0,0 +1,137 @@
+import glob
+import os
+import stat
+import argparse
+import signal
+import subprocess
+import sys
+
+from sinol_make import util
+from sinol_make.helpers import compiler, package_util, compile
+
+
+def get_ingen(task_id=None, ingen_path=None):
+    """
+    Find ingen source file in `prog/` directory.
+    If `ingen_path` is specified, then it will be used (if exists).
+    :param task_id: task id, for example abc. If None, then
+                    will return any ingen matching "*ingen.*"
+    :param ingen_path: path to ingen source file
+    :return: path to ingen source file or None if not found
+    """
+
+    if ingen_path is not None:
+        if os.path.exists(ingen_path):
+            return ingen_path
+        else:
+            util.exit_with_error(f'Ingen source file {ingen_path} does not exist.')
+
+    if task_id is None:
+        task_id = '*'
+    ingen = glob.glob(os.path.join(os.getcwd(), 'prog', task_id + 'ingen.*'))
+    if len(ingen) == 0:
+        util.exit_with_error(f'Ingen source file for task {task_id} does not exist.')
+
+    # Sio2 first chooses shell scripts, then non-shell source codes.
+    if os.path.splitext(ingen[0])[1] == '.sh' and len(ingen) > 1:
+        return ingen[1]
+    else:
+        return ingen[0]
+
+
+def compile_ingen(ingen_path: str, args: argparse.Namespace, weak_compilation_flags=False):
+    """
+    Compiles ingen and returns path to compiled executable.
+    """
+    compilers = compiler.verify_compilers(args, [ingen_path])
+    ingen_exe, compile_log_path = compile.compile_file(ingen_path, package_util.get_executable(ingen_path), compilers, weak_compilation_flags)
+
+    if ingen_exe is None:
+        util.exit_with_error('Failed ingen compilation.', lambda: compile.print_compile_log(compile_log_path))
+    else:
+        print(util.info('Successfully compiled ingen.'))
+    return ingen_exe
+
+
+def get_correct_solution(task_id):
+    """
+    Returns path to correct solution for given task.
+    :param task_id: task id, for example abc
+    :return: path to correct solution or None if not found
+    """
+    correct_solution = glob.glob(os.path.join(os.getcwd(), 'prog', task_id + '.*'))
+    if len(correct_solution) == 0:
+        util.exit_with_error(f'Correct solution for task {task_id} does not exist.')
+    return correct_solution[0]
+
+
+def compile_correct_solution(solution_path: str, args: argparse.Namespace, weak_compilation_flags=False):
+    """
+    Compiles correct solution and returns path to compiled executable.
+    """
+    compilers = compiler.verify_compilers(args, [solution_path])
+    correct_solution_exe, compile_log_path = compile.compile_file(solution_path, package_util.get_executable(solution_path), compilers,
+                                weak_compilation_flags)
+    if correct_solution_exe is None:
+        util.exit_with_error('Failed compilation of correct solution.',
+                                  lambda: compile.print_compile_log(compile_log_path))
+    else:
+        print(util.info('Successfully compiled correct solution.'))
+
+    return correct_solution_exe
+
+
+
+def run_ingen(ingen_exe):
+    """
+    Runs ingen and generates all input files.
+    :param ingen_exe: path to ingen executable
+    :return: True if ingen was successful, False otherwise
+    """
+    is_shell = os.path.splitext(ingen_exe)[1] == '.sh'
+    if is_shell:
+        util.fix_file_endings(ingen_exe)
+        st = os.stat(ingen_exe)
+        os.chmod(ingen_exe, st.st_mode | stat.S_IEXEC)
+
+    process = subprocess.Popen([ingen_exe], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                               cwd=os.path.join(os.getcwd(), 'in'), shell=is_shell)
+    while process.poll() is None:
+        print(process.stdout.readline().decode('utf-8'), end='')
+
+    print(process.stdout.read().decode('utf-8'), end='')
+    exit_code = process.returncode
+
+    return exit_code == 0
+
+
+def generate_output(arguments):
+    """
+    Generates output file for given input file.
+    :param arguments: arguments for output generation (type OutputGenerationArguments)
+    :return: True if the output was successfully generated, False otherwise
+    """
+    input_test = arguments.input_test
+    output_test = arguments.output_test
+    correct_solution_exe = arguments.correct_solution_exe
+
+    input_file = open(input_test, 'r')
+    output_file = open(output_test, 'w')
+    process = subprocess.Popen([correct_solution_exe], stdin=input_file, stdout=output_file, preexec_fn=os.setsid)
+    previous_sigint_handler = signal.getsignal(signal.SIGINT)
+
+    def sigint_handler(signum, frame):
+        try:
+            os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+        except ProcessLookupError:
+            pass
+        sys.exit(1)
+    signal.signal(signal.SIGINT, sigint_handler)
+
+    process.wait()
+    signal.signal(signal.SIGINT, previous_sigint_handler)
+    exit_code = process.returncode
+    input_file.close()
+    output_file.close()
+
+    return exit_code == 0

src/sinol_make/commands/gen/structs.py

@@ -0,0 +1,14 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class OutputGenerationArguments:
+    """
+    Arguments used for function that generates output file.
+    """
+    # Path to correct solution executable
+    correct_solution_exe: str
+    # Path to input file
+    input_test: str
+    # Path to output file
+    output_test: str

src/sinol_make/commands/inwer/__init__.py

@@ -42,9 +42,7 @@ def configure_subparser(self, subparser: argparse.ArgumentParser):
     def compile_inwer(self, args: argparse.Namespace):
         self.inwer_executable, compile_log_path = inwer_util.compile_inwer(self.inwer, args, args.weak_compilation_flags)
         if self.inwer_executable is None:
-            print(util.error('Compilation failed.'))
-            compile.print_compile_log(compile_log_path)
-            exit(1)
+            util.exit_with_error('Compilation failed.', lambda: compile.print_compile_log(compile_log_path))
         else:
             print(util.info('Compilation successful.'))
 

src/sinol_make/helpers/compile.py

@@ -1,9 +1,13 @@
 from typing import Tuple
+import os
+import sys
+import stat
+import subprocess
 
 import sinol_make.helpers.compiler as compiler
 from sinol_make.interfaces.Errors import CompilationError
 from sinol_make.structs.compiler_structs import Compilers
-import os, subprocess, sys
+
 
 def compile(program, output, compilers: Compilers = None, compile_log = None, weak_compilation_flags = False):
     """
@@ -32,7 +36,8 @@ def compile(program, output, compilers: Compilers = None, compile_log = None, we
         else:
             open(output, 'w').write('#!/usr/bin/python3\n')
             open(output, 'a').write(open(program, 'r').read())
-            subprocess.call(['chmod', '+x', output])
+            st = os.stat(output)
+            os.chmod(output, st.st_mode | stat.S_IEXEC)
         arguments = [compilers.python_interpreter_path, '-m', 'py_compile', program]
     elif ext == '.java':
         raise NotImplementedError('Java compilation is not implemented')

src/sinol_make/helpers/compiler.py

@@ -80,6 +80,18 @@ def get_java_compiler_path():
         return 'javac'
 
 
+def get_default_compilers():
+    """
+    Get the default compilers
+    """
+    return argparse.Namespace(
+        c_compiler_path=get_c_compiler_path(),
+        cpp_compiler_path=get_cpp_compiler_path(),
+        python_interpreter_path=get_python_interpreter_path(),
+        java_compiler_path=get_java_compiler_path()
+    )
+
+
 def verify_compilers(args: argparse.Namespace, solutions: list[str]) -> Compilers:
     for solution in solutions:
         ext = os.path.splitext(solution)[1]

src/sinol_make/util.py

@@ -263,6 +263,14 @@ def get_terminal_size():
         terminal_height = 30
     return has_terminal, terminal_width, terminal_height
 
+
+def fix_file_endings(file):
+    with open(file, "rb") as f:
+        content = f.read()
+    with open(file, "wb") as f:
+        f.write(content.replace(b"\r\n", b"\n"))
+
+
 def color_red(text): return "\033[91m{}\033[00m".format(text)
 def color_green(text): return "\033[92m{}\033[00m".format(text)
 def color_yellow(text): return "\033[93m{}\033[00m".format(text)
@@ -275,6 +283,9 @@ def warning(text):
 def error(text):
     return bold(color_red(text))
 
-def exit_with_error(text):
+
+def exit_with_error(text, func=None):
     print(error(text))
+    if func is not None:
+        func()
     exit(1)