Compare commits

..

6 Commits

15 changed files with 72 additions and 801 deletions

10
LICENSE
View File

@ -1,10 +0,0 @@
Permission to use, copy, modify, and/or distribute this software for
any purpose with or without fee is hereby granted.
THE SOFTWARE IS PROVIDED “AS IS” AND THE AUTHOR DISCLAIMS ALL
WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES
OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE
FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN
AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

View File

@ -1,54 +0,0 @@
# DP32-proto
Прототип эмулятора процессора DP32
## Описание
Это проект программной эмуляции и отладки процессора DP32 на языке python. В этом проекте можно найти 2 компонента - отладчик dp32dbg и эмулятор dp32emu. Первый полностью опирается на отладочную информацию в следующем формате. Это очень простой отладчик и очень нестабильный, поскольку писался с упором на скорость, а не на качество.
```json
{
"src": "absolute/path/to/source"
"labels" : {
"label1": 1
"label2": 2
...
},
"instructions" : {
"<offset-in-words>" : {
"lenght": 2,
"srcline": 3
}
...
}
}
```
## Установка
Стяните данный репозиторий и выполните в корне проекта команду `pip install .`
## Использование
После установки у вас появятся 2 программы:
- dp32emu - программный эмулятор процессора dp32. Принимает на вход бинарный файл с командами dp32, выполняет их и в результате своей работы выдает дамп памяти процессора после завершения работы
- dp32dbg - отладчик, опирающийся на вышеуказанный эмулятор. Позволяет выполнять действия пошаговой отладки и предоставляет базовую функциональность точек останова (но с ньюансами)
Посмотреть инструкцию по запуску проще всего передав флаг `-h` каждой из этих программ
## Использование отладчика
help не встроен в отладчик, поскольку до этого его функционал не был документирован. Приведу здесь основной набор команд отладчика
- `step/s` - выполняет одну инструкцию, на которую сейчас указывает регистр pc виртуальной машины
- `breakpoint <lineno>` - устанавливает точку останова на определенной строке исходного кода. С точками останова 2 ньюанса:
1. Они они не могут быть установлены на комментарии или пустые строки. Если на строке больше одной операции, точка будет установлена на самую раннюю
2. Когда виртуальная машина встретит точку останова, временно она не сможет выполнять команду continue, поскольку не предусмотрено ротации точек останова. Если нужно продолжить выполнение программы дальше - выполните одну команду step
- `continue/c` - выполняет программу до ближайшей точки останова
- `reset` - сбрасывает состояние виртуальной машины и ее память, но не точки останова
- `run` - сбрасывает состояние виртуальной машины и запускает выполнение кода
- `cbp` - очищает все точки останова
- `print/p <reg>` - печатает содержимое регистров. Регистры записываются в формате `r1-r255`, позволяет также посмотреть адреса меток (указав название меток в качестве параметра) и системные регистры: pc, cc
- `inspect/i <size> <amount> <place>` - печатает содержимое памяти по определенному адресу в определенном формате (строчная - знаковые, заглавная - беззнаковые): b/B - байт, h/H - полуслова - 16 бит, w/W - слова - 32 бита. `amount` - количество подряд идущих ячеек, содержимое которых нужно прочитать. `place` - место, с которого следует начать печать. Можеть представлять собой либо имя метки, либо полный адрес в десятичной или 16-ричной системе счисления

View File

@ -1,17 +1,7 @@
[project] [project]
name = "dp32-proto" name = "dp32-proto"
version = "0.1.1" version = "0.1.0"
description = "Add your description here" description = "Add your description here"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = [ dependencies = []
"py-cui>=0.1.6",
]
[project.scripts]
dp32emu = "dp32proto.main:main"
dp32dbg = "dp32proto.dbg_main:main"
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

View File

@ -1,3 +0,0 @@
import setuptools
setuptools.setup()

Binary file not shown.

View File

@ -1,4 +0,0 @@
# Почему такая некрасивая структура
Этот репозиторий нужен был для того, чтобы отладить программу для процессора DP32. Структура проекта сделана такой для того, чтобы сократить время, необходимое на структуризацию и разрешение зависимостей.

View File

@ -1,98 +0,0 @@
{
"labels": {
"_start": 0,
"_start_return": 9,
"f_fibo": 11,
"f_fibo_loop": 14,
"return": 27,
"result": 28,
"end": 29
},
"instructions": {
"0": {
"length": 1,
"srcline": 6
},
"1": {
"length": 1,
"srcline": 7
},
"2": {
"length": 2,
"srcline": 8
},
"4": {
"length": 2,
"srcline": 9
},
"6": {
"length": 2,
"srcline": 10
},
"8": {
"length": 1,
"srcline": 11
},
"9": {
"length": 2,
"srcline": 13
},
"11": {
"length": 1,
"srcline": 16
},
"12": {
"length": 1,
"srcline": 17
},
"13": {
"length": 1,
"srcline": 18
},
"14": {
"length": 1,
"srcline": 20
},
"15": {
"length": 1,
"srcline": 21
},
"16": {
"length": 1,
"srcline": 22
},
"17": {
"length": 1,
"srcline": 23
},
"18": {
"length": 1,
"srcline": 24
},
"19": {
"length": 2,
"srcline": 25
},
"21": {
"length": 2,
"srcline": 27
},
"23": {
"length": 2,
"srcline": 28
},
"25": {
"length": 2,
"srcline": 29
},
"27": {
"length": 1,
"srcline": 31
},
"28": {
"length": 1,
"srcline": 32
}
},
"src": "C:\\Users\\etdia\\code\\dp32-proto\\src\\test.dasm"
}

View File

@ -1,60 +0,0 @@
"""
Это очень грязное решение, но из-за страданий с пакетированием всех скриптов
оказалось тупо проще засунуть дебаггер сразу в эту папку и не мучить себя
"""
from typing import cast
from .dpdebugger import Debugger, DbgDict, DbgInstrDesc
from argparse import ArgumentParser
import json
from . import dbg_tui
def parse_dbg(dbg_json_dict: dict) -> DbgDict:
"""
При сериализации json всегда переводит ключи в формат строк и с этим
ничего поделать нельзя. Но для работы отладчика мне нужно, чтобы
ключами были числа, поэтому эта функция пытается привести к int
все ключи, лежащие в "instructions"
"""
instrs: dict[str | str, DbgInstrDesc] = dbg_json_dict["instructions"]
result: dict[int, DbgInstrDesc] = {
int(k) : instrs[k] for k in instrs
}
dbg_copy = dbg_json_dict.copy()
dbg_copy["instructions"] = result
dbg_copy = cast(DbgDict, dbg_copy)
return dbg_copy
def main():
parser = ArgumentParser(
prog="dp32dbg",
description="Bad debugger for DP32 processor"
)
parser.add_argument(
"-m",
"--memory",
required=True,
help="Initial memory for virtual machine"
)
parser.add_argument(
"-d",
"--debug-file",
required=True,
help="JSON with debug information. by default named dbg.json"
)
args = parser.parse_args()
with open(args.memory, "rb") as f:
mem = bytearray(f.read())
with open(args.debug_file, 'r') as f:
dbg_dict: DbgDict = parse_dbg(json.load(f))
dbg = Debugger(mem, dbg_dict)
with open (dbg_dict["src"], 'r', encoding="utf8") as f:
src = f.read()
dbg_tui.run_tui(dbg, src)
if __name__ == "__main__":
main()

View File

@ -1,164 +0,0 @@
from .dpdebugger import Debugger
from .vm import VMException, VMExceptionType, VMStatus
from dataclasses import dataclass
from typing import cast
import math
import py_cui
import py_cui.keys
from collections import deque
@dataclass
class SourceLines:
active: int
begining_number: int
lines: list[str]
def _get_source_lines(
srcline: int,
lines: list[str],
max_height: int
) -> SourceLines:
"""
Собирает выборку из строк исходного кода так, чтобы она
подходила по количесту строчек
"""
if srcline > len(lines):
raise IndexError(f"Cant take line number {srcline} from passed lines, "
"index is out of range")
start_bound = srcline - math.floor(max_height//2)
end_bound = srcline + math.ceil(max_height//2)
lines_len = len(lines)
if start_bound < 1:
end_bound += abs(start_bound) + 1
if end_bound > lines_len:
start_bound -= end_bound - lines_len
start_bound = max(start_bound, 1)
end_bound = min(end_bound, lines_len)
return SourceLines(
active=srcline,
begining_number=start_bound,
lines=list(lines[start_bound-1:end_bound])
)
class DebuggerUI:
ACTIVE_LINE_POINTER = "~> "
def __init__(self, master: py_cui.PyCUI, debugger: Debugger, src: str):
self.master = master
self.dbg = debugger
self.src = src
self.srclines = src.split('\n')[:-1]
self.source_panel = self.master.add_text_block(
"Source code", 0, 0
)
self.prompt_panel = self.master.add_text_box("Prompt",
1, 0,
row_span=1
)
self.prompt_panel.add_key_command(py_cui.keys.KEY_ENTER, self._process_prompt)
self.history_index = 0
self.prompt_panel.add_key_command(py_cui.keys.KEY_UP_ARROW, self._history_up)
self.prompt_panel.add_key_command(py_cui.keys.KEY_DOWN_ARROW, self._history_down)
self.terminal_panel = self.master.add_text_block("Debugger Console",
2, 0,
initial_text="to exit type in \"exit\""
)
self.master.move_focus(self.prompt_panel)
self.master.add_key_command(
py_cui.keys.KEY_ENTER,
lambda: self.master.move_focus(self.prompt_panel)
)
self.commands_history: deque[str] = deque(maxlen=10)
self._display_source_lines()
def _process_prompt(self):
user_input = self.prompt_panel.get()
self.prompt_panel.clear()
if user_input == "exit":
exit()
if user_input == "clear":
self.terminal_panel.clear()
return
if user_input.strip() == "":
if len(self.commands_history) < 1:
return
user_input = self.commands_history[0]
self.history_index = 0
self.commands_history.appendleft("")
try:
dbg_output = self.dbg.do_command(user_input.split())
except VMException as vme:
self._terminal_append(
"VM Exception: " + vme.message
)
return
# self._terminal_append(dbg_output)
self._terminal_append(dbg_output)
self._display_source_lines()
self.commands_history[0] = user_input
def _terminal_append(self, text: str):
self.terminal_panel.write(text)
viewport_height = self.terminal_panel.get_viewport_height()
contents = self.terminal_panel.get().split('\n')[-viewport_height:]
self.terminal_panel.set_text("\n".join(contents))
def _display_source_lines(self):
if self.dbg._vm.status == VMStatus.FINISHED:
return
try:
active_line = self.dbg.get_current_source_line_number()
except KeyError:
self._terminal_append(
"Cant find source line your are on. "
"May be you reached end of program. "
)
self._terminal_append(
"Please type \"reset\" to reset vm "
"it's initial state"
)
return
lines = _get_source_lines(
srcline=active_line,
lines=self.srclines,
max_height=self.source_panel.get_viewport_height()
)
# emplace line lumbers
linenos = [lines.begining_number + i
for i in range(len(lines.lines))]
lines_with_linenos: list[str] =[]
# длина самого большого числа для вычисления выравнивания
align_with = len(str(linenos[-1]))
for lineno, line in zip(linenos, lines.lines):
if lineno == lines.active:
padding = len(line) - len(stripped := line.lstrip(" "))
line = self.ACTIVE_LINE_POINTER.rjust(padding) + stripped
lines_with_linenos.append(
f"{str(lineno).rjust(align_with)}. {line}"
)
self.source_panel.set_text("\n".join(lines_with_linenos))
def _history_up(self):
if len(self.commands_history) == 0:
return
self.history_index += 1
self.history_index %= len(self.commands_history)
self.prompt_panel.set_text(self.commands_history[self.history_index])
def _history_down(self):
if len(self.commands_history) == 0:
return
self.history_index -= 1
self.history_index %= len(self.commands_history)
self.prompt_panel.set_text(self.commands_history[self.history_index])
def run_tui(debugger: Debugger, srcfile: str):
root = py_cui.PyCUI(3, 1)
root.set_title("Stupid DP32 debugger")
root.set_status_bar_text("")
DebuggerUI(root, debugger, srcfile)
root.start()

View File

@ -1,212 +0,0 @@
"""
Отладчик принимает на вход довольно много всего и пытается вам
хоть как-то помочь
"""
from .vm import VM, VMException, Breakpoint, Condition, WORD_SIZE
from dataclasses import dataclass
from typing import TypedDict, Callable, cast
import re
import struct
class DbgInstrDesc(TypedDict):
"""
Описание отдельно взятой инструкции по определенному оффсету
"""
length: int
srcline: int
class DbgDict(TypedDict):
"""
Урезанная версия полной отладочной информации, предоставляемой
в файле dbg.json. Содержит ту информацию, которую может
предоставить ассемблер
"""
src: str
labels: dict[str, int]
instructions: dict[int, DbgInstrDesc]
@dataclass
class Debugger:
_vm: VM
_dbg_dict: DbgDict
_callbacks_table: dict[str, Callable[[list[str]], str]]
_breakpoints: set[int]
_source_lines: list[str]
# will be nessesary in _inspect
UNPACK_SYMBOL_TABLE = {
"b": "b",
"B": "B",
"h": "h",
"H": "H",
"w": "i",
"W": "I"
}
SIZE_SYMBOL_TABLE = {
"b": 1,
"B": 1,
"h": 2,
"H": 2,
"w": 4,
"W": 4
}
def __init__(self, mem: bytearray, dbg_dict: DbgDict):
with open(dbg_dict["src"], 'r', encoding="utf8") as f:
self._source_lines = f.readlines()
self._vm = VM(mem)
self._dbg_dict = dbg_dict
self._breakpoints = set()
self.__init_callbacks__()
self.last_breakpoint: Breakpoint | None = None
def __init_callbacks__(self):
self._callbacks_table = {
"step": self._step,
"s": self._step,
"continue": self._continue,
"c": self._continue,
"breakpoint": self._breakpoint,
"b": self._breakpoint,
"clearbreakpoints": self._breakpointsclear,
"cbp": self._breakpointsclear,
"print": self._print,
"p": self._print,
"run": self._run,
"r": self._run,
"inspect": self._inspect,
"i": self._inspect,
"reset": self._reset
}
def do_command(self, command: list[str]) -> str:
callback_name = command[0]
if not callback_name in self._callbacks_table:
return "Unknown command"
return self._callbacks_table[callback_name](command[1:])
def get_current_source_line_number(self) -> int:
return self._dbg_dict["instructions"][self._vm.pc.value]['srcline']
def _reset(self, args: list[str]) -> str:
self._vm.reset()
return "VM reseted"
def _step(self, args: list[str]) -> str:
try:
self._vm.step()
except VMException as e:
# TODO: тут необходимо выкорчевать из
# исключения текст ошибки
return f"Virtual machine exception: {e.message}"
return ""
def _continue(self, args: list[str]) -> str:
try:
self._vm.continue_()
except Breakpoint:
return ""
return "Program finished"
def _breakpoint(self, args: list[str]) -> str:
if len(args) == 0:
addr = self._vm.pc.value
self._breakpoints.add(addr)
src_line = self._dbg_dict["instructions"][addr]["srcline"]
return f"Breakpoint succesfully set on line {src_line}"
desired_addr: int = int(args[0])
addr = -1
# находим за линейное время. Плохая практика, но так как
# сходные коды вряд ли будут длиннее 1000 строк - приемлемо
for addr, desc in self._dbg_dict["instructions"].items():
desc = cast(DbgInstrDesc, desc)
if not desc["srcline"] == desired_addr:
continue
self._breakpoints.add(int(addr))
self._vm.breakpoints.add(int(addr))
return f"Breakpoint succesfully set on line {desired_addr}"
return f"Couldn't place breakpoint on src line {desired_addr}"
def _breakpointsclear(self, args: list[str]):
self._breakpoints.clear()
self._vm.breakpoints.clear()
return "Cleared all breakpoints"
def _print(self, args: list[str]) -> str:
to_print = args[0]
if to_print == "pc":
return "pc: " + str(self._vm.pc.value)
elif to_print == "cc":
flags = Condition(self._vm.cc.value)
return (
f"v: {flags.v}\n"
f"n: {flags.n}\n"
f"z: {flags.z}\n"
)
elif re.fullmatch(
# r0-r255
r'r([0-9]|[1-9][0-9]|1[0-9][0-9]|2[1-4][1-9]|25[1-5])',
to_print):
index = int(to_print[1:])
return f"{to_print}: {self._vm.registers[index].value}"
elif to_print in self._dbg_dict["labels"]:
return f'{to_print}: {self._dbg_dict["labels"][to_print]}'
else:
return "Can't print required value"
def _run(self, args: list[str]) -> str:
try:
self._vm.run()
except Breakpoint as b:
return (f'breakpoint on line '
f'{self._dbg_dict["instructions"][b.address]["srcline"]}')
return "Program finished"
def _inspect(self, args: list[str]) -> str:
size_arg = args.pop(0)
amount = int(args.pop(0))
mem_location = args.pop(0)
offset = -1
if mem_location in self._dbg_dict["labels"]:
offset = self._dbg_dict["labels"][mem_location]
elif mem_location.isdigit():
offset = int(mem_location) * WORD_SIZE
elif mem_location.startswith("0x"):
offset = int(mem_location, 16)
else:
return ("You passed wrong offset parameter. It should be either "
"name of some label or explicit decimal or hexdecimal "
"number, in the second case it should start with 0x")
offset = self._vm._to_raw_bytes_offset(offset)
try:
sym = self.UNPACK_SYMBOL_TABLE[size_arg]
size = self.SIZE_SYMBOL_TABLE[size_arg]
except KeyError:
return ("Size Error. Please select one of the following "
"options as size: b/B - byte, h/H - 2 bytes, "
"w/W - 4 bytes. Big - unsigned, small - signed")
contents = struct.unpack(
">"+sym*amount,
self._vm.mem[offset:offset+size*amount]
)
return f"{mem_location}:" + " ".join(map(str, contents))
def _recover_from_breakpoint(self, cur_bp: Breakpoint):
"""
Когда виртуальная машина сталкивается с точкой останова, она на
ней зависает навечно. Для продолжения работы надо текущую точку
останова убрать, сделать шаг, а потом добавить. Предыдущая точка
остановка как раз хранится в дебаггере
"""
if not self.last_breakpoint:
return
if cur_bp.address != self.last_breakpoint.address:
return
self._vm.breakpoints.remove(cur_bp.address)

View File

@ -1,5 +1,5 @@
from argparse import ArgumentParser from argparse import ArgumentParser
from .vm import VM from vm import VM
def main(): def main():
parser = ArgumentParser( parser = ArgumentParser(
@ -15,6 +15,7 @@ def main():
default="out.mem" default="out.mem"
) )
args = parser.parse_args() args = parser.parse_args()
print(args)
with open(args.mem_file, 'rb') as f: with open(args.mem_file, 'rb') as f:
mem = bytearray(f.read()) mem = bytearray(f.read())
vm = VM(mem) vm = VM(mem)

View File

@ -43,13 +43,13 @@ OpD = OpcodeDescription
OPCODES = { OPCODES = {
# block 1 # block 1
0x00: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.ADD), 0x00: OpD(OpF(0), OpL.MATH, OpA.ADD),
0x10: OpD(OpF.QUICK, OpL.MATH, OpA.ADD), 0x10: OpD(OpF.QUICK, OpL.MATH, OpA.ADD),
0x01: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.SUB), 0x01: OpD(OpF(0), OpL.MATH, OpA.SUB),
0x11: OpD(OpF.QUICK, OpL.MATH, OpA.SUB), 0x11: OpD(OpF.QUICK, OpL.MATH, OpA.SUB),
0x02: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.MUL), 0x02: OpD(OpF(0), OpL.MATH, OpA.MUL),
0x12: OpD(OpF.QUICK, OpL.MATH, OpA.MUL), 0x12: OpD(OpF.QUICK, OpL.MATH, OpA.MUL),
0x03: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.DIV), 0x03: OpD(OpF(0), OpL.MATH, OpA.DIV),
0x13: OpD(OpF.QUICK, OpL.MATH, OpA.DIV), 0x13: OpD(OpF.QUICK, OpL.MATH, OpA.DIV),
0x04: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.AND), 0x04: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.AND),
0x05: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.OR), 0x05: OpD(OpF.UNEXPANDED, OpL.MATH, OpA.OR),
@ -63,6 +63,9 @@ OPCODES = {
# block 3 # block 3
0x40: OpD(OpF(0), OpL.BRANCH, OpA.BRANCH), 0x40: OpD(OpF(0), OpL.BRANCH, OpA.BRANCH),
0x50: OpD(OpF.QUICK, OpL.BRANCH, OpA.BRANCH), 0x50: OpD(OpF.QUICK, OpL.BRANCH, OpA.BRANCH),
0x41: OpD(OpF(0), OpL.BRANCH, OpA.IND_BRANCH), 0x51: OpD(OpF(0), OpL.BRANCH, OpA.IND_BRANCH),
} }
if __name__ == "__main__":
print(hex(12))
print(~12)

View File

@ -1,16 +1,14 @@
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import ClassVar, Callable from typing import ClassVar, Callable
from ctypes import c_uint32, c_int32 from ctypes import c_uint32, c_int32
import struct import struct
from .optable import OPCODES, OpcodeDescription, OpL, OpA, OpF, OpD from optable import OPCODES, OpcodeDescription, OpL, OpA, OpF, OpD
from enum import IntFlag, Enum, auto from enum import IntFlag, Enum, auto
WORD_SIZE: int = 4
class VMFlags(IntFlag): class VMFlags(IntFlag):
# if last instruction is big, then you should # if last instruction is big, then you should
# skip more memory on step # skip more memory on step
AFTER_BRANCH = auto()
EXPANDED_INSTR = auto() EXPANDED_INSTR = auto()
class Condition: class Condition:
@ -25,11 +23,6 @@ class VMCC(IntFlag):
NEGATIVE = 1 << 1 NEGATIVE = 1 << 1
ZERO = 1 << 0 ZERO = 1 << 0
class VMStatus(Enum):
RUNNING = auto()
INITED = auto()
FINISHED = auto()
@dataclass @dataclass
class Breakpoint(Exception): class Breakpoint(Exception):
address: int address: int
@ -42,7 +35,6 @@ class VMExceptionType(Enum):
class VMException(Exception): class VMException(Exception):
cause: VMExceptionType cause: VMExceptionType
pc: int pc: int
message: str = ""
@dataclass @dataclass
class VM: class VM:
@ -53,36 +45,38 @@ class VM:
registers: list[c_int32] registers: list[c_int32]
breakpoints: set[int] breakpoints: set[int]
_vm_flags: VMFlags _vm_flags: VMFlags
status: VMStatus
def __init__(self, mem: bytearray): def __init__(self, mem: bytearray):
self._initial_mem = mem.copy() self.mem: bytearray = mem
self.__init_callbacks__() self.cc: VMCC = VMCC(0)
self.reset() self.pc: c_uint32 = c_uint32(0)
self.registers: list[c_int32] = [c_int32(0) for _ in range(256)]
self.breakpoints: set[int] = set() self.breakpoints: set[int] = set()
self._vm_flags: VMFlags = VMFlags(0)
self.__init_callbacks__()
def __init_callbacks__(self): def __init_callbacks__(self):
VM.instr_callbacks = { VM.instr_callbacks = {
# ariphmetic # ariphmetic
OpD(OpF.UNEXPANDED, OpL.MATH, OpA.ADD): OpD(OpF(0), OpL.MATH, OpA.ADD):
self._math_callback_gen(lambda lhs, rhs: lhs + rhs), self._math_callback_gen(lambda lhs, rhs: lhs + rhs),
OpD(OpF.QUICK, OpL.MATH, OpA.ADD): OpD(OpF.QUICK, OpL.MATH, OpA.ADD):
self._math_quick_callback_gen(lambda lhs, rhs: lhs + rhs), self._math_quick_callback_gen(lambda lhs, rhs: lhs + rhs),
OpD(OpF.UNEXPANDED, OpL.MATH, OpA.SUB): OpD(OpF(0), OpL.MATH, OpA.SUB):
self._math_callback_gen(lambda lhs, rhs: lhs - rhs), self._math_callback_gen(lambda lhs, rhs: lhs - rhs),
OpD(OpF.QUICK, OpL.MATH, OpA.SUB): OpD(OpF.QUICK, OpL.MATH, OpA.SUB):
self._math_quick_callback_gen(lambda lhs, rhs: lhs - rhs), self._math_quick_callback_gen(lambda lhs, rhs: lhs - rhs),
OpD(OpF.UNEXPANDED, OpL.MATH, OpA.MUL): OpD(OpF(0), OpL.MATH, OpA.MUL):
self._math_callback_gen(lambda lhs, rhs: lhs * rhs), self._math_callback_gen(lambda lhs, rhs: lhs * rhs),
OpD(OpF.QUICK, OpL.MATH, OpA.MUL): OpD(OpF.QUICK, OpL.MATH, OpA.MUL):
self._math_quick_callback_gen(lambda lhs, rhs: lhs * rhs), self._math_quick_callback_gen(lambda lhs, rhs: lhs * rhs),
OpD(OpF.UNEXPANDED, OpL.MATH, OpA.DIV): OpD(OpF(0), OpL.MATH, OpA.DIV):
self._math_callback_gen(lambda lhs, rhs: lhs // rhs), self._math_callback_gen(lambda lhs, rhs: lhs // rhs),
OpD(OpF.QUICK, OpL.MATH, OpA.DIV): OpD(OpF.QUICK, OpL.MATH, OpA.DIV):
@ -122,103 +116,56 @@ class VM:
self._branch_indexed_callback self._branch_indexed_callback
} }
def reset(self):
self.mem: bytearray = self._initial_mem.copy()
self.cc: VMCC = VMCC(0)
self.pc: c_uint32 = c_uint32(0)
self.registers: list[c_int32] = [c_int32(0) for _ in range(256)]
self._vm_flags: VMFlags = VMFlags(0)
self.status = VMStatus.INITED
def step(self) -> None: def step(self) -> None:
""" """
Make one step (only step into) Make one step (only step into)
""" """
if self._to_raw_bytes_offset(self.pc) > len(self.mem) - WORD_SIZE: opcode = self.mem[self.pc.value * 4]
self.status = VMStatus.FINISHED opdesc = self._fetch_opcode_desc(opcode)
raise VMException( args = self._parse_arguments(opdesc)
VMExceptionType.END_OF_MEM, # По какой-то причине адрессация работает
self.pc.value, # так, будто мы на 1 слово впереди опкода
"couldn't perform step because end of memory occured" self.pc = c_uint32(self.pc.value + 1)
) # сбрасываем флаг AFTER_BRANCH
self.status = VMStatus.RUNNING
opcode, *_ = instr = self._fetch_instr()
opdesc = self._get_opcode_desc(opcode)
args: tuple[int, ...] = self._parse_instr_fields(bytes(instr))
if not (OpF.UNEXPANDED in opdesc.flags or
OpF.QUICK in opdesc.flags):
extra = struct.unpack(">i", self._fetch_instr())[0]
args = (*args, extra)
self._run_callback(opdesc, args)
self._vm_flags &= ~(VMFlags.EXPANDED_INSTR)
self._vm_flags &= ~(VMFlags.AFTER_BRANCH) self._vm_flags &= ~(VMFlags.AFTER_BRANCH)
self._run_callback(opdesc, args)
if VMFlags.EXPANDED_INSTR in self._vm_flags:
self.pc = c_uint32(self.pc.value + 1)
def continue_(self) -> None: def continue_(self) -> None:
""" """
Continue from current breakpoint Continue from current breakpoint
""" """
while self._to_raw_bytes_offset(self.pc.value) < len(self.mem): while self.pc.value < len(self.mem):
if self.pc.value in self.breakpoints: if self.pc.value in self.breakpoints:
raise Breakpoint(self.pc.value) raise Breakpoint(self.pc.value)
self.step() self.step()
self.status = VMStatus.FINISHED
def run(self) -> None: def run(self) -> None:
""" """
Run from very beginning Run from very beginning
""" """
self.reset() self.pc = c_uint32(0)
while (self._to_raw_bytes_offset(self.pc.value) < len(self.mem)): while (self.pc.value < len(self.mem) // 4):
self.continue_() self.continue_()
self.status = VMStatus.FINISHED
def _fetch_instr(self) -> bytearray: def _fetch_opcode_desc(self, opcode: int):
"""
Берет из памяти следущее слово, возвращает его
ВАЖНО: инкрементирует счетчик команд на каждом вызове
"""
addr = self._to_raw_bytes_offset(self.pc)
instr = self.mem[addr:addr+WORD_SIZE]
self._increment_pc()
return instr
def _get_opcode_desc(self, opcode: int):
"""
Обертка над обращением к таблице опкодов из файлика optable.py.
Нужна чтобы прокидывать исключения виртуальной машины, а не
KeyError
"""
if not opcode in OPCODES:
raise VMException(
VMExceptionType.INVALID_OPCODE,
self.pc.value,
f"Couldn't resolve an opcode {hex(opcode)}"
)
return OPCODES[opcode] return OPCODES[opcode]
def _to_raw_bytes_offset(self, pointer: int | c_uint32 | c_int32) -> int: def _parse_arguments(self, opdesc: OpcodeDescription) -> tuple[int, ...]:
""" addr = self.pc.value * 4
Переводит адресс из указателей в рамках процессора, которые main_part = struct.unpack(">BBBb", self.mem[addr:addr+4])
являются 32-битными, в формат смещения в сырых байтах в памяти
"""
if isinstance(pointer, int):
return pointer * WORD_SIZE
return pointer.value * WORD_SIZE
def _parse_instr_fields( if not OpF.UNEXPANDED in opdesc.flags or OpF.QUICK in opdesc.flags:
self, upper_part = struct.unpack(">i", self.mem[addr+4:addr+8])
instr: bytes return (*main_part, *upper_part)
) -> tuple[int, int, int, int]: return main_part
return struct.unpack(">BBBb", instr)
def _run_callback( def _run_callback(
self, self,
opdesc: OpcodeDescription, opdesc: OpcodeDescription,
args: tuple[int, ...] args: tuple[int, ...]
) -> None: ) -> None:
if not (OpF.QUICK in opdesc.flags or OpF.UNEXPANDED in opdesc.flags):
self._vm_flags |= VMFlags.EXPANDED_INSTR
if opdesc.layout == OpL.MATH: if opdesc.layout == OpL.MATH:
assert len(args) == 4 assert len(args) == 4
_, r3, r1, r2_or_i8 = args _, r3, r1, r2_or_i8 = args
@ -233,13 +180,13 @@ class VM:
assert len(args) == 4 assert len(args) == 4
_, r3, r1, i8 = args _, r3, r1, i8 = args
self.instr_callbacks[opdesc]( self.instr_callbacks[opdesc](
r3, r1, i8 self, r3, r1, i8
) )
else: else:
assert len(args) == 5 assert len(args) == 5
_, r3, r1, _, disp = args _, r3, r1, _, disp = args
self.instr_callbacks[opdesc]( self.instr_callbacks[opdesc](
r3, r1, disp self, r3, r1, disp
) )
if opdesc.layout == OpL.BRANCH: if opdesc.layout == OpL.BRANCH:
@ -262,25 +209,6 @@ class VM:
cond, disp cond, disp
) )
@staticmethod
def _perform_ariphmetic_operation(
lhs: int,
rhs: int,
op: Callable[[int, int], int]
) -> tuple[int, VMCC]:
cc = VMCC(0)
result = op(lhs, rhs)
if result < 0:
cc |= VMCC.NEGATIVE
elif result == 0:
cc |= VMCC.ZERO
# самая дорогая проверка на переполнение)
try:
struct.pack('i', result)
except struct.error:
cc |= VMCC.OVERFLOW
return result, cc
def _math_callback_gen( def _math_callback_gen(
self, self,
operation: Callable[[int, int], int] operation: Callable[[int, int], int]
@ -293,16 +221,10 @@ class VM:
def callback(self, r3: int, r1: int, r2: int): def callback(self, r3: int, r1: int, r2: int):
lhs = self.registers[r1].value lhs = self.registers[r1].value
rhs = self.registers[r2].value rhs = self.registers[r2].value
result, cc = self._perform_ariphmetic_operation( self.registers[r3] = c_int32(operation(lhs, rhs))
lhs,
rhs,
operation)
self.registers[r3] = c_int32(result)
self.cc = cc
return callback return callback
def _math_quick_callback_gen( def _math_quick_callback_gen(
self, self,
operation: Callable[[int, int], int] operation: Callable[[int, int], int]
@ -318,45 +240,41 @@ class VM:
def callback(self, r3: int, r1: int, i8: int) -> None: def callback(self, r3: int, r1: int, i8: int) -> None:
self.cc = VMCC(0) self.cc = VMCC(0)
lhs = self.registers[r1].value lhs = self.registers[r1].value
result, flags = self._perform_ariphmetic_operation( result = operation(lhs, i8)
lhs, if result < 0:
i8, self.cc |= VMCC.NEGATIVE
operation) elif result == 0:
self.cc |= VMCC.ZERO
# самая дорогая проверка на переполнение)
try:
struct.pack('i', result)
except struct.error:
self.cc |= VMCC.OVERFLOW
self.registers[r3] = c_int32(result) self.registers[r3] = c_int32(result)
self.cc = flags
return callback return callback
def _load_callback(self, r3: int, r1: int, disp: int) -> None: def _load_callback(self, r3: int, r1: int, disp: int) -> None:
addr = self._to_raw_bytes_offset(self.registers[r1].value + disp) addr = (self.registers[r1].value + disp) * 4
raw_bytes = self.mem[addr:addr+WORD_SIZE]
self.registers[r3] = c_int32( self.registers[r3] = c_int32(
struct.unpack( struct.unpack(">i", self.mem[addr:addr+4])[0]
">i", raw_bytes
)[0]
) )
def _store_callback(self, r3: int, r1: int, disp: int) -> None: def _store_callback(self, r3: int, r1: int, disp: int) -> None:
addr = self._to_raw_bytes_offset(self.registers[r1].value + disp) addr = (self.registers[r1].value + disp) * 4
self.mem[addr:addr+WORD_SIZE] = struct.pack( self.mem[addr:addr+4] = struct.pack(">i", self.registers[r3].value)
">i",
self.registers[r3].value
)
def _branch_callback(self, cond: int, disp: int) -> None: def _branch_callback(self, cond: int, disp: int) -> None:
c = Condition(cond) c = Condition(cond)
vm_c = Condition(self.cc) vm_c = Condition(self.cc)
if (c.v & vm_c.v) | (c.n & vm_c.n) | (c.z & vm_c.z) == c.i: if (c.v & vm_c.v) & (c.n & vm_c.n) & (c.z & vm_c.z) == c.i:
self._vm_flags = VMFlags.AFTER_BRANCH self._vm_flags |= VMFlags.AFTER_BRANCH
self.pc = c_uint32(self.pc.value + disp) self.pc = c_uint32(self.pc.value + disp)
def _branch_indexed_callback(self, cond: int, r1: int, disp: int) -> None: def _branch_indexed_callback(self, cond: int, r1: int, disp: int) -> None:
c = Condition(cond) c = Condition(cond)
vm_c = Condition(self.cc.value) vm_c = Condition(self.cc.value)
if (c.v & vm_c.v) | (c.n & vm_c.n) | (c.z & vm_c.z) == c.i: if (c.v & vm_c.v) & (c.n & vm_c.n) & (c.z & vm_c.z) == c.i:
self._vm_flags = VMFlags.AFTER_BRANCH self._vm_flags |= VMFlags.AFTER_BRANCH
addr = self.registers[r1].value + disp addr = self.registers[r1].value + disp
self.pc = c_uint32(addr) self.pc = c_uint32(addr)
def _increment_pc(self):
self.pc = c_uint32(self.pc.value + 1)

36
uv.lock generated
View File

@ -1,36 +0,0 @@
version = 1
revision = 1
requires-python = ">=3.11"
[[package]]
name = "dp32-proto"
version = "0.1.1"
source = { editable = "." }
dependencies = [
{ name = "py-cui" },
]
[package.metadata]
requires-dist = [{ name = "py-cui", specifier = ">=0.1.6" }]
[[package]]
name = "py-cui"
version = "0.1.6"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "windows-curses", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/da/d8/bb92087dc36dc94a60993955133d0f6473237a6ce1a6145ad95dce3b2a21/py_cui-0.1.6.tar.gz", hash = "sha256:91186f33f26216cd82a676e6b02e85110a7b562d8ab9792359cfdde4ee6b9abc", size = 63009 }
[[package]]
name = "windows-curses"
version = "2.4.1"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/74/b3/46a2508fff83b5affb5a311797c584c494b4b0c4c8796a1afa2c1b1e03ac/windows_curses-2.4.1-cp311-cp311-win32.whl", hash = "sha256:4fa1a176bfcf098d0c9bb7bc03dce6e83a4257fc0c66ad721f5745ebf0c00746", size = 71129 },
{ url = "https://files.pythonhosted.org/packages/50/29/fd7c0c80177d8df55f841504a5f332b95b0002c80f82055913b6caac94e6/windows_curses-2.4.1-cp311-cp311-win_amd64.whl", hash = "sha256:fd7d7a9cf6c1758f46ed76b8c67f608bc5fcd5f0ca91f1580fd2d84cf41c7f4f", size = 81431 },
{ url = "https://files.pythonhosted.org/packages/f4/99/60f34e51514c82631aa5c6d21eab88ce1701562de9844608411e39462a46/windows_curses-2.4.1-cp312-cp312-win32.whl", hash = "sha256:bdbe7d58747408aef8a9128b2654acf6fbd11c821b91224b9a046faba8c6b6ca", size = 71489 },
{ url = "https://files.pythonhosted.org/packages/62/8f/d908bcab1954375b156a9300fa86ceccb110f39457cd6fd59c72777626ab/windows_curses-2.4.1-cp312-cp312-win_amd64.whl", hash = "sha256:5c9c2635faf171a229caca80e1dd760ab00db078e2a285ba2f667bbfcc31777c", size = 81777 },
{ url = "https://files.pythonhosted.org/packages/25/a0/e8d074f013117633f6b502ca123ecfc377fe0bd36818fe65e8935c91ca9c/windows_curses-2.4.1-cp313-cp313-win32.whl", hash = "sha256:05d1ca01e5199a435ccb6c8c2978df4a169cdff1ec99ab15f11ded9de8e5be26", size = 71390 },
{ url = "https://files.pythonhosted.org/packages/2b/4b/2838a829b074a68c570d54ae0ae8539979657d3e619a4dc5a4b03eb69745/windows_curses-2.4.1-cp313-cp313-win_amd64.whl", hash = "sha256:8cf653f8928af19c103ae11cfed38124f418dcdd92643c4cd17239c0cec2f9da", size = 81636 },
]