Coverage for wasmtime/bindgen/__init__.py: 67%
93 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-02-20 16:25 +0000
« prev ^ index » next coverage.py v7.6.12, created at 2025-02-20 16:25 +0000
1from .generated import Root, RootImports, Err, imports
2from .generated.imports import streams
3from .generated.imports.types import Descriptor, Filesize, ErrorCode, DescriptorType
4from .generated.imports.terminal_input import TerminalInput
5from .generated.imports.terminal_output import TerminalOutput
6from .generated.imports import terminal_stderr
7from .generated import types as core_types
8from typing import Mapping, Tuple, List, Optional
10import sys
11import os
12from wasmtime import Store
15class WasiRandom(imports.HostRandom):
16 def get_random_bytes(self, len: int) -> bytes:
17 return os.urandom(len)
20class WasiStdin(imports.HostStdin):
21 def get_stdin(self) -> streams.InputStream:
22 return 0
25class WasiStdout(imports.HostStdout):
26 def get_stdout(self) -> streams.OutputStream:
27 return 1
30class WasiStderr(imports.HostStderr):
31 def get_stderr(self) -> streams.OutputStream:
32 return 2
35class WasiPreopens(imports.HostPreopens):
36 def get_directories(self) -> List[Tuple[Descriptor, str]]:
37 return []
40class WasiStreams(imports.HostStreams):
41 def drop_input_stream(self, this: streams.InputStream) -> None:
42 return None
44 def write(self, this: streams.OutputStream, buf: bytes) -> core_types.Result[Tuple[int, streams.StreamStatus], None]:
45 if this == 1:
46 sys.stdout.buffer.write(buf)
47 elif this == 2:
48 sys.stderr.buffer.write(buf)
49 else:
50 raise NotImplementedError
51 return core_types.Ok((len(buf), streams.StreamStatus.OPEN))
53 def blocking_write(self, this: streams.OutputStream, buf: bytes) -> core_types.Result[Tuple[int, streams.StreamStatus], None]:
54 return self.write(this, buf)
56 def drop_output_stream(self, this: streams.OutputStream) -> None:
57 return None
60class WasiEnvironment(imports.HostEnvironment):
61 def get_environment(self) -> List[Tuple[str, str]]:
62 return []
65class WasiTypes(imports.HostTypes):
66 def write_via_stream(self, this: Descriptor, offset: Filesize) -> core_types.Result[streams.OutputStream, ErrorCode]:
67 raise NotImplementedError
69 def append_via_stream(self, this: Descriptor) -> core_types.Result[streams.OutputStream, ErrorCode]:
70 raise NotImplementedError
72 def get_type(self, this: Descriptor) -> core_types.Result[DescriptorType, ErrorCode]:
73 raise NotImplementedError
75 def drop_descriptor(self, this: Descriptor) -> None:
76 raise NotImplementedError
79class WasiExit(imports.HostExit):
80 def exit(self, status: core_types.Result[None, None]) -> None:
81 raise NotImplementedError
84class WasiTerminalInput(imports.HostTerminalInput):
85 def drop_terminal_input(self, this: TerminalInput) -> None:
86 pass
89class WasiTerminalOutput(imports.HostTerminalOutput):
90 def drop_terminal_output(self, this: TerminalOutput) -> None:
91 pass
94class WasiTerminalStdin(imports.HostTerminalStdin):
95 def get_terminal_stdin(self) -> Optional[TerminalInput]:
96 if sys.stdin.isatty():
97 return sys.stdin.fileno()
98 return None
101class WasiTerminalStdout(imports.HostTerminalStdout):
102 def get_terminal_stdout(self) -> Optional[TerminalOutput]:
103 if sys.stdout.isatty():
104 return sys.stdout.fileno()
105 return None
108class WasiTerminalStderr(imports.HostTerminalStderr):
109 def get_terminal_stderr(self) -> Optional[terminal_stderr.TerminalOutput]:
110 if sys.stderr.isatty():
111 return sys.stderr.fileno()
112 return None
115root = None
116store = None
119def init() -> Tuple[Root, Store]:
120 global store
121 global root
122 if root is None:
123 store = Store()
124 root = Root(store, RootImports(WasiStreams(),
125 WasiTypes(),
126 WasiPreopens(),
127 WasiRandom(),
128 WasiEnvironment(),
129 WasiExit(),
130 WasiStdin(),
131 WasiStdout(),
132 WasiStderr(),
133 WasiTerminalInput(),
134 WasiTerminalOutput(),
135 WasiTerminalStdin(),
136 WasiTerminalStdout(),
137 WasiTerminalStderr()))
138 return root, store
141# Generates Python bindings for the given component.
142#
143# The `name` provided is used as the name of the `component` binary provided.
144# The `component` argument is expected to be the binary representation of a
145# component.
146#
147# This function returns a mapping of filename to contents of files that are
148# generated to represent the Python bindings here.
149def generate(name: str, component: bytes) -> Mapping[str, bytes]:
150 root, store = init()
151 result = root.generate(store, name, component)
152 if isinstance(result, Err):
153 raise RuntimeError(result.value)
154 ret = {}
155 for name, contents in result.value:
156 ret[name] = contents
157 return ret
160__all__ = ['generate']