forked from maciejkula/spotlight
-
Notifications
You must be signed in to change notification settings - Fork 0
/
utils.py
75 lines (60 loc) · 2.11 KB
/
utils.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
66
67
68
69
70
71
72
73
74
75
#### Those are the functions that make your life easier while coding. Enjoy :)
import subprocess
import os
import threading
def call_subprocess(command: str, params: list, outfile=None, chdir=None):
# When we want to pipe the result to a text file, then we have to use the outfile option.
# If the program asks you to specify the output with -o etc. then leave the outfile param None
if outfile:
stdout_buffer = open(outfile, "wb", buffering=0)
else:
stdout_buffer = subprocess.PIPE
popen_args = dict(
args=[command] + params,
preexec_fn=os.setsid,
stdin=subprocess.DEVNULL,
stdout=stdout_buffer,
stderr=subprocess.PIPE,
bufsize=0,
cwd=chdir,
)
process = subprocess.Popen(**popen_args)
stdout, stderr = process.communicate()
return_code = process.returncode
if return_code != 0:
full_command = " ".join(popen_args['args'])
raise Exception(full_command, stdout, stderr)
retstdout = stdout.decode() if stdout is not None else None
return return_code, retstdout
def load_json(json_file):
import json
json_file = str(json_file)
with open(json_file) as fh:
return json.load(fh)
def dump_json(jdict, jfile):
import json
with open(jfile, "w") as jfh:
json.dump(jdict, jfh, indent=4)
class CustomThreading(threading.Thread):
"""
This edited threading class is used to catch exceptions
in the child thread and raise it in the main thread
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.keyword_args = kwargs
def run(self):
# Variable that stores the exception, if raised by someFunction
self.exc = None
try:
## starts running the targeted process
self.keyword_args["target"]()
except BaseException as e:
self.exc = e
def join(self):
threading.Thread.join(self)
# Since join() returns in caller thread
# we re-raise the caught exception
# if any was caught
if self.exc:
raise self.exc