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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
|
import ast
from contextvars import ContextVar, Token
from copy import deepcopy
from typing import List, Tuple, Optional, Union, MutableMapping, Any, Mapping
from public import public
from .formula import Formula
from .mod import Mod
from .op import CodeOp
from .point import Point
@public
class OpResult(object):
parents: Tuple
op: ast.operator
name: str
value: Mod
def __init__(self, name: str, value: Mod, op: ast.operator, *parents: Any):
self.parents = tuple(parents)
self.name = name
self.value = value
self.op = op
def __str__(self):
return self.name
def __repr__(self):
char = ""
if isinstance(self.op, ast.Mult):
char = "*"
elif isinstance(self.op, ast.Add):
char = "+"
elif isinstance(self.op, ast.Sub):
char = "-"
elif isinstance(self.op, ast.Div):
char = "/"
parents = char.join(str(parent) for parent in self.parents)
return f"{self.name} = {parents}"
@public
class Action(object):
formula: Formula
input_points: List[Point]
inputs: MutableMapping[str, Mod]
intermediates: MutableMapping[str, Union[Mod, OpResult]]
roots: MutableMapping[str, OpResult]
output_points: List[Point]
def __init__(self, formula: Formula, *points: Point, **inputs: Mod):
self.formula = formula
self.input_points = list(points)
self.inputs = inputs
self.intermediates = {}
self.roots = {}
self.output_points = []
def add_operation(self, op: CodeOp, value: Mod):
parents = []
for parent in {*op.variables, *op.parameters}:
if parent in self.intermediates:
parents.append(self.intermediates[parent])
elif parent in self.inputs:
parents.append(self.inputs[parent])
self.intermediates[op.result] = OpResult(op.result, value, op.operator, *parents)
def add_result(self, point: Point, **outputs: Mod):
for k in outputs:
self.roots[k] = self.intermediates[k]
self.output_points.append(point)
@public
class Context(object):
def _log_action(self, formula: Formula, *points: Point, **inputs: Mod):
raise NotImplementedError
def _log_operation(self, op: CodeOp, value: Mod):
raise NotImplementedError
def _log_result(self, point: Point, **outputs: Mod):
raise NotImplementedError
def _execute(self, formula: Formula, *points: Point, **params: Mod) -> Tuple[Point, ...]:
if len(points) != formula.num_inputs:
raise ValueError
coords = {}
for i, point in enumerate(points):
if point.coordinate_model != formula.coordinate_model:
raise ValueError
for coord, value in point.coords.items():
coords[coord + str(i + 1)] = value
locals = {**coords, **params}
self._log_action(formula, *points, **locals)
for op in formula.code:
op_result = op(**locals)
self._log_operation(op, op_result)
locals[op.result] = op_result
result = []
for i in range(formula.num_outputs):
ind = str(i + formula.output_index)
resulting = {}
full_resulting = {}
for variable in formula.coordinate_model.variables:
full_variable = variable + ind
if full_variable not in locals:
continue
resulting[variable] = locals[full_variable]
full_resulting[full_variable] = locals[full_variable]
point = Point(formula.coordinate_model, **resulting)
self._log_result(point, **full_resulting)
result.append(point)
return tuple(result)
def execute(self, formula: Formula, *points: Point, **params: Mod) -> Tuple[Point, ...]:
return self._execute(formula, *points, **params)
def __str__(self):
return self.__class__.__name__
@public
class NullContext(Context):
def _log_action(self, formula: Formula, *points: Point, **inputs: Mod):
pass
def _log_operation(self, op: CodeOp, value: Mod):
pass
def _log_result(self, point: Point, **outputs: Mod):
pass
@public
class DefaultContext(Context):
actions: List[Action]
def __init__(self):
self.actions = []
def _log_action(self, formula: Formula, *points: Point, **inputs: Mod):
self.actions.append(Action(formula, *points, **inputs))
def _log_operation(self, op: CodeOp, value: Mod):
self.actions[-1].add_operation(op, value)
def _log_result(self, point: Point, **outputs: Mod):
self.actions[-1].add_result(point, **outputs)
_actual_context: ContextVar[Context] = ContextVar("operational_context", default=NullContext())
class ContextManager(object):
def __init__(self, new_context):
self.new_context = deepcopy(new_context)
def __enter__(self) -> Context:
self.saved_context = getcontext()
setcontext(self.new_context)
return self.new_context
def __exit__(self, t, v, tb):
setcontext(self.saved_context)
@public
def getcontext():
return _actual_context.get()
@public
def setcontext(ctx: Context) -> Token:
return _actual_context.set(ctx)
@public
def local(ctx: Optional[Context] = None) -> ContextManager:
if ctx is None:
ctx = getcontext()
return ContextManager(ctx)
|
