fix(mock): реалистичные задержки + нет приветствия при коннекте

- ATZ: ~2.5 сек (как реальный ELM)
- ATSP0: SEARCHING... + 1.8 сек
- OBD/PID: 0.2-0.5 сек
- ±30% случайный разброс для всех задержек
- Убрано авто-приветствие при подключении (реальный ELM не шлёт)
This commit is contained in:
Repinoid
2026-05-26 10:51:54 +03:00
parent 0a2a6888a0
commit 00044b91e1
+23 -9
View File
@@ -57,10 +57,7 @@ class ElmHandler(socketserver.StreamRequestHandler):
def handle(self): def handle(self):
print(f"\n🔌 Подключение: {self.client_address}") print(f"\n🔌 Подключение: {self.client_address}")
# Отправляем приветствие # Реальный ELM327 НЕ шлёт приветствие при BT-подключении — ждёт ATZ
self.wfile.write(b"\r\nELM327 v1.5\r\n>")
self.wfile.flush()
buf = b"" buf = b""
while True: while True:
try: try:
@@ -90,37 +87,49 @@ class ElmHandler(socketserver.StreamRequestHandler):
print(f"🔌 Отключение: {self.client_address}") print(f"🔌 Отключение: {self.client_address}")
def _process(self, cmd: str) -> bytes: def _process(self, cmd: str) -> bytes:
"""Обрабатывает AT/OBD команду, возвращает ответ.""" """Обрабатывает AT/OBD команду, возвращает ответ с реалистичными задержками."""
print(f" 📥 {cmd}") print(f" 📥 {cmd}")
# ── AT команды ─────────────────────── # ── AT команды ───────────────────────
if cmd == "ATZ": if cmd == "ATZ":
resp = b"\r\nELM327 v1.5\r\n>" self._sleep(2.5) # реальный ELM перезагружается ~3 сек
resp = b"\r\n\r\nELM327 v1.5\r\n>"
elif cmd == "ATSP0":
# Имитация SEARCHING как у реального ELM
self.wfile.write(b"SEARCHING...\r\n")
self.wfile.flush()
self._sleep(1.8)
resp = b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>"
elif cmd.startswith("ATE"): elif cmd.startswith("ATE"):
resp = b"OK\r\n>" resp = b"OK\r\n>"
elif cmd.startswith("ATL"): elif cmd.startswith("ATL"):
resp = b"OK\r\n>" resp = b"OK\r\n>"
elif cmd == "ATSP0":
resp = b"OK\r\n>"
elif cmd.startswith("ATH"): elif cmd.startswith("ATH"):
resp = b"OK\r\n>" resp = b"OK\r\n>"
elif cmd == "ATI": elif cmd == "ATI":
resp = b"ELM327 v1.5\r\n>" resp = b"ELM327 v1.5\r\n>"
elif cmd == "ATDP": elif cmd == "ATDP":
resp = b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>" resp = b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>"
# ── OBD запросы ───────────────────── # ── OBD запросы ─────────────────────
elif cmd == "0902": # VIN elif cmd == "0902": # VIN — ответ длинный, пауза побольше
self._sleep(0.5)
resp = self._vin_response() resp = self._vin_response()
elif cmd == "03": # Stored DTC elif cmd == "03": # Stored DTC
self._sleep(0.3)
resp = (FAKE_DTC[0] + "\r\n" + FAKE_DTC[1] + "\r\n>").encode() resp = (FAKE_DTC[0] + "\r\n" + FAKE_DTC[1] + "\r\n>").encode()
elif cmd == "07": # Pending DTC elif cmd == "07": # Pending DTC
self._sleep(0.3)
resp = (FAKE_NO_DTC + "\r\n>").encode() resp = (FAKE_NO_DTC + "\r\n>").encode()
elif cmd.startswith("01"): # PID запрос elif cmd.startswith("01"): # PID запрос
pid = cmd[2:4] pid = cmd[2:4]
self._sleep(0.2)
if pid in FAKE_PIDS: if pid in FAKE_PIDS:
resp = (FAKE_PIDS[pid] + "\r\n>").encode() resp = (FAKE_PIDS[pid] + "\r\n>").encode()
else: else:
@@ -132,6 +141,11 @@ class ElmHandler(socketserver.StreamRequestHandler):
print(f" 📤 {resp.decode('utf-8', errors='ignore').strip()[:60]}") print(f" 📤 {resp.decode('utf-8', errors='ignore').strip()[:60]}")
return resp return resp
def _sleep(self, seconds: float):
"""Пауза с случайным разбросом ±30%, как у реального железа."""
import random
time.sleep(seconds * random.uniform(0.7, 1.3))
def _vin_response(self) -> bytes: def _vin_response(self) -> bytes:
"""Формирует ответ VIN (mode 09 PID 02).""" """Формирует ответ VIN (mode 09 PID 02)."""
vin_hex = "".join(f"{ord(c):02X}" for c in FAKE_VIN) vin_hex = "".join(f"{ord(c):02X}" for c in FAKE_VIN)