studentCodeEvaluator.py

#!/usr/bin/env python
# studentCodeEvaluator.py
# *
# *  Copyright (C) 2020 John Moore
# *
# *  This Program is free software; you can redistribute it and/or modify
# *  it under the terms of the GNU General Public License as published by
# *  the Free Software Foundation; either version 2, or (at your option)
# *  any later version.
# *
# *  This Program is distributed in the hope that it will be useful,
# *  but WITHOUT ANY WARRANTY; without even the implied warranty of
# *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# *  GNU General Public License for more details.
# *
# * write to the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
# * http://www.gnu.org/copyleft/gpl.html
# *

"""
studentCodeEvaluator is a program that helps an instructor grade python code written by their students.

The program expects the student code files to be in their own directory. This means that the instructor can create
unique assignments and create folders that contain the student's code for that assignment.

If the student's programs read data from stdin, the instructor can create a response file that will supply input to
their programs. The response file is a simple text file that has one response per line. The response file is located
in the assignment directory by default.

studentCodeEvaluator generates a log file named after the folder the code resides in. The log file contains the output of
stdout and stderr for each program run. By default, the log file if located in the assignment directory.

syntax:
    studentCodeEvaluator assignmentDir -r <responsefile> -l <logfilepath> -t <timeout> -p <pgmArgs> -n <numReturns>

    assignmentDir (required) is the absolute path to the directory containing the student's code

    responsefile (optional) is the path to the response file. The response file is assigned to stdin in the student's
        program.

    logfilepath (optional) is the path to the location where the log file will be created. The logfile will have the
        have the same name as the assignmentDir but with a .log extension.

    timeout (optional) The instructor can specify the maximum time a student's program can run in seconds.
        The default value is 10 seconds.

    pgmArgs (optional) are args that can be passed to a student's program.
        The default value is None

    pgmKwArgs (optional) are kwargs that can be passed to a student's program.
        The default value is None

    numReturns (optional) is a value that represents the number of lines the student's program will write to stdout.
        If a student's program does not use all of the responsefile, an error will appear in the log file.

"""

import argparse
import subprocess
import json
import sys, os
from time import sleep


def serialize(data, key=None):
    if key is None:
        return json.dumps(data)
    else:
        return key+":"+json.dumps(data)


def deserialize(data, key=None):
    if key is None:
        return json.loads(data)
    else:
        tmp = data.split(':',1)
        if tmp[0] == key and len(tmp) == 2:
            return json.loads(tmp[1])


def isDir(path):
    return os.path.isdir(path)

def isFullpath(path):
    return not os.path.isdir(path) and os.path.isabs(path)

def isRelativepath(path):
    return path[:2] == '.'+os.path.sep

def dirjoin(dir, fname):
    if not isFullpath(fname):
        return os.path.join(dir, fname)

    return fname

def getDirname(path):
    if os.path.isdir(path):
        return path

    tmp = os.path.split(path)
    if len(tmp[0]) == 0:
        return None
    else:
        return tmp[0]

def list2Dictionary(kwlist):
    return dict(zip(*[iter(kwlist)]*2))

def generateOutputfile(codefile, outputfile, responsefile = None, pgmArgs = None):
    if pgmArgs is None:
        cmd = [sys.executable, codefile]
    else:
        cmd = [sys.executable, codefile, *pgmArgs]

    try:
        if responsefile is None:
            result = subprocess.run(cmd, shell=True, capture_output=True, encoding='utf8')
        else:
            responsefp = open(responsefile, 'r')
            result = subprocess.run(cmd, shell=True, stdin=responsefp, capture_output=True, encoding='utf8')
            responsefp.close()

        outputfilefp = open(outputfile, 'w')
        outputfilefp.write(result.stdout)
        outputfilefp.close()
        return True
    except Exception as e:
        print(str(e))
        return False


def runCode(codefile, responsefp, logfilefp, timeout, numOutputlines, pgmArgs = None):
    if pgmArgs is None:
        cmd = [sys.executable, codefile]
    else:
        cmd = [sys.executable, codefile, *pgmArgs]

    try:
        if responsefp is None:
            result = subprocess.run(cmd, shell=True, timeout=timeout, capture_output=True, encoding='utf8')
        else:
            result = subprocess.run(cmd, shell=True, stdin=responsefp, timeout=timeout, capture_output=True, encoding='utf8')

        logfilefp.write('STDOUT:\n')
        logfilefp.write(result.stdout)

        logfilefp.write('\nSTDERR:\n')
        logfilefp.write(result.stderr)
        if numOutputlines is not None:
            N = len(result.stdout.split("\n"))
            # print(f"numReturns:{N}<{numReturns}={N<numReturns}")
            if N < numOutputlines and result.returncode == 0:
                logfilefp.write("Program Error: Not All Input Utilized\n\n")
            else:
                logfilefp.write('\n\n')
    except subprocess.TimeoutExpired as e:
        logfilefp.write('STDOUT:\n')
        logfilefp.write(str(e.stdout))

        logfilefp.write('\nSTDERR:\n')
        logfilefp.write("Execution Time Expired Error\n")
        logfilefp.write(str(e)+"\n\n")
    except Exception as e:
        logfilefp.write(str(e)+"\n\n")



def join(dir, file):
    return(dir + os.path.sep + file)


def studentCodeEvaluator(assignmentDir, logfilepath=None, responseFile=None, timeout=None, numOutputlines = None, pgmArgs=None, pgmKwArgs=None):
    print(f"assignmentDir:{assignmentDir}")
    basename = os.path.basename(assignmentDir)
    print(f"logfilepath:{join(logfilepath,basename+'.log')}")
    print(f"responsefile:{responseFile}")
    print(f"responsefilepath:{join(assignmentDir, responseFile)}")
    print(f"timeout:{timeout}")

    logname = os.path.basename(assignmentDir) + '.log'
    logfilepath = os.path.join(logfilepath,logname)
    logfilerp = open(logfilepath, 'w', encoding='utf8')

    if pgmArgs is not None and pgmKwArgs is not None:
        pgmArgs = pgmArgs + (pgmKwArgs,)
    elif pgmKwArgs is not None:
        pgmArgs = pgmKwArgs

    rfp = None
    if responseFile is not None:
        if not isFullpath(responseFile):
            if not isRelativepath(responseFile):
                responsefilepath = os.path.join(assignmentDir, responseFile)
            else:
                responsefilepath = responseFile
        else:
            responsefilepath = responseFile

        rfp = open(responsefilepath,'r', encoding='utf8')

    files = next(os.walk(assignmentDir))[2]
    for file in files:
        ext = os.path.splitext(file)[1]
        if '.py' in ext:
            print(file)
            logfilerp.write("="*len(file))
            logfilerp.write('\n'+file+'\n')
            codepath = os.path.join(assignmentDir, file)
            runCode(codepath, rfp, logfilerp, timeout, numOutputlines, pgmArgs)
            rfp.seek(0, 0)

    rfp.close()
    logfilerp.close()

def countCodefileInputs(codefile):
    count = 0
    with open(codefile, 'r') as fp:
            for line in fp:
                if line[0] == '#' or len(line) == 0:
                    continue

                if 'input(' in line:
                    count += 1


    return count


def countOutputfileNreturns(outputfile):
    count = 0
    with open(outputfile, 'r') as fp:
        for line in fp:
            line = line.strip()
            if len(line) == 0 or line[0] == '#':
                continue

            if len(line) > 1:
                count += 1

    return count


def countOutputLines(codefile, outputfile, responseFile=None, pgmArgs=None, pgmKwArgs=None):
    """
    This function executes the codefile and counts and reports the number of lines of text
    the codefile writes to stdout.If present the content of the responseFile replaces the codefile's stdin
    :param str codefile: The codefile to be run
    :param str outputfile: The file to write the stdout to
    :param str responseFile: A file of responses that will replace the codefile's stdin
    :param tuple pgmArgs: A list of arguments
    :param str pgmKwArgs: a json kwargs string
    :return: None
    """
    print("Computing nReturns:")
    print(f"\tcodefile:{os.path.basename(codefile)} pgmArgs:{pgmArgs}")
    print(f"\toutputfile:{outputfile}")
    print(f"\tresponsefile:{responseFile}")

    nReturns = 0
    dir = getDirname(codefile)

    # convert simple filename to absolute filename
    if not isFullpath(outputfile):
        if dir is not None:
            outputfile = dirjoin(dir,outputfile)

    # convert simple filename to absolute filename
    if not isFullpath(responseFile):
        if dir is not None:
            responseFile = dirjoin(dir, responseFile)

    if pgmArgs is not None and pgmKwArgs is not None:
        pgmArgs = pgmArgs + (pgmKwArgs,)
    elif pgmKwArgs is not None:
        pgmArgs = pgmKwArgs

    if generateOutputfile(codefile, outputfile, responseFile, pgmArgs):
        try:
            nReturns = countOutputfileNreturns(outputfile)
        except Exception as e:
            print(str(e))

    print(f"\tnReturns:{nReturns}")


def parseArgs():
    parser = argparse.ArgumentParser(
        prog='studentcodeEvaluator',
        formatter_class=argparse.RawTextHelpFormatter,
        description="Student Code Evaluator")
    parser.add_argument(
        'assignmentDir',
        type=str,
        metavar='Assignment Directory',
        help="(Required) Path to assignment directory that contains the student's code.\n \n"
    )
    parser.add_argument(
        '-a',
        metavar='Program Args',
        dest='pgmArgs',
        nargs='*',
        default=None,
        help="These are args to be passed to the student's program.\n \n"
    )
    parser.add_argument(
        '-kwa',
        metavar='Program kwArgs',
        dest='pgmKwArgs',
        nargs='*',
        default=None,
        help="These are kwargs to be passed to the student's program.\n \n"
    )
    parser.add_argument(
        '-c',
        metavar='OutputLines File',
        dest='outputLinesfile',
        default = None,
        help="A utility to count the number of lines of text the student's code outputs to stdout.\n"\
            "This option requires the following parameters:\n"\
            "\t(1) The codefile to be run.\n"
            "\t(2) The output file.\n"
            "\t(3) The response file (optional).\n"
            "\t(4) pgmArgs to be passed to the codefile (optional).\n"
            "The utility will print the number of lines of text generated by the codefile.\n \n"
    )
    parser.add_argument(
        '-l',
        metavar='Logfile Path',
        dest='logfilepath',
        default='assignmentDir',
        help='The location of the teacher log file\n'\
            'The default location is the assignmentDir.\n \n'
    )
    parser.add_argument(
        '-n',
        type = int,
        metavar='Number of Output Lines',
        dest='numOutputlines',
        default = None,
        help="The number of lines of output stdout should contain.\n"\
            "This parameter is used to determine if a student's code reads and\n"\
            "prints all of the input from the responsefile.\n \n"
    )
    parser.add_argument(
        '-r',
        metavar='Response File',
        dest='responseFile',
        default=None,
        help="The response file provides input to stdin of the student's program.\n \n"
    )
    parser.add_argument(
        '-t',
        metavar='Timeout',
        dest='timeout',
        type=int,
        default=10,
        help="The timeout is the maximum amount of time in seconds the students\n"\
            "program will be allowed to run. If a student's code runs longer than this value,\n"\
            "it will generate a time expired error.\n"\
            "The default value for timeout is %(default)s seconds\n \n"
    )
    parser.add_argument(
        '--version',
        action='version',
        version='%(prog)s 1.0'
    )
    args = parser.parse_args()
    if args.logfilepath == 'assignmentDir':
        args.logfilepath = args.assignmentDir

    if args.pgmKwArgs is not None:
        args.pgmKwArgs = serialize(list2Dictionary(args.pgmKwArgs),'kwargs')

    if args.pgmArgs is not None:
        args.pgmArgs = tuple(args.pgmArgs)

    return args

def main():
    args=parseArgs()
    if args.outputLinesfile is not None:
        countOutputLines(args.assignmentDir, args.outputLinesfile, responseFile=args.responseFile, pgmArgs=args.pgmArgs, pgmKwArgs=args.pgmKwArgs)
    else:
        kwargs = vars(args)
        del kwargs['outputLinesfile']
        studentCodeEvaluator(**kwargs)

    print("Finished...")

if __name__ == '__main__':
    main()