-
Notifications
You must be signed in to change notification settings - Fork 54
Expand file tree
/
Copy pathapi.py
More file actions
241 lines (200 loc) · 8.02 KB
/
Copy pathapi.py
File metadata and controls
241 lines (200 loc) · 8.02 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
240
241
"""
PC-BASIC - api.py
Session API
(c) 2013--2026 Rob Hagemans
This file is released under the GNU GPL version 3 or later.
"""
import os
import io
from ..compat import text_type
from .base import error
from .devices import NameWrapper
from . import implementation
from . import state
from ..data import read_codepage as codepage
from ..data import read_fonts as font
from .values import TYPE_TO_CLASS as SIGILS
class Session(object):
"""Public API to BASIC session."""
def __init__(self, **kwargs):
"""Set up session object."""
self._kwargs = kwargs
self._impl = None
def __enter__(self):
"""Context guard."""
return self
def __exit__(self, ex_type, ex_val, tb):
"""Context guard."""
self.close()
# catch Exit and Break events
if ex_type in (error.Exit, error.Break):
return True
def __getstate__(self):
"""Pickle the session."""
pickle_dict = self.__dict__.copy()
return pickle_dict
def __setstate__(self, pickle_dict):
"""Unpickle and resume the session."""
self.__dict__.update(pickle_dict)
def start(self):
"""Start the session."""
if not self._impl:
self._impl = implementation.Implementation(**self._kwargs)
return True
return False
def attach(self, interface=None):
"""Attach interface to interpreter session."""
self.start()
self._impl.attach_interface(interface)
return self
def bind_file(self, file_name_or_object, name=None, create=False):
"""Bind a native file name or Python stream to a BASIC file name."""
self.start()
# if a file name, resolve
if (
not isinstance(file_name_or_object, (bytes, text_type))
or os.path.isfile(file_name_or_object)
):
# if it's an object or the file name exists, use it
return self._impl.files.get_device(b'@:').bind(file_name_or_object, name)
elif create and (
not os.path.dirname(file_name_or_object) or
os.path.isdir(os.path.dirname(file_name_or_object))
):
# if it doesn't and we're allowed to create and the directory exists, create new
return self._impl.files.get_device(b'@:').bind(file_name_or_object, name)
# not resolved, try to use/create as internal name
return NameWrapper(self._impl.codepage, file_name_or_object)
def execute(self, command, as_type=None):
"""Execute a BASIC statement."""
self.start()
if as_type is None:
as_type = type(command)
output = io.BytesIO() if as_type == bytes else io.StringIO()
with self._impl.io_streams.activate():
self.add_pipes(output_streams=output)
for cmd in command.splitlines():
if isinstance(cmd, text_type):
cmd = self._impl.codepage.unicode_to_bytes(cmd)
self._impl.execute(cmd)
self.remove_pipes(output_streams=output)
return output.getvalue()
def add_pipes(self, input_streams=None, output_streams=None):
"""Add input/output pipes to session."""
self.start()
self._impl.io_streams.add_pipes(input_streams, output_streams)
def remove_pipes(self, input_streams=None, output_streams=None):
"""Remove input/output pipes from session."""
self.start()
self._impl.io_streams.remove_pipes(input_streams, output_streams)
def evaluate(self, expression):
"""Evaluate a BASIC expression."""
self.start()
with self._impl.io_streams.activate():
if isinstance(expression, text_type):
expression = self._impl.codepage.unicode_to_bytes(expression)
return self._impl.evaluate(expression)
def set_variable(self, name, value):
"""Set a variable in memory."""
self.start()
if isinstance(name, text_type):
name = name.encode('ascii')
name = name.upper()
if name.split(b'(')[0][-1:] not in SIGILS:
raise ValueError('Sigil must be explicit')
self._impl.set_variable(name, value)
def get_variable(self, name, as_type=None):
"""Get a variable in memory."""
self.start()
if isinstance(name, text_type):
name = name.encode('ascii')
if name.split(b'(')[0][-1:] not in SIGILS:
raise ValueError('Sigil must be explicit')
return self._impl.get_variable(name, as_type)
def convert(self, value, to_type):
"""Convert a Python value to another type, consistent with BASIC rules."""
self.start()
return self._impl.get_converter(type(value), to_type)(value)
def press_keys(self, keys):
"""Insert keypresses."""
self.start()
self._impl.keyboard.inject_keystrokes(keys)
def get_chars(self, as_type=bytes):
"""Get currently displayed characters, as tuple of list of bytes / unicode."""
self.start()
return self._impl.text_screen.get_chars(as_type=as_type)
def get_pixels(self):
"""Get currently displayed pixels, as tuple of tuples of int attributes."""
self.start()
return self._impl.display.vpage.pixels[:, :].to_rows()
def greet(self):
"""Emit the interpreter greeting and show the key bar."""
self.start()
self._impl.execute(implementation.GREETING)
def interact(self):
"""Interactive interpreter session."""
self.start()
with self._impl.io_streams.activate():
self._impl.interact()
def suspend(self, session_filename):
"""Save session object to file."""
state.save_session(self, session_filename)
@classmethod
def resume(self, session_filename):
"""Load new session object from file."""
return state.load_session(session_filename)
def close(self):
"""Close the session."""
if self._impl:
self._impl.close()
@property
def info(self):
"""Get a session information object."""
self.start()
return SessionInfo(self)
def set_hook(self, step_function):
"""Set function to be called on interpreter step."""
self.start()
self._impl.interpreter.step = step_function
class SessionInfo(object):
"""Retrieve information about current session."""
def __init__(self, session):
"""Initialise the SessionInfo object."""
self._session = session
self._impl = session._impl
def repr_scalars(self):
"""Get a representation of all scalars."""
return repr(self._impl.scalars)
def repr_arrays(self):
"""Get a representation of all arrays."""
return repr(self._impl.scalars)
def repr_strings(self):
"""Get a representation of string space."""
return repr(self._impl.strings)
def repr_text_screen(self):
"""Get a representation of the text screen."""
return repr(self._impl.display.text_screen)
def repr_program(self):
"""Get a marked-up hex dump of the program."""
return repr(self._impl.program)
def get_current_code(self, as_type=bytes):
"""Obtain statement being executed."""
if self._impl.interpreter.run_mode:
codestream = self._impl.program.bytecode
bytepos = codestream.tell()
from_line = self._impl.program.get_line_number(bytepos-1)
try:
codestream.seek(self._impl.program.line_numbers[from_line]+1)
_, output, _ = self._impl.lister.detokenise_line(codestream)
code_line = bytes(output)
except KeyError:
code_line = b''
else:
codestream = self._impl.interpreter.direct_line
bytepos = codestream.tell()
codestream.seek(0)
code_line = bytes(
self._impl.lister.detokenise_compound_statement(codestream)[0]
)
codestream.seek(bytepos)
return self._session.convert(code_line, as_type)