#!/usr/bin/env python3 """ elm_relay.py — Интерактивная консоль для удалённого управления ELM327 через Android. Работает через HTTP-очередь на сервере: 1. Ставит команду → POST /api/v1/elm/raw/cmd 2. Ждёт ответ → GET /api/v1/elm/raw/response?wait=N 3. Показывает сырой ответ 4. Анализирует → следующая команда ЗАПУСК: python3 tools/elm_relay.py # сервер по умолчанию http://localhost:5005 python3 tools/elm_relay.py --server https://obdai.ru python3 tools/elm_relay.py --timeout 1000 # таймаут команд 1000мс ИНТЕРАКТИВНЫЕ КОМАНДЫ: ATZ — отправить "ATZ" 0105 — отправить PID !status — статус устройства !history [N] — последние N ответов !drain — очистить буфер (ATPC) !mode raw — включить raw-режим на сервере !mode normal — выключить !timeout N — таймаут команд (мс) !help — справка !quit — выход """ import argparse import cmd import json import sys import time import urllib.request import urllib.error DEFAULT_SERVER = "http://localhost:5005" class ElmRelay(cmd.Cmd): """Интерактивная консоль для удалённого ELM327.""" intro = """ ╔══════════════════════════════════════════════════════╗ ║ ELM327 Remote Relay Console ║ ║ Сервер → Android → ELM327 → ответ → анализ ║ ║ !help для списка команд ║ ╚══════════════════════════════════════════════════════╝ """ prompt = "\nelm-relay> " def __init__(self, server: str, timeout: int): super().__init__() self.server = server.rstrip("/") self.timeout = timeout self._last_seq = 0 # ── Отправка команд ────────────────────────────── def default(self, line: str): """Любая не-! команда → отправить в ELM327.""" cmd_str = line.strip() if not cmd_str: return if cmd_str.startswith("!"): print(f"Неизвестная команда: {cmd_str}") return self._send_and_wait(cmd_str) def _send_and_wait(self, cmd: str, drain_first: bool = False): """Поставить команду в очередь и дождаться ответа.""" # 1. Отправить команду try: enq = self._post("/api/v1/elm/raw/cmd", { "cmd": cmd, "timeout_ms": self.timeout, "drain_first": drain_first, }) except Exception as e: print(f"❌ Ошибка отправки: {e}") return seq = enq.get("seq", 0) print(f"→ {cmd} (seq={seq}, timeout={self.timeout}ms)") # 2. Ждать ответ try: resp = self._get(f"/api/v1/elm/raw/response?wait=30&seq={self._last_seq}") except Exception as e: print(f"❌ Ошибка ожидания: {e}") return self._last_seq = resp.get("seq", seq) # 3. Показать self._print_response(resp) # ── Вывод ответа ────────────────────────────────── def _print_response(self, r: dict): raw = r.get("raw", "") prompt = r.get("prompt", False) elapsed = r.get("elapsed_ms", 0) nbytes = r.get("bytes", 0) error = r.get("error") print(f"← {raw!r}" if raw else "← (пусто)") status = [] if prompt: status.append("✅ >") else: status.append("❌ нет >") status.append(f"{elapsed}ms") status.append(f"{nbytes}B") if error: status.append(f"⚠️ {error}") print(f" {' | '.join(status)}") # ── Специальные команды ─────────────────────────── def do_status(self, arg): """!status — статус устройства.""" try: s = self._get("/api/v1/elm/raw/status") except Exception as e: print(f"❌ {e}") return print(f"Устройство: {'✅ готово' if s.get('device_ready') else '❌ не подключено'}") info = s.get("device_info", {}) if info: print(f" ID: {info.get('device_id', '?')}") print(f" ELM: {info.get('elm_version', '?')}") print(f" Протокол: {info.get('protocol', '?')}") print(f" Напряжение: {info.get('voltage', '?')}") print(f"Очередь: {'есть команда' if s.get('pending_cmd') else 'пусто'}") print(f"Последний seq: {s.get('last_response_seq', 0)}") print(f"История: {s.get('history_count', 0)} команд") def do_history(self, arg): """!history [N] — последние N ответов.""" try: n = int(arg.strip()) if arg.strip() else 10 except ValueError: n = 10 try: h = self._get(f"/api/v1/elm/raw/history?n={n}") except Exception as e: print(f"❌ {e}") return items = h.get("history", []) if not items: print("📭 История пуста.") return print(f"Последние {len(items)} из {h.get('total', 0)}:\n") for r in items: seq = r.get("seq", "?") cmd = r.get("cmd", "?") raw = (r.get("raw") or "")[:60] elapsed = r.get("elapsed_ms", 0) prompt = "✅>" if r.get("prompt") else "❌" error = f" ⚠️{r['error']}" if r.get("error") else "" print(f" #{seq} → {cmd} ← {raw}{'…' if len(r.get('raw',''))>60 else ''} ({elapsed}ms, {prompt}){error}") def do_drain(self, arg): """!drain — попросить Android очистить буфер ELM327.""" self._send_and_wait("ATPC", drain_first=False) print("🗑 Буфер очищен.") def do_mode(self, arg): """!mode raw|normal — переключить режим сервера.""" mode = arg.strip().lower() if mode not in ("raw", "normal"): print("❌ !mode raw или !mode normal") return on = mode == "raw" try: self._post("/api/v1/elm/raw/mode", {"raw_mode": on}) print(f"✅ Режим: {'RAW' if on else 'NORMAL'}") except Exception as e: print(f"❌ {e}") def do_timeout(self, arg): """!timeout N — таймаут команд (мс).""" try: self.timeout = int(arg.strip()) print(f"⏱ Таймаут: {self.timeout}ms") except ValueError: print(f"❌ Нужно число: !timeout 1000") def do_help(self, arg): """!help — справка.""" print(""" ╔══════════════════════════════════════════════════════╗ ║ ELM327 КОМАНДЫ (вводи как есть): ║ ║ ATZ ATI ATE0 ATL0 ATS0 ║ ║ ATH1 ATSP0 ATRV ATDPN ATSTxx ║ ║ 0105 (ОЖ) 010C (RPM) 010D (Speed) ║ ║ 03 (DTC) 07 (pending) 0902 (VIN) ║ ║ ║ ║ КОНСОЛЬ: ║ ║ !status — статус устройства ║ ║ !history [N] — последние ответы ║ ║ !drain — очистить буфер ║ ║ !mode raw — включить raw-режим ║ ║ !timeout N — таймаут команд ║ ║ !help — эта справка ║ ║ !quit — выход ║ ╚══════════════════════════════════════════════════════╝ """) def do_quit(self, arg): print("👋") return True def do_exit(self, arg): return self.do_quit(arg) do_q = do_quit do_h = do_help do_s = do_status # ── HTTP-хелперы ────────────────────────────────── def _get(self, path: str) -> dict: url = f"{self.server}{path}" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=35) as resp: return json.loads(resp.read().decode()) def _post(self, path: str, data: dict) -> dict: url = f"{self.server}{path}" body = json.dumps(data).encode() req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) # ── main ───────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="ELM327 Remote Relay Console — удалённое управление через Android" ) parser.add_argument("--server", default=DEFAULT_SERVER, help=f"URL сервера (default: {DEFAULT_SERVER})") parser.add_argument("--timeout", type=int, default=500, help="Таймаут команд ms (default: 500)") args = parser.parse_args() console = ElmRelay(server=args.server, timeout=args.timeout) # Проверим связь с сервером try: s = console._get("/api/v1/elm/raw/status") ready = s.get("device_ready", False) print(f"Сервер: {args.server}") print(f"Устройство: {'✅ готово' if ready else '❌ не подключено (запусти Android-приложение)'}") except Exception as e: print(f"⚠️ Сервер недоступен: {e}") print(f" Проверь: curl {args.server}/api/v1/ping") try: console.cmdloop() except KeyboardInterrupt: print("\n👋") if __name__ == "__main__": main()