-
Notifications
You must be signed in to change notification settings - Fork 2
/
command.py
65 lines (52 loc) · 2.38 KB
/
command.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import subprocess
import threading
import time
import os
_isWindows = (os.name == "nt")
def makeTupleIfNot(what):
""" If the given argument is a tuple or a list, returns it without any modifications. Otherwise creates a tuple with the single element. """
if (type(what) == tuple or type(what) == list):
return what
else:
return (what,)
# ---------------------------------------------------------------------------------------------------------------------
# Command
# ---------------------------------------------------------------------------------------------------------------------
class Command:
""" Class used to run external programs. When initialized, the command name is provided. The command can then be launched multiple times. An input can be specified, in which case this is sent to the external application to its stdin. For each execution, different arguments can also be specified. """
STD_OUT = 0
STD_ERR = 1
RETURN_CODE = 2
EXEC_TIME = 3
SUCCESS = 0
FAILURE = 1
TIMEOUT = "TIMEOUT"
class Result:
""" Possible results of the execution - SUCCESS and FAILURE known from C and the TIMEOUT marker for when the application execution timeouts. """
def __init__(self, command):
""" Creates the command instance for the given command. """
self.command = command
def run(self, args = (), input = None, timeout = 10):
""" Executes the command with given arguments, possible input, and a timeout in seconds. Returns tuple consisting of standard output and error strings of the command, the return code from Command.Result enum and the duration of the execution. """
def runner():
cmd = self.command
if (len(args) != 0):
cmd = cmd + " " + " ".join(makeTupleIfNot(args))
self.time = time.time()
self.proc = subprocess.Popen(cmd, shell=not _isWindows, stdout = subprocess.PIPE, stderr = subprocess.PIPE, stdin = None if input == None else subprocess.PIPE)
self.procOutput = self.proc.communicate(None if input == None else bytes(input,"UTF-8"))
self.time = time.time() - self.time
self.time = "TIMEOUT"
thread = threading.Thread(target = runner)
thread.start()
thread.join(timeout)
if (thread.is_alive()):
self.proc.terminate()
stdout = ""
stderr = ""
rc = Command.Result.TIMEOUT
else:
rc = self.proc.returncode
stdout = self.procOutput[0].decode("UTF-8")
stderr = self.procOutput[1].decode("UTF-8")
return (stdout, stderr, rc, self.time)