-
Notifications
You must be signed in to change notification settings - Fork 52
/
__init__.py
160 lines (117 loc) · 5.05 KB
/
__init__.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
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
from .generated import Root, RootImports, Err, imports
from .generated.imports import streams
from .generated.imports.types import Descriptor, Filesize, ErrorCode, DescriptorType
from .generated.imports.terminal_input import TerminalInput
from .generated.imports.terminal_output import TerminalOutput
from .generated.imports import terminal_stderr
from .generated import types as core_types
from typing import Mapping, Tuple, List, Optional
import sys
import os
from wasmtime import Store
class WasiRandom(imports.HostRandom):
def get_random_bytes(self, len: int) -> bytes:
return os.urandom(len)
class WasiStdin(imports.HostStdin):
def get_stdin(self) -> streams.InputStream:
return 0
class WasiStdout(imports.HostStdout):
def get_stdout(self) -> streams.OutputStream:
return 1
class WasiStderr(imports.HostStderr):
def get_stderr(self) -> streams.OutputStream:
return 2
class WasiPreopens(imports.HostPreopens):
def get_directories(self) -> List[Tuple[Descriptor, str]]:
return []
class WasiStreams(imports.HostStreams):
def drop_input_stream(self, this: streams.InputStream) -> None:
return None
def write(self, this: streams.OutputStream, buf: bytes) -> core_types.Result[Tuple[int, streams.StreamStatus], None]:
if this == 1:
sys.stdout.buffer.write(buf)
elif this == 2:
sys.stderr.buffer.write(buf)
else:
raise NotImplementedError
return core_types.Ok((len(buf), streams.StreamStatus.OPEN))
def blocking_write(self, this: streams.OutputStream, buf: bytes) -> core_types.Result[Tuple[int, streams.StreamStatus], None]:
return self.write(this, buf)
def drop_output_stream(self, this: streams.OutputStream) -> None:
return None
class WasiEnvironment(imports.HostEnvironment):
def get_environment(self) -> List[Tuple[str, str]]:
return []
class WasiTypes(imports.HostTypes):
def write_via_stream(self, this: Descriptor, offset: Filesize) -> core_types.Result[streams.OutputStream, ErrorCode]:
raise NotImplementedError
def append_via_stream(self, this: Descriptor) -> core_types.Result[streams.OutputStream, ErrorCode]:
raise NotImplementedError
def get_type(self, this: Descriptor) -> core_types.Result[DescriptorType, ErrorCode]:
raise NotImplementedError
def drop_descriptor(self, this: Descriptor) -> None:
raise NotImplementedError
class WasiExit(imports.HostExit):
def exit(self, status: core_types.Result[None, None]) -> None:
raise NotImplementedError
class WasiTerminalInput(imports.HostTerminalInput):
def drop_terminal_input(self, this: TerminalInput) -> None:
pass
class WasiTerminalOutput(imports.HostTerminalOutput):
def drop_terminal_output(self, this: TerminalOutput) -> None:
pass
class WasiTerminalStdin(imports.HostTerminalStdin):
def get_terminal_stdin(self) -> Optional[TerminalInput]:
if sys.stdin.isatty():
return sys.stdin.fileno()
return None
class WasiTerminalStdout(imports.HostTerminalStdout):
def get_terminal_stdout(self) -> Optional[TerminalOutput]:
if sys.stdout.isatty():
return sys.stdout.fileno()
return None
class WasiTerminalStderr(imports.HostTerminalStderr):
def get_terminal_stderr(self) -> Optional[terminal_stderr.TerminalOutput]:
if sys.stderr.isatty():
return sys.stderr.fileno()
return None
root = None
store = None
def init() -> Tuple[Root, Store]:
global store
global root
if root is None:
store = Store()
root = Root(store, RootImports(WasiStreams(),
WasiTypes(),
WasiPreopens(),
WasiRandom(),
WasiEnvironment(),
WasiExit(),
WasiStdin(),
WasiStdout(),
WasiStderr(),
WasiTerminalInput(),
WasiTerminalOutput(),
WasiTerminalStdin(),
WasiTerminalStdout(),
WasiTerminalStderr()))
return root, store
# Generates Python bindings for the given component.
#
# The `name` provided is used as the name of the `component` binary provided.
# The `component` argument is expected to be the binary representation of a
# component.
#
# This function returns a mapping of filename to contents of files that are
# generated to represent the Python bindings here.
def generate(name: str, component: bytes) -> Mapping[str, bytes]:
root, store = init()
result = root.generate(store, name, component)
if isinstance(result, Err):
raise RuntimeError(result.value)
ret = {}
for name, contents in result.value:
ret[name] = contents
return ret
__all__ = ['generate']