"""Постоянный владелец DebugSession и локальный JSON-RPC для адаптеров.""" from __future__ import annotations from collections import deque import json from pathlib import Path import socket import socketserver import string import threading import time import uuid from .session import DebugSession, SessionError from .transport import BridgeError from .macros import validate_log_message class SessionController: CONTROL_TTL = 30.0 def __init__(self, session: DebugSession): self.session = session self.lock = threading.RLock() self.changed = threading.Condition() self.events = deque(maxlen=1024) self.dropped_events = 0 self.sequence = 0 self.running = False self.source_step = None self.closed = False self.session_id = uuid.uuid4().hex self.control_owner: str | None = None self.control_deadline = 0.0 self.owner_deadlines: dict[str, float] = {} self.held_inputs: dict[str, set[tuple[str, int]]] = {} self.source_breakpoints: dict[str, list[int]] = {} self.function_breakpoints: list[int] = [] self.breakpoint_info: dict[int, dict] = {} self.attached = session.attach() self.mame_console = bool(self.attached.get('capabilities', {}).get('console_print')) self._emit('stopped', {'reason': 'entry', 'location': self.attached['location']}) self._install_macro_logs() self.poller = threading.Thread(target=self._poll, name='sdbg-poller', daemon=True) self.poller.start() def _install_macro_logs(self) -> None: """Авторские точки принадлежат пакету сборки, не DAP source-набору.""" created = [] try: for macro in getattr(self.session.model, 'logpoints', []): if not macro['verified'] or macro['source'] in self.session.model.stale_sources: self._emit('output', {'category': 'stderr', 'output': 'sdbg: SDBG_LOG ' + macro['tag'] + ' не активирован: ' + str(macro.get('reason') or 'устаревший исходник') + '\n'}) continue self._validate_log_message(macro['message']) item = self.session.break_anchor(macro, enabled=False) created.append(item['id']) self.breakpoint_info[item['id']] = { 'kind': 'log', 'message': macro['message'], 'condition': macro['condition'], 'module': macro['module'], 'tag': macro['tag'], 'locations': item['locations'], 'hits': 0, 'owner': 'build', 'guard_conditions': item['conditions'], } if created: self.session.activate_breakpoints() except BaseException: for identifier in created: self.breakpoint_info.pop(identifier, None) try: self.session.clear_breakpoint(identifier) except BaseException: pass raise def _emit(self, name: str, body: dict) -> None: with self.changed: if len(self.events) == self.events.maxlen: self.dropped_events += 1 self.sequence += 1 self.events.append({'seq': self.sequence, 'event': name, 'body': body}) self.changed.notify_all() def _poll(self) -> None: next_idle_snapshot = 0.0 while not self.closed: if not self.running: delay = next_idle_snapshot - time.monotonic() if delay > 0: time.sleep(min(delay, .05)) continue try: with self.lock: snapshot = self.session.bridge.request('snapshot') if snapshot['state'] == 'stopped': if self.running: self._release_all_inputs() location = self.session.where(snapshot) if self.source_step is not None: self._source_step_stopped(location) elif self._handle_logpoints(location): self.session.bridge.request('continue') continue else: self.running = False self._emit('stopped', {'reason': 'breakpoint', 'location': location}) elif snapshot['state'] == 'running' and not self.running: # Команда из родного окна MAME тоже меняет состояние CPU. self.running = True self._emit('continued', {'reason': 'external'}) elif snapshot['state'] == 'invalidated': self.running = False self.source_step = None self.closed = True self._emit('invalidated', {'reason': 'reset_or_load'}) if not self.running: next_idle_snapshot = time.monotonic() + .5 self._reap_owners() except (BridgeError, SessionError, ValueError, OSError) as error: self.running = False self.source_step = None self.closed = True self._emit('invalidated', {'reason': str(error)}) time.sleep(.01) def close(self) -> None: self.closed = True self.poller.join(timeout=1) with self.lock: try: self._release_all_inputs() except (BridgeError, SessionError, OSError): pass self.session.bridge.close() @staticmethod def _owner(arguments: dict) -> str: owner = arguments.get('owner', 'cli') if not isinstance(owner, str) or not owner or len(owner) > 128: raise SessionError('owner должен быть непустой строкой до 128 символов') return owner def _control(self, arguments: dict) -> dict: """Захватить или продлить управление; чужой живой lease не отбирать.""" owner = self._owner(arguments) self._touch_owner(owner) now = time.monotonic() if self.control_owner is not None and self.control_deadline <= now: self.control_owner = None if self.control_owner not in (None, owner): raise SessionError(f'CPU управляет {self.control_owner}; дождитесь release или истечения lease') if self.control_owner != owner: self.control_owner = owner self._emit('control', {'owner': owner}) self.control_deadline = now + self.CONTROL_TTL return {'owner': owner, 'expires_in': self.CONTROL_TTL} def _touch_owner(self, owner: str) -> None: if owner.startswith('mcp:'): self.owner_deadlines[owner] = time.monotonic() + self.CONTROL_TTL def _release_inputs(self, owner: str) -> None: for tag, mask in self.held_inputs.pop(owner, set()): self.session.bridge.request('key', tag=tag, mask=mask, down=False) def _release_all_inputs(self) -> None: for owner in list(self.held_inputs): self._release_inputs(owner) def _reap_owners(self) -> None: now = time.monotonic() for owner, deadline in list(self.owner_deadlines.items()): if deadline > now: continue self._release_inputs(owner) identifiers = [identifier for identifier, info in self.breakpoint_info.items() if info.get('owner') == owner] for identifier in identifiers: self.session.clear_breakpoint(identifier) self.breakpoint_info.pop(identifier, None) if self.control_owner == owner: self.control_owner = None self.control_deadline = 0.0 self._emit('control', {'owner': None}) self.owner_deadlines.pop(owner, None) self._emit('owner_expired', {'owner': owner, 'cleared_breakpoints': identifiers}) def call(self, method: str, arguments: dict) -> object: if method == 'events': owner = arguments.get('owner') if owner is not None: with self.lock: if self.control_owner == self._owner(arguments) and \ self.control_deadline > time.monotonic(): self.control_deadline = time.monotonic() + self.CONTROL_TTL after = int(arguments.get('after', 0)) timeout = min(max(float(arguments.get('timeout', 0)), 0), 30) deadline = time.monotonic() + timeout with self.changed: while self.sequence <= after and not self.closed and time.monotonic() < deadline: self.changed.wait(deadline - time.monotonic()) return {'events': [event for event in self.events if event['seq'] > after], 'first': self.events[0]['seq'] if self.events else self.sequence + 1, 'last': self.sequence, 'lost': self.dropped_events, 'closed': self.closed} with self.lock: if self.closed: raise SessionError('Сессия закрыта или инвалидирована') if method == 'claim_control': return self._control(arguments) if method == 'renew_control': owner = self._owner(arguments) self._touch_owner(owner) if self.control_owner == owner and self.control_deadline > time.monotonic(): self.control_deadline = time.monotonic() + self.CONTROL_TTL return {'owner': owner, 'has_control': True, 'expires_in': self.CONTROL_TTL} return {'owner': owner, 'has_control': False} if method == 'release_control': owner = self._owner(arguments) self._release_inputs(owner) if self.control_owner == owner: self.control_owner = None self.control_deadline = 0.0 self._emit('control', {'owner': None}) return {'released': owner} if self.running and method in ('where', 'registers', 'read_memory', 'read_program_memory', 'disassemble_logical', 'read_variable', 'break_line', 'break_function'): raise SessionError('Операция требует остановленного CPU; сначала Pause') if method == 'status': return {**self.attached, 'running': self.running, 'generation': getattr(self.session.bridge, 'generation', self.attached.get('generation')), 'event_sequence': self.sequence, 'session_id': self.session_id, 'control_owner': self.control_owner if self.control_deadline > time.monotonic() else None} if method == 'list_breakpoints': breakpoints = [] for identifier, info in sorted(self.breakpoint_info.items()): guards = info.get('guard_conditions', []) locations = [] for index, location in enumerate(info['locations']): locations.append({**location, 'bank_guard': guards[index] if index < len(guards) else ''}) breakpoints.append({'id': identifier, 'owner': info.get('owner', 'dap'), 'kind': info['kind'], 'hits': info['hits'], 'locations': locations, 'tag': info.get('tag')}) return {'breakpoints': breakpoints} if method == 'snapshot': return self.session.bridge.request('snapshot') if method == 'mame_console_tail': return self.session.bridge.request( 'console_tail', count=int(arguments.get('count', 40))) if method == 'list_ports': return self.session.bridge.request('list_ports') if method == 'list_shares': return self.session.bridge.request('list_shares') if method in ('read_share', 'read_vram'): address = int(arguments['address']) length = int(arguments['length']) if address < 0 or address > 0x10000000 or length < 1 or length > 4096: raise SessionError('Share: адрес 0..0x10000000, длина 1..4096') if method == 'read_share': tag = arguments['tag'] if not isinstance(tag, str) or not tag or len(tag) > 128: raise SessionError('Нужен точный tag share до 128 символов') return self.session.bridge.request('read_share', tag=tag, address=address, length=length) return self.session.bridge.request('read_vram', address=address, length=length) if method == 'read_screen_pixels': x, y = int(arguments['x']), int(arguments['y']) width, height = int(arguments['width']), int(arguments['height']) if x < 0 or y < 0 or x > 4095 or y > 4095 or \ width < 1 or height < 1 or width > 512 or height > 512 or \ width * height > 8192: raise SessionError('Экран: координаты 0..4095, размер 1..512, максимум 8192 пикселей') return self.session.bridge.request('screen_pixels', x=x, y=y, width=width, height=height) if method == 'screenshot': result = self.session.bridge.request('screen_snapshot') root = (Path(self.session.bridge.directory).parent / 'snapshot').resolve() path = Path(result['path']).resolve() if path.parent != root or path.suffix.lower() != '.png' or \ not path.is_file() or path.stat().st_size != result['size'] or \ result['size'] > 8 * 1024 * 1024: raise SessionError('Снимок MAME вне каталога сессии или повреждён') return result if method == 'input_key': if not self.running and arguments['down'] is not False: raise SessionError('Нажатие возможно только при running CPU') self._control(arguments) tag, mask = arguments['tag'], int(arguments['mask']) result = self.session.bridge.request('key', tag=tag, mask=mask, down=arguments['down']) held = self.held_inputs.setdefault(self._owner(arguments), set()) if arguments['down']: held.add((tag, mask)) else: held.discard((tag, mask)) return result if method == 'where': return self.session.where() if method == 'registers': mapping = self.session.refresh() return {'generation': mapping.generation, 'registers': mapping.registers, 'bank_pages': mapping.bank_pages} if method == 'read_memory': address = int(arguments['address']) length = int(arguments['length']) if address < 0 or address > 0xffff or length < 1 or length > 256 or \ address + length > 0x10000: raise SessionError('Чтение памяти: адрес 0..65535, длина 1..256 без выхода за 64 КБ') mapping = self.session.refresh() data = self.session.bridge.request('memory', address=address, length=length) return {'address': address, 'length': length, 'hex': data['hex'], 'generation': mapping.generation, 'bank_pages': mapping.bank_pages} if method == 'read_program_memory': address = int(arguments['address']) length = int(arguments['length']) if address < 0 or address > 0x3ffff or length < 1 or length > 4096 or \ address + length > 0x40000: raise SessionError('Raw program: адрес 0..0x3ffff, длина 1..4096') mapping = self.session.refresh() result = self.session.bridge.request('program_memory', address=address, length=length) return {**result, 'bank_pages': mapping.bank_pages} if method == 'disassemble_logical': address = int(arguments['address']) length = int(arguments['length']) if address < 0 or address > 0xffff or length < 1 or length > 256 or \ address + length > 0x10000: raise SessionError('Дизассемблирование: адрес 0..0xffff, длина 1..256') mapping = self.session.refresh() result = self.session.bridge.request('disassemble_logical', address=address, length=length) return {**result, 'bank_pages': mapping.bank_pages} if method == 'variables': return self.session.model.variables if method == 'read_variable': return self.session.read_variable(arguments['name'], arguments.get('module')) if method == 'break_line': owner = self._owner(arguments) self._touch_owner(owner) item = self.session.break_line(arguments['file'], int(arguments['line'])) self.breakpoint_info[item['id']] = { 'kind': 'stop', 'message': None, 'locations': item['locations'], 'hits': 0, 'owner': owner, 'guard_conditions': item['conditions']} return item if method == 'break_function': owner = self._owner(arguments) self._touch_owner(owner) item = self.session.break_function(arguments['name']) self.breakpoint_info[item['id']] = { 'kind': 'stop', 'message': None, 'locations': item['locations'], 'hits': 0, 'owner': owner, 'guard_conditions': item['conditions']} return item if method == 'clear_breakpoint': identifier = int(arguments['id']) owner = arguments.get('owner') if identifier in self.breakpoint_info and owner is not None and \ self.breakpoint_info[identifier].get('owner') != owner: raise SessionError('Точка не принадлежит этому клиенту') result = self.session.clear_breakpoint(identifier) self.breakpoint_info.pop(identifier, None) return result if method == 'clear_owned_breakpoints': owner = arguments.get('owner') if not isinstance(owner, str) or not owner: raise SessionError('Для очистки точек требуется owner') identifiers = [identifier for identifier, info in self.breakpoint_info.items() if info.get('owner') == owner] for identifier in identifiers: self.session.clear_breakpoint(identifier) self.breakpoint_info.pop(identifier, None) return {'cleared': identifiers} if method == 'set_source_breakpoints': specifications = arguments.get('breakpoints') if specifications is None: specifications = [{'line': line} for line in arguments.get('lines', [])] return self._set_source_breakpoints(arguments['file'], specifications) if method == 'set_function_breakpoints': return self._set_function_breakpoints(arguments.get('names', [])) if method == 'continue': if self.running: raise SessionError('CPU уже выполняется; сначала Pause') self._control(arguments) result = self.session.bridge.request('continue') self.running = True self._emit('continued', {}) return result if method == 'pause': self._control(arguments) self.source_step = None self.session.bridge.request('pause') location = self.session.where(self.session.bridge.wait_stopped()) self.running = False self._emit('stopped', {'reason': 'pause', 'location': location}) return location if method in ('step_over_instruction', 'step_out_instruction'): if self.running: raise SessionError('CPU уже выполняется; сначала Pause') command = {'step_over_instruction': 'step_over', 'step_out_instruction': 'step_out'}[method] request_args = {} if method != 'step_out_instruction': count = arguments.get('count', 1) if type(count) is not int or count < 1 or count > 64: raise SessionError('Число машинных шагов должно быть 1..64') request_args['count'] = count self._control(arguments) self.session.bridge.request(command, **request_args) self.running = True self._emit('continued', {'reason': command}) return {'accepted': True, 'command': command} if method == 'step': if self.running: raise SessionError('CPU уже выполняется; сначала Pause') count = arguments.get('count', 1) if type(count) is not int or count < 1 or count > 64: raise SessionError('Число машинных шагов должно быть 1..64') self._control(arguments) self.session.bridge.request('step', count=count) self.running = True self._emit('continued', {'reason': 'step'}) location = self.session.where(self.session.bridge.wait_stopped()) self.running = False self._emit('stopped', {'reason': 'step', 'location': location}) return location if method == 'source_step': self._control(arguments) return self._source_step(arguments.get('kind', 'into')) raise SessionError('Неизвестный RPC-метод: ' + method) @staticmethod def _source_identity(location: dict) -> frozenset[tuple[str, int]]: return frozenset((item['file'], int(item['line'])) for item in location.get('sources', [])) def _has_stop_breakpoint(self, location: dict) -> bool: address = location.get('link_address') return any(info['kind'] == 'stop' and any(item.get('link_address') == address for item in info['locations']) for info in self.breakpoint_info.values()) def _source_step(self, kind: str) -> dict: commands = {'into': 'step', 'over': 'step_over', 'out': 'step_out'} if kind not in commands: raise SessionError('Неизвестный вид source step') if self.running: raise SessionError('CPU уже выполняется; сначала Pause') start = self.session.where() if start.get('stale_source'): raise SessionError('Исходник изменён после сборки; пересоберите программу') initial = self._source_identity(start) if not initial: raise SessionError('Текущий PC не имеет проверенной C-позиции') self.source_step = { 'initial': initial, 'kind': kind, 'command': commands[kind], 'instructions': 1, 'location': start, } try: self.session.bridge.request(commands[kind]) except BaseException: self.source_step = None raise if kind == 'out': # Первый out выходит из машинного frame. У банкового вызова # дальше идём через over до первой позиции C вызывающей функции. self.source_step['command'] = 'step_over' self.running = True self._emit('continued', {'reason': 'step'}) return {'accepted': True} def _source_step_stopped(self, location: dict) -> None: step = self.source_step self._handle_logpoints(location) reason = None if self._has_stop_breakpoint(location): reason = 'breakpoint' elif self._source_identity(location) and self._source_identity(location) != step['initial']: reason = 'step' elif step['instructions'] >= 512: reason = 'step' self._emit('output', {'category': 'stderr', 'output': 'sdbg: source step достиг лимита; CPU остановлен\n'}) if reason is not None: self.source_step = None self.running = False self._emit('stopped', {'reason': reason, 'location': location}) return # Машинный over может ждать клавишу сколь угодно долго. Пока CPU # выполняется, этот автомат не занимает session lock и не ставит # таймер; пользователь может направить ввод в MAME или нажать Pause. self.session.bridge.request(step['command']) step['instructions'] += 1 step['location'] = location def _set_source_breakpoints(self, filename: str, specifications) -> dict: filename = str(Path(filename).resolve()) requested = [] for value in specifications: if not isinstance(value, dict): raise SessionError('Описание breakpoint должно быть объектом') line = int(value['line']) if line <= 0 or any(item['line'] == line for item in requested): raise SessionError('Номер строки должен быть положительным и уникальным') if value.get('condition') or value.get('hitCondition'): raise SessionError('Условия и hitCondition пока не поддержаны') message = value.get('logMessage') if message is not None: self._validate_log_message(message) requested.append({'line': line, 'logMessage': message}) created = [] results = [] try: for specification in requested: line = specification['line'] item = self.session.break_line(filename, line, enabled=False) created.append(item['id']) kind = 'log' if specification['logMessage'] is not None else 'stop' self.breakpoint_info[item['id']] = { 'kind': kind, 'message': specification['logMessage'], 'locations': item['locations'], 'hits': 0, 'owner': 'dap', 'guard_conditions': item['conditions'], } results.append({'line': line, 'verified': True, 'logMessage': specification['logMessage'], **item}) except BaseException: for identifier in created: self.breakpoint_info.pop(identifier, None) try: self.session.clear_breakpoint(identifier) except BaseException: pass raise previous = self.source_breakpoints.get(filename, []) for identifier in previous: self.breakpoint_info.pop(identifier, None) self.session.clear_breakpoint(identifier) self.source_breakpoints[filename] = created self.session.activate_breakpoints() return {'file': filename, 'breakpoints': results} def _set_function_breakpoints(self, names) -> dict: requested = [] for value in names: name = str(value) if not name or name in requested: raise SessionError('Имя функции должно быть непустым и уникальным') requested.append(name) created = [] results = [] try: for name in requested: item = self.session.break_function(name, enabled=False) created.append(item['id']) self.breakpoint_info[item['id']] = { 'kind': 'stop', 'message': None, 'locations': item['locations'], 'hits': 0, 'owner': 'dap', 'guard_conditions': item['conditions']} results.append({'name': name, 'verified': True, **item}) except BaseException: for identifier in created: self.breakpoint_info.pop(identifier, None) try: self.session.clear_breakpoint(identifier) except BaseException: pass raise for identifier in self.function_breakpoints: self.breakpoint_info.pop(identifier, None) self.session.clear_breakpoint(identifier) self.function_breakpoints = created self.session.activate_breakpoints() return {'breakpoints': results} @staticmethod def _validate_log_message(message: str) -> None: try: validate_log_message(message) except ValueError as error: raise SessionError(str(error)) from error def _read_log_variable(self, name: str, module: str | None) -> dict: if module is None: return self.session.read_variable(name) candidates = [item for item in self.session.model.variables if item['name'] == name and item['module'] in (None, module)] if len(candidates) != 1: raise SessionError('Переменная не найдена или имя неоднозначно') return self.session.read_variable(name, candidates[0]['module']) def _render_log_message(self, message: str, module: str | None = None) -> str: output = [] for literal, name, _, _ in string.Formatter().parse(message): output.append(literal) if name is not None: try: output.append(str(self._read_log_variable(name, module)['value'])) except SessionError as error: output.append('') return ''.join(output) def _handle_logpoints(self, location: dict) -> bool: address = location.get('link_address') matched_logs = [] matched_any_log = False matched_stop = False for info in self.breakpoint_info.values(): if not any(item.get('link_address') == address for item in info['locations']): continue if info['kind'] == 'stop': matched_stop = True else: matched_any_log = True condition = info.get('condition') if condition is not None: try: if not self._read_log_variable(condition, info.get('module'))['value']: continue except SessionError as error: if not info.get('condition_warned'): info['condition_warned'] = True self._emit('output', {'category': 'stderr', 'output': 'sdbg: SDBG_LOGIF ' + info.get('tag', '') + ': ' + str(error) + '\n'}) continue info['hits'] += 1 matched_logs.append(info) for info in matched_logs: rendered = self._render_log_message(info['message'], info.get('module')) self._emit('output', {'category': 'console', 'output': rendered + '\n', 'location': location, 'hit': info['hits'], 'tag': info.get('tag')}) if self.mame_console: try: self.session.bridge.request('console_print', text=rendered) except (BridgeError, OSError) as error: if not info.get('console_warned'): info['console_warned'] = True self._emit('output', {'category': 'stderr', 'output': 'sdbg: MAME console: ' + str(error) + '\n'}) return matched_any_log and not matched_stop class _ThreadedUnixServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): daemon_threads = True RPC_MUTATIONS = frozenset({ 'claim_control', 'renew_control', 'release_control', 'input_key', 'break_line', 'break_function', 'clear_breakpoint', 'clear_owned_breakpoints', 'set_source_breakpoints', 'set_function_breakpoints', 'continue', 'pause', 'step', 'step_over_instruction', 'step_out_instruction', 'source_step', }) RPC_GENERATION_MUTATIONS = RPC_MUTATIONS - { 'claim_control', 'renew_control', 'release_control', } class SessionRpcServer: def __init__(self, path, controller: SessionController): self.path = Path(path) self.controller = controller self.path.parent.mkdir(parents=True, exist_ok=True) if self.path.exists(): try: with socket.socket(socket.AF_UNIX) as probe: probe.connect(str(self.path)) except OSError: self.path.unlink() else: raise OSError('RPC socket уже занят: ' + str(self.path)) controller_ref = controller class Handler(socketserver.StreamRequestHandler): def handle(self): raw = self.rfile.readline(1_048_577) response = {'id': None, 'ok': False} try: if len(raw) > 1_048_576: raise ValueError('RPC-запрос слишком велик') request = json.loads(raw) if not isinstance(request, dict) or not isinstance(request.get('method'), str): raise ValueError('Неверная структура RPC-запроса') arguments = request.get('arguments', {}) if not isinstance(arguments, dict): raise ValueError('RPC arguments должен быть объектом') method = request['method'] with controller_ref.lock: if method != 'unknown' and (method != 'status' or request.get('session_id') is not None or request.get('build_id') is not None): if request.get('session_id') != controller_ref.session_id or \ request.get('build_id') != controller_ref.attached['build_id']: raise SessionError('Устаревшая RPC-сессия или build ID; переподключитесь') if method in RPC_GENERATION_MUTATIONS and \ request.get('generation') != \ getattr(controller_ref.session.bridge, 'generation', None): raise SessionError('Устаревшая generation; обновите статус сессии') if method != 'events': result = controller_ref.call(method, arguments) result_generation = getattr(controller_ref.session.bridge, 'generation', None) if method == 'events': result = controller_ref.call(method, arguments) result_generation = getattr(controller_ref.session.bridge, 'generation', None) response = {'id': request.get('id'), 'ok': True, 'session_id': controller_ref.session_id, 'build_id': controller_ref.attached['build_id'], 'generation': result_generation, 'result': result} except (BridgeError, SessionError, ValueError, TypeError, OSError, KeyError) as error: response.update(error=str(error)) self.wfile.write((json.dumps(response, ensure_ascii=False) + '\n').encode()) self.server = _ThreadedUnixServer(str(self.path), Handler) self.path.chmod(0o600) def serve_forever(self): try: self.server.serve_forever(poll_interval=.1) finally: self.close() def close(self): self.server.server_close() self.path.unlink(missing_ok=True) def rpc_call(path, method: str, arguments=None, timeout=10, *, session_id=None, build_id=None, generation=None): request = {'id': 1, 'method': method, 'arguments': arguments or {}, 'session_id': session_id, 'build_id': build_id, 'generation': generation} with socket.socket(socket.AF_UNIX) as client: client.settimeout(timeout) client.connect(str(path)) client.sendall((json.dumps(request, ensure_ascii=False) + '\n').encode()) file = client.makefile('rb') raw = file.readline(1_048_577) if not raw: raise SessionError('RPC server закрыл соединение без ответа') response = json.loads(raw) if not response.get('ok'): raise SessionError(response.get('error', 'Неизвестная RPC-ошибка')) if method != 'status' and (response.get('session_id') != session_id or response.get('build_id') != build_id): raise SessionError('RPC-ответ от другой сессии или сборки') return response['result']