97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
from typing import Callable, TextIO, Tuple, Union
|
|
|
|
from phasmplatform.common import valuetype
|
|
from phasmplatform.common.container import Container
|
|
from phasmplatform.common.methodcall import MethodCall, MethodCallError
|
|
from phasmplatform.common.router import MethodCallRouterInterface
|
|
from phasmplatform.common.value import Value
|
|
from phasmplatform.common.valuetype import ValueType
|
|
|
|
|
|
WasmValue = Union[None, int, float]
|
|
|
|
|
|
class RunnerInterface:
|
|
__slots__ = ('method_call_router', 'container', 'container_log', )
|
|
|
|
method_call_router: MethodCallRouterInterface
|
|
container: Container
|
|
container_log: Tuple[Callable[[str], None]] # Tuple for typing issues
|
|
|
|
def __init__(self, method_call_router: MethodCallRouterInterface, container: Container, container_log: Callable[[str], None]) -> None:
|
|
self.method_call_router = method_call_router
|
|
self.container = container
|
|
self.container_log = (container_log, )
|
|
|
|
def do_call(self, call: MethodCall) -> Union[Value, MethodCallError]:
|
|
"""
|
|
Executes the call on the current container
|
|
|
|
This method is responsible for calling the on_success or on_error method.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
|
|
class BaseRunner(RunnerInterface):
|
|
__slots__ = ('method_call_router', )
|
|
|
|
def alloc_bytes(self, data: bytes) -> int:
|
|
"""
|
|
Calls upon stdlib.types.__alloc_bytes__ to allocate a bytes object
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def read_bytes(self, ptr: int) -> bytes:
|
|
"""
|
|
Reads a byte object allocated by stdlib.types.__alloc_bytes__
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def value_to_wasm(self, val: Value) -> WasmValue:
|
|
if val.value_type is valuetype.none:
|
|
assert val.data is None # type hint
|
|
return None
|
|
|
|
if val.value_type is valuetype.bytes:
|
|
assert isinstance(val.data, bytes) # type hint
|
|
return self.alloc_bytes(val.data)
|
|
|
|
raise NotImplementedError(val)
|
|
|
|
def value_from_wasm(self, value_type: ValueType, val: WasmValue) -> Value:
|
|
if value_type is valuetype.none:
|
|
assert val is None # type hint
|
|
return Value(value_type, None)
|
|
|
|
if value_type is valuetype.u32:
|
|
assert isinstance(val, int) # type hint
|
|
return Value(value_type, val)
|
|
|
|
if value_type is valuetype.bytes:
|
|
assert isinstance(val, int) # type hint
|
|
return Value(value_type, self.read_bytes(val))
|
|
|
|
raise NotImplementedError(value_type, val)
|
|
|
|
|
|
def dump_memory(textio: TextIO, mem: bytes) -> None:
|
|
line_width = 16
|
|
|
|
prev_line = None
|
|
skip = False
|
|
for idx in range(0, len(mem), line_width):
|
|
line = ''
|
|
for idx2 in range(0, line_width):
|
|
line += f'{mem[idx + idx2]:02X}'
|
|
if idx2 % 2 == 1:
|
|
line += ' '
|
|
|
|
if prev_line == line:
|
|
if not skip:
|
|
textio.write('**\n')
|
|
skip = True
|
|
else:
|
|
textio.write(f'{idx:08x} {line}\n')
|
|
|
|
prev_line = line
|