"""Честный минимальный Debug Adapter Protocol поверх постоянной sdbg-сессии.""" from __future__ import annotations from pathlib import Path from .server import rpc_call from .session import SessionError class DapEngine: def __init__(self, rpc=None): self.rpc_path = None self._rpc_override = rpc self.attached = None self.event_sequence = 0 def rpc(self, method, arguments=None, timeout=10): if self._rpc_override is not None: return self._rpc_override(method, arguments or {}) if not self.rpc_path: raise SessionError('DAP ещё не подключён к session server') return rpc_call(self.rpc_path, method, arguments, timeout) def handle(self, command: str, arguments: dict) -> tuple[dict, list[tuple[str, dict]]]: events = [] if command == 'initialize': body = { 'supportsConfigurationDoneRequest': True, 'supportsFunctionBreakpoints': True, 'supportsInstructionBreakpoints': False, 'supportsEvaluateForHovers': True, 'supportsSetVariable': False, 'supportsStepBack': False, 'supportsRestartRequest': False, 'supportsTerminateRequest': False, 'supportsDisassembleRequest': False, 'supportsSteppingGranularity': True, } return body, events if command == 'attach': if self._rpc_override is None: path = arguments.get('socket') if not isinstance(path, str) or not path: raise SessionError('В attach требуется socket session server') self.rpc_path = path self.attached = self.rpc('status') self.event_sequence = self.attached['event_sequence'] events.append(('initialized', {})) return {'buildId': self.attached['build_id']}, events if command == 'configurationDone': location = self.rpc('where') events.append(('stopped', {'reason': 'entry', 'threadId': 1, 'allThreadsStopped': True, 'description': self._description(location)})) return {}, events if command == 'disconnect': return {}, events if command == 'threads': return {'threads': [{'id': 1, 'name': 'Sprinter Z80'}]}, events if command == 'stackTrace': location = self.rpc('where') return {'stackFrames': [self._frame(location)], 'totalFrames': 1}, events if command == 'scopes': return {'scopes': [ {'name': 'Registers', 'variablesReference': 1, 'expensive': False}, {'name': 'Globals/statics', 'variablesReference': 2, 'expensive': False}, ]}, events if command == 'variables': reference = int(arguments.get('variablesReference', 0)) if reference == 1: values = self.rpc('registers')['registers'] variables = [{'name': name, 'value': f'{value:#x}', 'variablesReference': 0} for name, value in sorted(values.items())] elif reference == 2: variables = self._global_variables() else: variables = [] return {'variables': variables}, events if command == 'evaluate': expression = arguments.get('expression', '') if not expression.isidentifier(): raise SessionError('MVP evaluate принимает только имя переменной') value = self.rpc('read_variable', {'name': expression}) return {'result': str(value['value']), 'type': value['type'], 'variablesReference': 0, 'memoryReference': hex(value['link_address'])}, events if command == 'setBreakpoints': source = arguments.get('source', {}) path = source.get('path') if not path: raise SessionError('Для source breakpoint нужен полный path') specifications = [{name: item[name] for name in ('line','condition','hitCondition','logMessage') if name in item} for item in arguments.get('breakpoints', [])] result = self.rpc('set_source_breakpoints', {'file': path, 'breakpoints': specifications}) return {'breakpoints': [self._breakpoint(item) for item in result['breakpoints']]}, events if command == 'setFunctionBreakpoints': names = [item['name'] for item in arguments.get('breakpoints', [])] result = self.rpc('set_function_breakpoints', {'names': names}) return {'breakpoints': [self._breakpoint(item) for item in result['breakpoints']]}, events if command == 'continue': self.rpc('continue') return {'allThreadsContinued': True}, events if command == 'pause': self.rpc('pause') return {}, events if command in ('next', 'stepIn'): if arguments.get('granularity') == 'instruction': self.rpc('step') else: self.rpc('source_step', { 'kind': 'over' if command == 'next' else 'into'}) return {}, events if command == 'stepOut': self.rpc('source_step', {'kind': 'out'}) return {}, events raise SessionError('DAP-команда пока не поддерживается: ' + command) def poll_events(self, timeout=1): previous = self.event_sequence result = self.rpc('events', {'after': self.event_sequence, 'timeout': timeout}, timeout=timeout + 2) self.event_sequence = result['last'] translated = [] gap = max(0, result.get('first', previous + 1) - previous - 1) if gap: translated.append(('output', {'category': 'stderr', 'output': f'sdbg: пропущено {gap} событий журнала; ' 'уменьшите частоту logpoint\n'})) for event in result['events']: if event['event'] == 'stopped': translated.append(('stopped', {'reason': event['body'].get('reason', 'breakpoint'), 'threadId': 1, 'allThreadsStopped': True, 'description': self._description( event['body'].get('location', {}))})) elif event['event'] == 'continued': translated.append(('continued', {'threadId': 1, 'allThreadsContinued': True})) elif event['event'] == 'invalidated': translated.append(('terminated', {'restart': False})) elif event['event'] == 'output': translated.append(('output', {'category': event['body'].get('category','console'), 'output': event['body']['output']})) if result['closed'] and not any(name == 'terminated' for name, _ in translated): translated.append(('terminated', {'restart': False})) return translated, result['closed'] @staticmethod def _description(location): status = location.get('status', 'unknown') bank = location.get('function', {}).get('bank') if location.get('function') else None suffix = '' if bank is None else f', bank {bank}' return f'{status}{suffix}, PC={location.get("pc", 0):#x}' @staticmethod def _frame(location): function = location.get('function') or {} sources = location.get('sources') or [] source = sources[0] if sources else None frame = { 'id': 1, 'name': function.get('name', '') + (' [ambiguous]' if location.get('status') == 'ambiguous' else ''), 'line': source['line'] if source else 1, 'column': 1, 'instructionPointerReference': hex(location.get('link_address', location.get('pc', 0))), } if source: frame['source'] = {'name': Path(source['file']).name, 'path': source['file']} return frame @staticmethod def _breakpoint(item): locations = item.get('locations', []) first = locations[0] if locations else {} result = {'id': item['id'], 'verified': bool(item.get('verified', locations))} line = item.get('line', first.get('line')) if line is not None: result['line'] = line if first.get('link_address') is not None: result['instructionReference'] = hex(first['link_address']) if len(locations) > 1: result['message'] = f'Разрешено в {len(locations)} адресов/банков' return result def _global_variables(self): result = [] for variable in self.rpc('variables'): label = variable['name'] if variable.get('module'): label += '@' + variable['module'] try: value = self.rpc('read_variable', {'name': variable['name'], 'module': variable.get('module')}) text = str(value['value']) except SessionError as error: text = '' result.append({'name': label, 'value': text, 'type': variable['type'], 'variablesReference': 0, 'memoryReference': hex(variable['link_address'])}) return result