Uh oh!
There was an error while loading.Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork252
Expand file tree
/
Copy pathcoderunner.py
More file actions
239 lines (188 loc) · 8.03 KB
/
coderunner.py
File metadata and controls
239 lines (188 loc) · 8.03 KB
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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
"""For running Python code that could interrupt itself at any time in order to,
for example, ask for a read on stdin, or a write on stdout
The CodeRunner spawns a greenlet to run code in, and that code can suspend its
own execution to ask the main greenlet to refresh the display or get
information.
Greenlets are basically threads that can explicitly switch control to each
other. You can replace the word "greenlet" with "thread" in these docs if that
makes more sense to you.
"""
importcode
importgreenlet
importlogging
importsignal
fromcurtsies.inputimportis_main_thread
logger=logging.getLogger(__name__)
classSigintHappened:
"""If this class is returned, a SIGINT happened while the main greenlet"""
classSystemExitFromCodeRunner(SystemExit):
"""If this class is returned, a SystemExit happened while in the code
greenlet"""
classRequestFromCodeRunner:
"""Message from the code runner"""
classWait(RequestFromCodeRunner):
"""Running code would like the main loop to run for a bit"""
classRefresh(RequestFromCodeRunner):
"""Running code would like the main loop to refresh the display"""
classDone(RequestFromCodeRunner):
"""Running code is done running"""
classUnfinished(RequestFromCodeRunner):
"""Source code wasn't executed because it wasn't fully formed"""
classSystemExitRequest(RequestFromCodeRunner):
"""Running code raised a SystemExit"""
def__init__(self,*args):
self.args=args
classCodeRunner:
"""Runs user code in an interpreter.
Running code requests a refresh by calling
request_from_main_context(force_refresh=True), which
suspends execution of the code and switches back to the main greenlet
After load_code() is called with the source code to be run,
the run_code() method should be called to start running the code.
The running code may request screen refreshes and user input
by calling request_from_main_context.
When this are called, the running source code cedes
control, and the current run_code() method call returns.
The return value of run_code() determines whether the method ought
to be called again to complete execution of the source code.
Once the screen refresh has occurred or the requested user input
has been gathered, run_code() should be called again, passing in any
requested user input. This continues until run_code returns Done.
The code greenlet is responsible for telling the main greenlet
what it wants returned in the next run_code call - CodeRunner
just passes whatever is passed in to run_code(for_code) to the
code greenlet
"""
def__init__(self,interp=None,request_refresh=lambda:None):
"""
interp is an interpreter object to use. By default a new one is
created.
request_refresh is a function that will be called each time the running
code asks for a refresh - to, for example, update the screen.
"""
self.interp=interporcode.InteractiveInterpreter()
self.source=None
self.main_context=greenlet.getcurrent()
self.code_context=None
self.request_refresh=request_refresh
# waiting for response from main thread
self.code_is_waiting=False
# sigint happened while in main thread
self.sigint_happened_in_main_context=False
self.orig_sigint_handler=None
@property
defrunning(self):
"""Returns greenlet if code has been loaded greenlet has been
started"""
returnself.sourceandself.code_context
defload_code(self,source):
"""Prep code to be run"""
assertself.sourceisNone, (
"you shouldn't load code when some is ""already running"
)
self.source=source
self.code_context=None
def_unload_code(self):
"""Called when done running code"""
self.source=None
self.code_context=None
self.code_is_waiting=False
defrun_code(self,for_code=None):
"""Returns Truthy values if code finishes, False otherwise
if for_code is provided, send that value to the code greenlet
if source code is complete, returns "done"
if source code is incomplete, returns "unfinished"
"""
ifself.code_contextisNone:
assertself.sourceisnotNone
self.code_context=greenlet.greenlet(self._blocking_run_code)
ifis_main_thread():
self.orig_sigint_handler=signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT,self.sigint_handler)
request=self.code_context.switch()
else:
assertself.code_is_waiting
self.code_is_waiting=False
ifis_main_thread():
signal.signal(signal.SIGINT,self.sigint_handler)
ifself.sigint_happened_in_main_context:
self.sigint_happened_in_main_context=False
request=self.code_context.switch(SigintHappened)
else:
request=self.code_context.switch(for_code)
logger.debug("request received from code was %r",request)
ifnotisinstance(request,RequestFromCodeRunner):
raiseValueError(
"Not a valid value from code greenlet: %r"%request
)
ifisinstance(request, (Wait,Refresh)):
self.code_is_waiting=True
ifisinstance(request,Refresh):
self.request_refresh()
returnFalse
elifisinstance(request, (Done,Unfinished)):
self._unload_code()
ifis_main_thread():
signal.signal(signal.SIGINT,self.orig_sigint_handler)
self.orig_sigint_handler=None
returnrequest
elifisinstance(request,SystemExitRequest):
self._unload_code()
raiseSystemExitFromCodeRunner(request.args)
defsigint_handler(self,*args):
"""SIGINT handler to use while code is running or request being
fulfilled"""
ifgreenlet.getcurrent()isself.code_context:
logger.debug("sigint while running user code!")
raiseKeyboardInterrupt()
else:
logger.debug(
"sigint while fulfilling code request sigint handler "
"running!"
)
self.sigint_happened_in_main_context=True
def_blocking_run_code(self):
try:
unfinished=self.interp.runsource(self.source)
exceptSystemExitase:
returnSystemExitRequest(*e.args)
returnUnfinished()ifunfinishedelseDone()
defrequest_from_main_context(self,force_refresh=False):
"""Return the argument passed in to .run_code(for_code)
Nothing means calls to run_code must be... ???
"""
ifforce_refresh:
value=self.main_context.switch(Refresh())
else:
value=self.main_context.switch(Wait())
ifvalueisSigintHappened:
raiseKeyboardInterrupt()
returnvalue
classFakeOutput:
def__init__(self,coderunner,on_write,real_fileobj):
"""Fakes sys.stdout or sys.stderr
on_write should always take unicode
fileno should be the fileno that on_write will
output to (e.g. 1 for standard output).
"""
self.coderunner=coderunner
self.on_write=on_write
self._real_fileobj=real_fileobj
defwrite(self,s,*args,**kwargs):
self.on_write(s,*args,**kwargs)
returnself.coderunner.request_from_main_context(force_refresh=True)
# Some applications which use curses require that sys.stdout
# have a method called fileno. One example is pwntools. This
# is not a widespread issue, but is annoying.
deffileno(self):
returnself._real_fileobj.fileno()
defwritelines(self,l):
forsinl:
self.write(s)
defflush(self):
pass
defisatty(self):
returnTrue
@property
defencoding(self):
returnself._real_fileobj.encoding