"""Тонкий MCP-клиент общей source-debug сессии без зависимости от MCP SDK.""" from __future__ import annotations from pathlib import Path import time import uuid from .server import RPC_GENERATION_MUTATIONS, rpc_call from .session import SessionError class McpSession: def __init__(self, socket_path: str, rpc=rpc_call): self.socket_path = str(Path(socket_path).expanduser()) self.owner = 'mcp:' + uuid.uuid4().hex self._rpc = rpc self.identity: dict[str, str] = {} def call(self, method: str, arguments: dict | None = None, timeout: float = 10): if method != 'status' and not self.identity: self.status() generation = None if method in RPC_GENERATION_MUTATIONS: generation = self.status().get('generation') identity = dict(self.identity) if method in RPC_GENERATION_MUTATIONS: identity['generation'] = generation result = self._rpc(self.socket_path, method, arguments or {}, timeout=timeout, **identity) if method == 'status' and 'session_id' in result: self.identity = {'session_id': result['session_id'], 'build_id': result['build_id']} return result def status(self): return self.call('status') def where(self): return self.call('where') def registers(self): return self.call('registers') def read_memory(self, address: str, length: int = 16): return self.call('read_memory', {'address': self._address(address), 'length': length}) @staticmethod def _address(address: str) -> int: try: return int(address, 16 if address.lower().startswith('0x') else 10) except (AttributeError, ValueError) as error: raise SessionError('Адрес памяти должен быть десятичным или 0xHEX') from error def list_shares(self): return self.call('list_shares') def list_ports(self): return self.call('list_ports') def read_program_memory(self, address: str, length: int = 16): return self.call('read_program_memory', {'address': self._address(address), 'length': length}) def read_share(self, tag: str, address: str, length: int = 16): return self.call('read_share', {'tag': tag, 'address': self._address(address), 'length': length}) def read_vram(self, address: str, length: int = 16): return self.call('read_vram', {'address': self._address(address), 'length': length}) def read_screen_pixels(self, x: int, y: int, width: int = 1, height: int = 1): return self.call('read_screen_pixels', {'x': x, 'y': y, 'width': width, 'height': height}) def screenshot(self): return self.call('screenshot') def press_key(self, key: str, frames: int = 3): """Один физический PC-key с отпусканием даже при stop/ошибке.""" from mame_interactive import SHIFT_KEY, resolve if not isinstance(key, str) or len(key) == 0 or len(key) > 16: raise SessionError('Нужна одна клавиша или enter/space/tab') if not isinstance(frames, int) or frames < 1 or frames > 60: raise SessionError('frames должен быть 1..60') named = {'enter': '\n', 'space': ' ', 'tab': '\t'} symbol = named.get(key.lower(), key) if len(symbol) != 1: raise SessionError('Пока поддерживаются один символ, enter, space, tab') try: tag, mask, shift = resolve(symbol) except ValueError as error: raise SessionError(str(error)) from error if not self.status()['running']: raise SessionError('Нажатие возможно только при running CPU') held = [] try: if shift: self.call('input_key', {'tag': SHIFT_KEY[0], 'mask': SHIFT_KEY[1], 'down': True, 'owner': self.owner}) held.append(SHIFT_KEY) self.call('input_key', {'tag': tag, 'mask': mask, 'down': True, 'owner': self.owner}) held.append((tag, mask)) start = self.call('snapshot')['frame'] deadline = time.monotonic() + 4 last = start while time.monotonic() < deadline: snapshot = self.call('snapshot') last = snapshot['frame'] if last - start >= frames or snapshot['state'] != 'running': break time.sleep(.01) else: raise SessionError('Клавиша не получила нужное число кадров за 4 с') return {'key': key, 'requested_frames': frames, 'observed_frames': max(0, last - start), 'stopped': snapshot['state'] != 'running'} finally: for release_tag, release_mask in reversed(held): self.call('input_key', {'tag': release_tag, 'mask': release_mask, 'down': False, 'owner': self.owner}) def variables(self): return self.call('variables') def read_variable(self, name: str, module: str | None = None): return self.call('read_variable', {'name': name, 'module': module}) def events(self, after: int = 0, timeout: float = 0): if after < 0 or timeout < 0 or timeout > 30: raise SessionError('after должен быть неотрицательным, timeout — от 0 до 30 с') return self.call('events', {'after': after, 'timeout': timeout, 'owner': self.owner}, timeout=max(10, timeout + 2)) def console_tail(self, count: int = 40): if count < 1 or count > 200: raise SessionError('count должен быть от 1 до 200') return self.call('mame_console_tail', {'count': count}) def break_line(self, file: str, line: int): return self.call('break_line', {'file': file, 'line': line, 'owner': self.owner}) def break_function(self, name: str): return self.call('break_function', {'name': name, 'owner': self.owner}) def clear_breakpoint(self, identifier: int): return self.call('clear_breakpoint', {'id': identifier, 'owner': self.owner}) def clear_owned_breakpoints(self): return self.call('clear_owned_breakpoints', {'owner': self.owner}) def continue_execution(self): return self.call('continue', {'owner': self.owner}) def pause_execution(self): return self.call('pause', {'owner': self.owner}) def step_instruction(self): return self.call('step', {'owner': self.owner}) def step_source(self, kind: str = 'into'): if kind not in ('into', 'over', 'out'): raise SessionError('kind должен быть into, over или out') return self.call('source_step', {'kind': kind, 'owner': self.owner}) def claim_control(self): return self.call('claim_control', {'owner': self.owner}) def renew_control(self): return self.call('renew_control', {'owner': self.owner}) def release_control(self): return self.call('release_control', {'owner': self.owner})