58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
from typing import TextIO
|
|
|
|
from phasmplatform.common.router import BaseRouter
|
|
|
|
|
|
class BaseRunner:
|
|
__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
|