Add last-ditch transceive probe: passive monitor + write-read-response
Sends ALL known packet families and reads MCU responses after each write. If the MCU responds, we can reverse-engineer the protocol from the response data — no logic analyzer needed. Tests both hidraw2 and hidraw1. If this also yields nothing, the only remaining path is hardware capture.
This commit is contained in:
parent
32979579d4
commit
bbc5694738
1 changed files with 337 additions and 0 deletions
337
tools/superx_rgb_lastditch.py
Normal file
337
tools/superx_rgb_lastditch.py
Normal file
|
|
@ -0,0 +1,337 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Last-ditch Super X RGB probe — no hardware needed.
|
||||
|
||||
Strategy:
|
||||
1. Monitor hidraw2 for spontaneous background traffic (MCU might send periodic data)
|
||||
2. Transceive: send ALL known packet families, read response after each
|
||||
3. If MCU responds, capture the response pattern — that's the protocol clue
|
||||
|
||||
Usage:
|
||||
sudo python superx_rgb_lastditch.py
|
||||
|
||||
This is the final software-only attempt before the logic analyzer.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import select
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
TARGET = "/dev/hidraw2"
|
||||
ALT_TARGET = "/dev/hidraw1"
|
||||
PACKET_LEN = 64
|
||||
MONITOR_TIME = 5 # seconds to listen for spontaneous traffic
|
||||
TRESPONSE_TIMEOUT = 1.0 # seconds to wait for response after write
|
||||
|
||||
|
||||
def eprint(*args, **kwargs):
|
||||
print(*args, file=sys.stderr, **kwargs)
|
||||
|
||||
|
||||
# ── Packet library ───────────────────────────────────────────────────────────
|
||||
|
||||
def pad(pkt: list[int]) -> bytes:
|
||||
return bytes(pkt + [0] * (PACKET_LEN - len(pkt)))
|
||||
|
||||
|
||||
def build_all_packets():
|
||||
"""Generate ALL known/tested packet variants as (name, category, bytes)."""
|
||||
pkts = []
|
||||
|
||||
# --- Known enable/disable ---
|
||||
pkts.append(("enable", "control", pad([0x07, 0xFF, 0x05, 0x00])))
|
||||
pkts.append(("disable", "control", pad([0x07, 0xFF, 0x05, 0x01])))
|
||||
pkts.append(("level-off", "control", pad([0x07, 0xFF, 0xFD, 0x00, 0x05, 0x01])))
|
||||
pkts.append(("level-1", "control", pad([0x07, 0xFF, 0xFD, 0x01, 0x05, 0x01])))
|
||||
pkts.append(("level-2", "control", pad([0x07, 0xFF, 0xFD, 0x01, 0x05, 0x03])))
|
||||
pkts.append(("level-3", "control", pad([0x07, 0xFF, 0xFD, 0x01, 0x05, 0x04])))
|
||||
|
||||
# --- Static colors ---
|
||||
pkts.append(("static-red", "color", pad([0x07, 0xFF, 0xFE, 0xFF, 0x00, 0x00])))
|
||||
pkts.append(("static-green", "color", pad([0x07, 0xFF, 0xFE, 0x00, 0xFF, 0x00])))
|
||||
pkts.append(("static-blue", "color", pad([0x07, 0xFF, 0xFE, 0x00, 0x00, 0xFF])))
|
||||
pkts.append(("static-white", "color", pad([0x07, 0xFF, 0xFE, 0xFF, 0xFF, 0xFF])))
|
||||
|
||||
# --- Modes ---
|
||||
pkts.append(("mode-static", "mode", pad([0x07, 0xFF, 0xFD, 0x00, 0x05, 0x04])))
|
||||
pkts.append(("mode-rainbow", "mode", pad([0x07, 0xFF, 0xFD, 0x01, 0x05, 0x04])))
|
||||
|
||||
# --- Presets ---
|
||||
for pid in [0x03, 0x05, 0x07, 0x09, 0x0B, 0x0D, 0x14]:
|
||||
pkts.append((f"preset-0x{pid:02x}", "preset", pad([0x07, 0xFF, pid, 0x00, 0x00])))
|
||||
|
||||
# --- Handshake variants ---
|
||||
pkts.append(("hs-connect", "handshake", pad([0x07, 0xFF, 0x01, 0x00])))
|
||||
pkts.append(("hs-isopen", "handshake", pad([0x07, 0xFF, 0x02, 0x00])))
|
||||
pkts.append(("hs-setopen2-0", "handshake", pad([0x07, 0xFF, 0x03, 0x01, 0x00])))
|
||||
pkts.append(("hs-setopen2-3", "handshake", pad([0x07, 0xFF, 0x03, 0x01, 0x03])))
|
||||
pkts.append(("hs-setopen2-4", "handshake", pad([0x07, 0xFF, 0x03, 0x01, 0x04])))
|
||||
pkts.append(("hs-oxp-gen2", "handshake", pad([0x07, 0x01, 0x01])))
|
||||
pkts.append(("hs-vendor-r06", "handshake", pad([0x06, 0xFF, 0x01, 0x00])))
|
||||
|
||||
# --- Alternate command bytes ---
|
||||
for cmd in range(0x00, 0x10):
|
||||
if cmd == 0x07:
|
||||
continue # already covered
|
||||
pkts.append((f"alt-cmd-0x{cmd:02x}", "alt-cmd", pad([cmd, 0xFF, 0x05, 0x00])))
|
||||
pkts.append((f"alt-cmd-0x{cmd:02x}-w", "alt-cmd", pad([cmd, 0xFF, 0xFE, 0xFF, 0xFF, 0xFF])))
|
||||
|
||||
# --- Zone selectors ---
|
||||
for zone in range(0x01, 0x08):
|
||||
pkts.append((f"zone-{zone}", "zone", pad([0x07, 0xFF, 0xFE, zone, 0xFF, 0xFF, 0xFF])))
|
||||
|
||||
# --- Zero-length variants ---
|
||||
pkts.append(("zero-short", "edge", pad([0x07])))
|
||||
pkts.append(("zero-full", "edge", pad([])))
|
||||
|
||||
return pkts
|
||||
|
||||
|
||||
# ── Monitor thread ───────────────────────────────────────────────────────────
|
||||
|
||||
class HidrawMonitor(threading.Thread):
|
||||
def __init__(self, device: str, duration: float):
|
||||
super().__init__(daemon=True)
|
||||
self.device = device
|
||||
self.duration = duration
|
||||
self.events: list[dict] = []
|
||||
self.stopped = threading.Event()
|
||||
|
||||
def run(self):
|
||||
deadline = time.monotonic() + self.duration
|
||||
try:
|
||||
fd = os.open(self.device, os.O_RDONLY | os.O_NONBLOCK)
|
||||
while time.monotonic() < deadline and not self.stopped.is_set():
|
||||
remaining = max(0.0, deadline - time.monotonic())
|
||||
r, _, _ = select.select([fd], [], [], remaining)
|
||||
if r:
|
||||
try:
|
||||
data = os.read(fd, PACKET_LEN)
|
||||
if data:
|
||||
self.events.append({
|
||||
"t": time.monotonic(),
|
||||
"data": data.hex(" "),
|
||||
"len": len(data),
|
||||
})
|
||||
except BlockingIOError:
|
||||
continue
|
||||
os.close(fd)
|
||||
except OSError as e:
|
||||
self.events.append({"t": time.monotonic(), "error": str(e)})
|
||||
|
||||
|
||||
# ── Transceive ───────────────────────────────────────────────────────────────
|
||||
|
||||
def transceive(device: str, packet: bytes, timeout: float = TRESPONSE_TIMEOUT):
|
||||
"""Write one packet, drain stale data, then collect any response."""
|
||||
result = {"write_hex": packet.hex(), "write_len": len(packet), "responses": []}
|
||||
|
||||
try:
|
||||
fd = os.open(device, os.O_RDWR | os.O_NONBLOCK)
|
||||
except OSError as e:
|
||||
result["error"] = str(e)
|
||||
return result
|
||||
|
||||
try:
|
||||
# Drain stale input
|
||||
stale = []
|
||||
while True:
|
||||
try:
|
||||
d = os.read(fd, PACKET_LEN)
|
||||
if d:
|
||||
stale.append(d.hex(" "))
|
||||
else:
|
||||
break
|
||||
except BlockingIOError:
|
||||
break
|
||||
if stale:
|
||||
result["stale"] = stale
|
||||
|
||||
# Write
|
||||
written = os.write(fd, packet)
|
||||
result["written"] = written
|
||||
|
||||
# Read responses
|
||||
deadline = time.monotonic() + timeout
|
||||
while time.monotonic() < deadline:
|
||||
remaining = max(0.0, deadline - time.monotonic())
|
||||
r, _, _ = select.select([fd], [], [], remaining)
|
||||
if not r:
|
||||
break
|
||||
try:
|
||||
resp = os.read(fd, PACKET_LEN)
|
||||
if resp:
|
||||
result["responses"].append({
|
||||
"t_offset_ms": round((time.monotonic() - (deadline - timeout)) * 1000, 1),
|
||||
"data": resp.hex(" "),
|
||||
"len": len(resp),
|
||||
})
|
||||
except BlockingIOError:
|
||||
continue
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ── Main ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="Last-ditch Super X RGB probe")
|
||||
ap.add_argument("--device", default=TARGET, help=f"target hidraw (default: {TARGET})")
|
||||
ap.add_argument("--yes", "-y", action="store_true", help="skip confirmations")
|
||||
ap.add_argument("--log", default=None, help="log JSONL output")
|
||||
ap.add_argument("--monitor-only", action="store_true", help="only passive monitoring")
|
||||
args = ap.parse_args()
|
||||
|
||||
print("╔══════════════════════════════════════════╗")
|
||||
print("║ Super X RGB — Last Ditch Probe ║")
|
||||
print("║ (no hardware, all software paths) ║")
|
||||
print("╚══════════════════════════════════════════╝")
|
||||
print()
|
||||
|
||||
# Phase 0: Check device
|
||||
for dev in [args.device, ALT_TARGET]:
|
||||
try:
|
||||
fd = os.open(dev, os.O_RDWR | os.O_NONBLOCK)
|
||||
os.close(fd)
|
||||
print(f"✓ {dev} — writable")
|
||||
except OSError as e:
|
||||
print(f"✗ {dev} — {e}")
|
||||
|
||||
if args.monitor_only:
|
||||
print(f"\n--- Passive Monitor ({MONITOR_TIME}s) ---")
|
||||
print("Watching for spontaneous MCU traffic. Touch keys/press buttons/change modes if possible.")
|
||||
mon = HidrawMonitor(args.device, MONITOR_TIME)
|
||||
mon.start()
|
||||
mon.join()
|
||||
if mon.events:
|
||||
print(f"\n{len(mon.events)} event(s) captured:")
|
||||
for ev in mon.events:
|
||||
if "data" in ev:
|
||||
print(f" +{ev['t']:.3f}s len={ev['len']} {ev['data']}")
|
||||
elif "error" in ev:
|
||||
print(f" ERROR: {ev['error']}")
|
||||
else:
|
||||
print("\nNo spontaneous traffic detected.")
|
||||
return 0
|
||||
|
||||
# Phase 1: Passive monitoring
|
||||
print(f"\n--- Phase 1: Passive Monitor ({MONITOR_TIME}s) ---")
|
||||
print("MCU might send periodic data. Watching...")
|
||||
mon = HidrawMonitor(args.device, MONITOR_TIME)
|
||||
mon.start()
|
||||
|
||||
for i in range(MONITOR_TIME, 0, -1):
|
||||
print(f" {i}...", end="\r", flush=True)
|
||||
time.sleep(1)
|
||||
print(" done. ")
|
||||
|
||||
mon.stopped.set()
|
||||
mon.join(timeout=1)
|
||||
|
||||
if mon.events:
|
||||
print(f"\n{len(mon.events)} spontaneous event(s) detected on {args.device}:")
|
||||
for ev in mon.events:
|
||||
if "data" in ev:
|
||||
print(f" len={ev['len']} {ev['data']}")
|
||||
elif "error" in ev:
|
||||
print(f" ERROR: {ev['error']}")
|
||||
print(" → MCU IS TALKING! These frames might be the RGB protocol!")
|
||||
else:
|
||||
print(" No spontaneous traffic.")
|
||||
|
||||
# Phase 2: Transceive all packets
|
||||
packets = build_all_packets()
|
||||
print(f"\n--- Phase 2: Transceive ({len(packets)} packets) ---")
|
||||
|
||||
if not args.yes:
|
||||
input("Press Enter to start sending packets and reading responses...")
|
||||
|
||||
log_entries = []
|
||||
responses_seen = 0
|
||||
|
||||
for name, category, pkt in packets:
|
||||
result = transceive(args.device, pkt)
|
||||
entry = {
|
||||
"name": name,
|
||||
"category": category,
|
||||
"write_hex": result["write_hex"],
|
||||
"response_count": len(result["responses"]),
|
||||
"responses": result["responses"],
|
||||
}
|
||||
if result.get("error"):
|
||||
entry["error"] = result["error"]
|
||||
log_entries.append(entry)
|
||||
|
||||
# Print summary
|
||||
if result["responses"]:
|
||||
responses_seen += 1
|
||||
responses_str = ", ".join(f"{r['len']}B @{r['t_offset_ms']}ms" for r in result["responses"])
|
||||
print(f" {name:25s} → {len(result['responses'])} response(s): {responses_str}")
|
||||
elif result.get("error"):
|
||||
print(f" {name:25s} → ERROR: {result['error']}")
|
||||
else:
|
||||
print(f" {name:25s} → no response")
|
||||
|
||||
print(f"\n--- Summary ---")
|
||||
print(f" Packets sent: {len(packets)}")
|
||||
print(f" Responses received: {responses_seen} packets triggered a response")
|
||||
|
||||
if responses_seen > 0:
|
||||
print("\n ⚡ MCU RESPONDS! These are the protocol clues we need:")
|
||||
for entry in log_entries:
|
||||
if entry["responses"]:
|
||||
print(f" {entry['name']}: {entry['write_hex']}")
|
||||
for r in entry["responses"]:
|
||||
print(f" → {r['data']}")
|
||||
|
||||
# Phase 3: Try alternate device
|
||||
print(f"\n--- Phase 3: Alternate Device ({ALT_TARGET}) ---")
|
||||
alt_has_response = False
|
||||
for name, category, pkt in packets[:10]: # just first 10 on alt device
|
||||
result = transceive(ALT_TARGET, pkt)
|
||||
if result.get("error"):
|
||||
continue
|
||||
if result["responses"]:
|
||||
alt_has_response = True
|
||||
print(f" {name:25s} → {len(result['responses'])} response(s) on {ALT_TARGET}!")
|
||||
break
|
||||
|
||||
if not alt_has_response:
|
||||
print(f" No responses on {ALT_TARGET} with quick test.")
|
||||
|
||||
# Save log
|
||||
if args.log:
|
||||
log_path = Path(args.log)
|
||||
else:
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||||
log_path = Path(f"/tmp/superx-lastditch-{stamp}.jsonl")
|
||||
|
||||
log_data = {
|
||||
"device": args.device,
|
||||
"alt_device": ALT_TARGET,
|
||||
"spontaneous_events": len(mon.events),
|
||||
"spontaneous_data": [{"data": e.get("data", ""), "len": e.get("len", 0)}
|
||||
for e in mon.events if "data" in e],
|
||||
"transceive_results": log_entries,
|
||||
"alt_device_has_response": alt_has_response,
|
||||
}
|
||||
log_path.write_text(json.dumps(log_data, indent=2) + "\n")
|
||||
print(f"\nLog saved: {log_path}")
|
||||
print("Send this file — it contains all the raw response data.")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Add table
Reference in a new issue