from typing import Callable, List, TextIO from phasmplatform.common.router import BaseRouter from phasmplatform.common.method import Method from phasmplatform.common.value import Value class RunnerInterface: __slots__ = ('router', ) def do_call(self, method: Method, args: List[Value], on_result: Callable[[Value], None]) -> None: raise NotImplementedError class BaseRunner(RunnerInterface): __slots__ = ('router', ) router: BaseRouter def __init__(self, router: BaseRouter) -> None: self.router = 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 handle_message(self, namespace: bytes, topic: bytes, kind: bytes, body: bytes) -> None: raise NotImplementedError def post_message(self, namespace_ptr: int, topic_ptr: int, kind_ptr: int, body_ptr: int) -> None: namespace = self.read_bytes(namespace_ptr) topic = self.read_bytes(topic_ptr) kind = self.read_bytes(kind_ptr) body = self.read_bytes(body_ptr) self.router.post_message(namespace, topic, kind, body) 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