"""Высокоуровневая 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.get(identifier) if ids is None: raise SessionError('Неизвестная логическая точка остановки') backend_ids = list(ids) for backend_id in backend_ids: self.bridge.request('clear', id=backend_id) ids.remove(backend_id) self.breakpoints.pop(identifier) return {'cleared': identifier, 'backend_ids': backend_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()}