-
Notifications
You must be signed in to change notification settings - Fork 6
/
backward.py
130 lines (104 loc) · 3.17 KB
/
backward.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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
import random
import time
from sympy import Symbol, diff, simplify
from infix_prefix import prefix_to_infix, infix_to_prefix
from random_trees import random_binary_trees, traverse_unary_binary_prefix
_FINISH = False
x = Symbol('x', real=True) # TODO
def simplify_timeout(expr):
from multiprocessing import Process, Manager
result = None
def f(d, expr):
from sympy import S; S.Half
try:
d['result'] = simplify(expr)
except:
d['result'] = "Error"
with Manager() as manager:
d = manager.dict()
p = Process(target=f, args=(d, expr))
p.start()
p.join(timeout=3)
if p.is_alive():
p.terminate()
else:
result = d["result"]
if result==None:
return expr
elif str(type(result))=="<class 'sympy.calculus.util.AccumulationBounds'>":
return "Error"
else:
return result
def parse_expr_timeout(string):
from multiprocessing import Process, Manager
from sympy.parsing.sympy_parser import parse_expr
expr = None
def f(d, string):
from sympy import S; S.Half
try:
d['expr'] = parse_expr(string, local_dict={'x': x}, evaluate=False)
except:
return None
with Manager() as manager:
d = manager.dict()
p = Process(target=f, args=(d, string))
p.start()
p.join(timeout=3)
if p.is_alive():
p.terminate()
else:
expr = d["expr"]
return expr
def generate_single_sequence():
n = random.randint(1, 15)
root = random_binary_trees(n)
result = traverse_unary_binary_prefix(root)
if 'x' not in result:
return None
result_infix = prefix_to_infix(result)
result_expr = " ".join(result_infix)
# convert to sympy expression
result_expr = parse_expr_timeout(result_expr)
if result_expr is None:
return None
# simplification
result_simp = simplify_timeout(result_expr)
if result_simp == "Error":
return None
# back to prefix
result_simp_prefix = infix_to_prefix(result_simp)
if not result_simp_prefix:
return None
# generate target
try:
expr_diff = diff(result_simp)
except ValueError:
return None
expression = simplify_timeout(expr_diff)
if expression == "Error":
return None
expression_prefix = infix_to_prefix(expression)
if not expression_prefix:
return None
expression_prefix = " ".join(expression_prefix)
result_simp_prefix = " ".join(result_simp_prefix)
return expression_prefix, result_simp_prefix
def generate_bwd(num):
while True:
if _FINISH:
break
sequence = []
start = time.time()
i = 1
while i<=num:
seq = generate_single_sequence()
if seq is not None:
expression_prefix, result_simp_prefix = seq
sequence.append(expression_prefix + "\t" + result_simp_prefix + "\n")
else:
continue
i += 1
end = time.time()
#print(end - start)
print('process finished')
return sequence