Coverage for wasmtime/bindgen/__init__.py: 67%

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 

9 

10import sys 

11import os 

12from wasmtime import Store 

13 

14 

15class WasiRandom(imports.HostRandom): 

16 def get_random_bytes(self, len: int) -> bytes: 

17 return os.urandom(len) 

18 

19 

20class WasiStdin(imports.HostStdin): 

21 def get_stdin(self) -> streams.InputStream: 

22 return 0 

23 

24 

25class WasiStdout(imports.HostStdout): 

26 def get_stdout(self) -> streams.OutputStream: 

27 return 1 

28 

29 

30class WasiStderr(imports.HostStderr): 

31 def get_stderr(self) -> streams.OutputStream: 

32 return 2 

33 

34 

35class WasiPreopens(imports.HostPreopens): 

36 def get_directories(self) -> List[Tuple[Descriptor, str]]: 

37 return [] 

38 

39 

40class WasiStreams(imports.HostStreams): 

41 def drop_input_stream(self, this: streams.InputStream) -> None: 

42 return None 

43 

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)) 

52 

53 def blocking_write(self, this: streams.OutputStream, buf: bytes) -> core_types.Result[Tuple[int, streams.StreamStatus], None]: 

54 return self.write(this, buf) 

55 

56 def drop_output_stream(self, this: streams.OutputStream) -> None: 

57 return None 

58 

59 

60class WasiEnvironment(imports.HostEnvironment): 

61 def get_environment(self) -> List[Tuple[str, str]]: 

62 return [] 

63 

64 

65class WasiTypes(imports.HostTypes): 

66 def write_via_stream(self, this: Descriptor, offset: Filesize) -> core_types.Result[streams.OutputStream, ErrorCode]: 

67 raise NotImplementedError 

68 

69 def append_via_stream(self, this: Descriptor) -> core_types.Result[streams.OutputStream, ErrorCode]: 

70 raise NotImplementedError 

71 

72 def get_type(self, this: Descriptor) -> core_types.Result[DescriptorType, ErrorCode]: 

73 raise NotImplementedError 

74 

75 def drop_descriptor(self, this: Descriptor) -> None: 

76 raise NotImplementedError 

77 

78 

79class WasiExit(imports.HostExit): 

80 def exit(self, status: core_types.Result[None, None]) -> None: 

81 raise NotImplementedError 

82 

83 

84class WasiTerminalInput(imports.HostTerminalInput): 

85 def drop_terminal_input(self, this: TerminalInput) -> None: 

86 pass 

87 

88 

89class WasiTerminalOutput(imports.HostTerminalOutput): 

90 def drop_terminal_output(self, this: TerminalOutput) -> None: 

91 pass 

92 

93 

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 

99 

100 

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 

106 

107 

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 

113 

114 

115root = None 

116store = None 

117 

118 

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 

139 

140 

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 

158 

159 

160__all__ = ['generate']