feat: elm_proto.py переписан по сырому коду AndrOBD + mock v2
This commit is contained in:
Repinoid
+47
-105
@@ -1,14 +1,13 @@
|
||||
#!/usr/bin/env python3
|
||||
"""mock_elm327_v2.py — эмулятор ELM327 с '>' промптами и симуляцией ошибок.
|
||||
"""Mock ELM327 v2 — совместим с AndrOBD-паттернами (elm_proto.py).
|
||||
|
||||
Соответствует паттернам OBD-Droid:
|
||||
- Каждый ответ заканчивается '>' (промпт)
|
||||
- SEARCHING... перед первым ответом после ATSP0
|
||||
- Случайные ошибки: BUS BUSY, UNABLE TO CONNECT
|
||||
- Побайтовая отправка с реалистичными задержками
|
||||
- Многострочные ответы для DTC (как мульти-ЭБУ)
|
||||
Эмулирует:
|
||||
- '>' как разделитель строк (AndrOBD StreamHandler)
|
||||
- Адаптивный таймаут ATST (AndrOBD AdaptiveTiming)
|
||||
- Ошибки: SEARCHING, NO DATA, BUS BUSY, UNABLE
|
||||
- Побайтовую отправку с микро-задержками
|
||||
|
||||
Запуск: python tools/mock_elm327_v2.py
|
||||
Запуск: python tools/mock_elm327_v2.py
|
||||
"""
|
||||
|
||||
import argparse
|
||||
@@ -16,29 +15,13 @@ import random
|
||||
import socketserver
|
||||
import time
|
||||
|
||||
|
||||
FAKE_VIN = "WVWZZZ1KZAW123456"
|
||||
|
||||
# Многострочные DTC — имитация нескольких ЭБУ
|
||||
FAKE_DTC_MULTI = [
|
||||
"43011300000000", # P0301 (ЭБУ 1)
|
||||
"43013300000000", # P0303 (ЭБУ 1)
|
||||
"47500047000000", # P0047 (ЭБУ 2, pending)
|
||||
]
|
||||
|
||||
FAKE_DTC = ["43011300000000", "43013300000000"]
|
||||
FAKE_NO_DTC = "43000000000000"
|
||||
|
||||
FAKE_PIDS = {
|
||||
"05": "4105 5A",
|
||||
"0C": "410C 1A F8",
|
||||
"0D": "410D 00",
|
||||
"11": "4111 80",
|
||||
"0B": "410B 64",
|
||||
"0F": "410F 46",
|
||||
"1F": "411F 02 58",
|
||||
"04": "4104 40",
|
||||
"06": "4106 80",
|
||||
"07": "4107 80",
|
||||
"05": "41055A", "0C": "410C1AF8", "0D": "410D00",
|
||||
"11": "411180", "0B": "410B64", "0F": "410F46",
|
||||
"1F": "411F0258", "04": "410440", "06": "410680", "07": "410780",
|
||||
}
|
||||
|
||||
|
||||
@@ -50,116 +33,75 @@ class ElmHandler(socketserver.StreamRequestHandler):
|
||||
while True:
|
||||
try:
|
||||
ch = self.rfile.read(1)
|
||||
if not ch:
|
||||
break
|
||||
if not ch: break
|
||||
if ch in (b"\n", b"\r"):
|
||||
if buf:
|
||||
cmd = buf.decode("utf-8", errors="ignore").strip().upper()
|
||||
cmd = buf.decode(errors="ignore").strip().upper()
|
||||
buf = b""
|
||||
if cmd:
|
||||
self._respond(cmd)
|
||||
if cmd: self._respond(cmd)
|
||||
else:
|
||||
buf += ch
|
||||
except (ConnectionResetError, BrokenPipeError, OSError):
|
||||
break
|
||||
|
||||
print(f"🔌 Откл: {self.client_address}")
|
||||
|
||||
def _respond(self, cmd: str):
|
||||
"""Обрабатывает команду и отправляет ответ побайтово."""
|
||||
print(f" 📥 {cmd}")
|
||||
|
||||
# Случайная симуляция ошибок (3%)
|
||||
if random.random() < 0.03 and cmd not in ("ATZ", "ATE0", "ATL0", "ATH0"):
|
||||
error = random.choice([
|
||||
b"BUS BUSY\r\n>",
|
||||
b"BUS BUSY\r\n>",
|
||||
b"UNABLE TO CONNECT\r\n>",
|
||||
])
|
||||
self._send_bytes(error)
|
||||
print(f" 📤 (ERROR) {error.decode().strip()}")
|
||||
return
|
||||
if cmd.startswith("ATST"):
|
||||
self._send(b"OK\r\n>"); return
|
||||
|
||||
# ── AT команды ───────────────────────
|
||||
if cmd == "ATZ":
|
||||
self._sleep(2.5)
|
||||
self._send_bytes(b"\r\nELM327 v1.5\r\n>")
|
||||
if random.random() < 0.03 and cmd not in ("ATZ",):
|
||||
err = random.choice([b"BUS BUSY\r\n>", b"BUS BUSY\r\n>",
|
||||
b"UNABLE TO CONNECT\r\n>"])
|
||||
self._send(err); return
|
||||
|
||||
elif cmd == "ATSP0":
|
||||
self._send_bytes(b"SEARCHING...\r\n")
|
||||
if cmd == "ATSP0":
|
||||
self._send(b"SEARCHING...\r\n")
|
||||
self._sleep(1.8)
|
||||
self._send_bytes(b"AUTO, ISO 15765-4 (CAN 11/500)\r\n>")
|
||||
|
||||
elif cmd in ("ATE0", "ATL0", "ATS0", "ATH0", "ATH1"):
|
||||
self._send_bytes(b"OK\r\n>")
|
||||
|
||||
elif cmd == "ATI":
|
||||
self._send_bytes(b"ELM327 v1.5\r\n>")
|
||||
|
||||
# ── OBD запросы ─────────────────────
|
||||
self._send(b"AUTO\r\n>")
|
||||
elif cmd in ("ATAT1", "ATE0", "ATL0", "ATS0", "ATH0", "ATH1", "ATPC"):
|
||||
self._send(b"OK\r\n>")
|
||||
elif cmd == "0902":
|
||||
self._sleep(0.5)
|
||||
resp = self._vin_bytes()
|
||||
self._send_bytes(resp)
|
||||
|
||||
v = "".join(f"{ord(c):02X}" for c in FAKE_VIN)
|
||||
self._send(f"014\r\n0:490201{v}\r\n>".encode())
|
||||
elif cmd == "03":
|
||||
self._sleep(0.3)
|
||||
# Многострочный ответ — как от нескольких ЭБУ
|
||||
for line in FAKE_DTC_MULTI:
|
||||
self._send_bytes(line.encode() + b"\r\n")
|
||||
for line in FAKE_DTC:
|
||||
self._send(line.encode() + b"\r\n")
|
||||
time.sleep(0.05)
|
||||
self._send_bytes(b">")
|
||||
|
||||
self._send(b">")
|
||||
elif cmd == "07":
|
||||
self._sleep(0.3)
|
||||
self._send_bytes(FAKE_NO_DTC.encode() + b"\r\n>")
|
||||
|
||||
self._send(FAKE_NO_DTC.encode() + b"\r\n>")
|
||||
elif cmd.startswith("01"):
|
||||
pid = cmd[2:4]
|
||||
self._sleep(0.2)
|
||||
if pid in FAKE_PIDS:
|
||||
self._send_bytes(FAKE_PIDS[pid].encode() + b"\r\n>")
|
||||
else:
|
||||
self._send_bytes(b"NO DATA\r\n>")
|
||||
|
||||
pid = cmd[2:4]; self._sleep(0.2)
|
||||
self._send((FAKE_PIDS.get(pid, "NODATA") + "\r\n>").encode())
|
||||
else:
|
||||
self._send_bytes(b"?\r\n>")
|
||||
self._send(b"?\r\n>")
|
||||
|
||||
def _send_bytes(self, data: bytes):
|
||||
"""Отправляет данные побайтово с микро-задержкой (как реальный ELM)."""
|
||||
def _send(self, data: bytes):
|
||||
for byte in data:
|
||||
self.wfile.write(bytes([byte]))
|
||||
self.wfile.flush()
|
||||
time.sleep(0.0005) # 0.5 мс между байтами
|
||||
time.sleep(0.0005)
|
||||
|
||||
def _sleep(self, seconds: float):
|
||||
"""Пауза ±30%, как у реального железа."""
|
||||
time.sleep(seconds * random.uniform(0.7, 1.3))
|
||||
|
||||
def _vin_bytes(self) -> bytes:
|
||||
vin_hex = "".join(f"{ord(c):02X}" for c in FAKE_VIN)
|
||||
return f"014\r\n0:49 02 01 {vin_hex}\r\n>".encode()
|
||||
def _sleep(self, s: float):
|
||||
time.sleep(s * random.uniform(0.7, 1.3))
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Mock ELM327 v2 (OBD-Droid совместимый)")
|
||||
parser.add_argument("--port", type=int, default=35000)
|
||||
parser.add_argument("--host", default="0.0.0.0")
|
||||
args = parser.parse_args()
|
||||
|
||||
class ReuseTCPServer(socketserver.ThreadingTCPServer):
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("--port", type=int, default=35000)
|
||||
p.add_argument("--host", default="0.0.0.0")
|
||||
args = p.parse_args()
|
||||
class S(socketserver.ThreadingTCPServer):
|
||||
allow_reuse_address = True
|
||||
|
||||
server = ReuseTCPServer((args.host, args.port), ElmHandler)
|
||||
print(f"🚗 Mock ELM327 v2 на {args.host}:{args.port}")
|
||||
print(f" Промпт '>' после каждого ответа")
|
||||
print(f" Случайные ошибки: 3%")
|
||||
print(f" Ctrl+C = стоп")
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
print("\n⏹️ Стоп")
|
||||
server.shutdown()
|
||||
server = S((args.host, args.port), ElmHandler)
|
||||
print(f"🚗 Mock ELM327 v2 (AndrOBD) на {args.host}:{args.port}")
|
||||
try: server.serve_forever()
|
||||
except KeyboardInterrupt: print("\n⏹️ Стоп"); server.shutdown()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user