Sprinter: добавить отладку C-исходников и интеграцию VS Code

2026-09-15 17:58:41 +03:00
parent 50c6e56b7b
commit e4695b8281
62 changed files with 7147 additions and 27 deletions
@@ -0,0 +1,276 @@
+"""Высокоуровневая source-debug сессия поверх проверенной карты и IPC."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from .image import read_ihx
+from .model import DebugMap
+from .transport import FileBridge
+
+
+class SessionError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class MappingSnapshot:
+    generation: int
+    registers: dict[str, int]
+    bank_pages: dict[int, int]
+
+
+class DebugSession:
+    """Связывает build identity, состояние CPU и логические операции IDE."""
+
+    def __init__(self, model: DebugMap, bridge: FileBridge):
+        self.model = model
+        self.bridge = bridge
+        self.mapping: MappingSnapshot | None = None
+        self.resident_pages: dict[int, int] = {}
+        self.breakpoints: dict[int, list[int]] = {}
+        self._next_breakpoint = 1
+        stem = Path(model.manifest['executable']).stem
+        self.image = read_ihx(model.directory / (stem + '.ihx'))
+
+    def attach(self) -> dict:
+        """Проверяет пакет, остановку CPU, резидентный код и bank mapping."""
+        try:
+            self.model.verify_executable()
+            hello = self.bridge.handshake()
+            snapshot = self.bridge.wait_stopped()
+            mapping = self._mapping(snapshot)
+            if mapping.bank_pages and not hello.get('capabilities', {}).get('bank_guard'):
+                raise SessionError('Backend не поддерживает безопасные банковские точки')
+            self._verify_loaded_code(mapping)
+        except (ValueError, OSError) as error:
+            raise SessionError(str(error)) from error
+        self.mapping = mapping
+        self._remember_resident_pages(mapping)
+        return {
+            'build_id': self.model.manifest['build_id'],
+            'state': snapshot['state'],
+            'protocol': hello['protocol'],
+            'capabilities': hello.get('capabilities', {}),
+            'generation': mapping.generation,
+            'pc': mapping.registers['PC'],
+            'bank_pages': mapping.bank_pages,
+            'stale_sources': self.model.stale_sources,
+            'location': self.where(snapshot),
+        }
+
+    def refresh(self) -> MappingSnapshot:
+        snapshot = self.bridge.wait_stopped()
+        self.mapping = self._mapping(snapshot)
+        self._remember_resident_pages(self.mapping)
+        self._verify_loaded_code(self.mapping)
+        return self.mapping
+
+    def _remember_resident_pages(self, mapping: MappingSnapshot) -> None:
+        bank_values = set(mapping.bank_pages.values())
+        windows = {instruction['window'] for instruction in self.model.instructions.values()
+                   if instruction['bank'] is not None}
+        for window in windows:
+            value = self._page(mapping, window)
+            if value is not None and value not in bank_values:
+                self.resident_pages[window] = value
+
+    @staticmethod
+    def _page(mapping: MappingSnapshot, window: int) -> int | None:
+        """Sprinter state PG может включать старшие флаги; port хранит byte."""
+        value = mapping.registers.get(f'PG{window}')
+        return None if value is None else value & 0xff
+
+    def _mapping(self, snapshot: dict) -> MappingSnapshot:
+        registers = snapshot.get('registers')
+        if snapshot.get('state') != 'stopped' or not isinstance(registers, dict):
+            raise SessionError('Backend не предоставил регистры остановленного CPU')
+        if 'PC' not in registers:
+            raise SessionError('Backend не предоставил PC')
+        maximum = max((location['bank'] or 0 for location in
+                       [*self.model.functions, *self.model.variables,
+                        *self.model.instructions.values()]), default=0)
+        pages: dict[int, int] = {}
+        table = self.model.symbols.get('_bank_pages')
+        if maximum:
+            if table is None:
+                raise SessionError('В банковской сборке отсутствует _bank_pages')
+            data = bytes.fromhex(self.bridge.request(
+                'memory', address=table, length=maximum + 1)['hex'])
+            pages = {bank: data[bank] for bank in range(1, maximum + 1)}
+            if any(value == 0 for value in pages.values()) or \
+                    len(set(pages.values())) != len(pages):
+                raise SessionError('_bank_pages ещё не готова или содержит нули/дубликаты')
+            windows = {location['window'] for location in
+                       [*self.model.functions, *self.model.variables]
+                       if location['bank'] is not None}
+            missing = [window for window in windows if f'PG{window}' not in registers]
+            if missing:
+                raise SessionError('Backend не предоставил PG для банковских окон')
+        generation = self.bridge.generation
+        if generation is None:
+            raise SessionError('Backend не предоставил generation')
+        return MappingSnapshot(generation, dict(registers), pages)
+
+    def _expected_ranges(self, mapping: MappingSnapshot):
+        expected: dict[int, int] = {}
+        active_windows = {instruction['window'] for instruction in self.model.instructions.values()
+                          if instruction['bank'] is not None and
+                          self._page(mapping, instruction['window']) ==
+                          mapping.bank_pages[instruction['bank']]}
+        for instruction in self.model.instructions.values():
+            bank = instruction['bank']
+            if bank is not None:
+                if self._page(mapping, instruction['window']) != mapping.bank_pages[bank]:
+                    continue
+            elif instruction['window'] in active_windows:
+                # Физическая страница банка закрыла весь resident window.
+                continue
+            start = instruction['link_address']
+            logical = instruction['logical_address']
+            for offset in range(instruction['size']):
+                if start + offset not in self.image:
+                    raise SessionError(f'В IHX нет инструкции по адресу {start + offset:#x}')
+                value = self.image[start + offset]
+                if logical + offset in expected and expected[logical + offset] != value:
+                    raise SessionError(f'Неоднозначный ожидаемый байт {logical + offset:#x}')
+                expected[logical + offset] = value
+        return expected
+
+    def _verify_loaded_code(self, mapping: MappingSnapshot) -> None:
+        expected = self._expected_ranges(mapping)
+        addresses = sorted(expected)
+        ranges: list[tuple[int, int]] = []
+        for address in addresses:
+            if not ranges or address != ranges[-1][1] or address - ranges[-1][0] >= 4096:
+                ranges.append((address, address + 1))
+            else:
+                ranges[-1] = (ranges[-1][0], address + 1)
+        for start, end in ranges:
+            actual = bytes.fromhex(self.bridge.request(
+                'memory', address=start, length=end - start)['hex'])
+            wanted = bytes(expected[address] for address in range(start, end))
+            if actual != wanted:
+                mismatch = next(i for i, pair in enumerate(zip(actual, wanted))
+                                if pair[0] != pair[1])
+                raise SessionError(
+                    f'Образ в MAME не соответствует build по адресу {start + mismatch:#x}')
+
+    def _link_address(self, pc: int, mapping: MappingSnapshot) -> int:
+        candidates = []
+        for instruction in self.model.instructions.values():
+            if instruction['bank'] is None:
+                continue
+            if not (instruction['logical_address'] <= pc <
+                    instruction['logical_address'] + instruction['size']):
+                continue
+            if self._page(mapping, instruction['window']) == mapping.bank_pages[instruction['bank']]:
+                candidates.append(instruction['link_address'] +
+                                  pc - instruction['logical_address'])
+        candidates = sorted(set(candidates))
+        if len(candidates) > 1:
+            raise SessionError(f'Неоднозначное банковское отображение PC={pc:#x}')
+        return candidates[0] if candidates else pc
+
+    def where(self, snapshot: dict | None = None) -> dict:
+        mapping = self._mapping(snapshot) if snapshot is not None else self.refresh()
+        if snapshot is not None:
+            self._verify_loaded_code(mapping)
+        pc = mapping.registers['PC']
+        link = self._link_address(pc, mapping)
+        return {'pc': pc, 'link_address': link, **self.model.addr2line(link)}
+
+    def _guard(self, location: dict, mapping: MappingSnapshot) -> dict:
+        bank = location['bank']
+        window = location['window']
+        banked_window = any(instruction['bank'] is not None and
+                            instruction['window'] == window
+                            for instruction in self.model.instructions.values())
+        if bank is not None:
+            return {'window': window, 'page': mapping.bank_pages[bank]}
+        if not banked_window:
+            return {}
+        if window not in self.resident_pages:
+            raise SessionError('Физическая страница resident window ещё не установлена')
+        return {'window': window, 'page': self.resident_pages[window]}
+
+    def _install(self, locations: list[dict], enabled: bool = True) -> dict:
+        if not locations:
+            raise SessionError('Для точки остановки нет исполняемых адресов')
+        mapping = self.refresh()
+        ids = []
+        conditions = []
+        try:
+            for location in locations:
+                arguments = self._guard(location, mapping)
+                if not enabled:
+                    arguments['enabled'] = False
+                result = self.bridge.request(
+                    'breakpoint', address=location['logical_address'],
+                    **arguments)
+                ids.append(result['id'])
+                conditions.append(result.get('condition', ''))
+        except BaseException:
+            for identifier in ids:
+                try:
+                    self.bridge.request('clear', id=identifier)
+                except BaseException:
+                    pass
+            raise
+        logical = self._next_breakpoint
+        self._next_breakpoint += 1
+        self.breakpoints[logical] = ids
+        return {'id': logical, 'backend_ids': ids, 'conditions': conditions,
+                'locations': locations}
+
+    def break_line(self, filename: str, line: int, enabled: bool = True) -> dict:
+        resolved = self.model.line_locations(filename, line)
+        if resolved['stale_source']:
+            raise SessionError('Исходник изменён после сборки; пересоберите программу')
+        return self._install(resolved['locations'], enabled)
+
+    def break_function(self, name: str, enabled: bool = True) -> dict:
+        locations = [function for function in self.model.functions
+                     if function['name'] == name]
+        return self._install(locations, enabled)
+
+    def break_anchor(self, location: dict, enabled: bool = True) -> dict:
+        if not location.get('verified') or location.get('source') in self.model.stale_sources:
+            raise SessionError('SDBG_LOG-якорь не проверен или исходник устарел')
+        return self._install([location], enabled)
+
+    def activate_breakpoints(self) -> dict:
+        return self.bridge.request('activate_breakpoints')
+
+    def deactivate_breakpoints(self) -> dict:
+        return self.bridge.request('deactivate_breakpoints')
+
+    def clear_breakpoint(self, identifier: int) -> dict:
+        ids = self.breakpoints.pop(identifier, None)
+        if ids is None:
+            raise SessionError('Неизвестная логическая точка остановки')
+        for backend_id in ids:
+            self.bridge.request('clear', id=backend_id)
+        return {'cleared': identifier, 'backend_ids': ids}
+
+    def read_variable(self, name: str, module: str | None = None) -> dict:
+        matches = [variable for variable in self.model.variables
+                   if variable['name'] == name and
+                   (module is None or variable['module'] == module)]
+        if len(matches) != 1:
+            raise SessionError('Переменная не найдена или имя неоднозначно')
+        variable = matches[0]
+        if not variable['supported'] or variable['size'] not in (1, 2, 4):
+            raise SessionError('Тип переменной пока не поддерживается')
+        mapping = self.refresh()
+        if variable['bank'] is not None:
+            if self._page(mapping, variable['window']) != mapping.bank_pages[variable['bank']]:
+                raise SessionError('Банк переменной сейчас не отображён')
+        else:
+            bank_pages = set(mapping.bank_pages.values())
+            if self._page(mapping, variable['window']) in bank_pages:
+                raise SessionError('Resident-страница переменной сейчас закрыта банком')
+        data = bytes.fromhex(self.bridge.request(
+            'memory', address=variable['logical_address'], length=variable['size'])['hex'])
+        value = int.from_bytes(data, 'little', signed=variable['signed'])
+        return {**variable, 'value': value, 'hex': data.hex()}