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