67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Host probe v6 — envelope variants via the library business API.
|
|
|
|
Sends the ``battery.PROBE_VARIANTS`` envelopes through
|
|
``LocalDeviceClient.send_probe`` (sequence numbers, ``opt``-seeded
|
|
payloads, socket ids and cipher all come from the library) and prints
|
|
exact TX/RX bytes plus any parsed reply/SoC.
|
|
|
|
Run from the repo root on the SoftAP-joined host:
|
|
python3 scripts/host_probe6.py --sn A03X2516013210060005 --ports 9000
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(
|
|
os.path.abspath(__file__))))
|
|
|
|
from franklinwh_agate_client.battery import ( # noqa: E402
|
|
extract_soc,
|
|
probe_device,
|
|
)
|
|
from franklinwh_agate_client.protocol import ( # noqa: E402
|
|
generate_secret_key,
|
|
)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--sn", required=True)
|
|
ap.add_argument("--host", default="10.100.1.1")
|
|
ap.add_argument("--ports", default="9000,18000")
|
|
ap.add_argument("--timeout", type=float, default=5.0)
|
|
args = ap.parse_args()
|
|
print("secret key for %s: %s (radix 16)"
|
|
% (args.sn, generate_secret_key(args.sn)))
|
|
ports = tuple(int(p) for p in args.ports.split(",") if p.strip())
|
|
try:
|
|
results = probe_device(args.sn, host=args.host, ports=ports,
|
|
timeout=args.timeout)
|
|
except (ConnectionError, socket.error, TimeoutError) as exc:
|
|
print("ERROR: %s: %s" % (type(exc).__name__, exc))
|
|
return 1
|
|
for out in results:
|
|
print("--- %s -> %s:%d cmd=%s"
|
|
% (out["variant"], args.host, out["port"], out["cmd"]))
|
|
if out.get("envelope") is not None:
|
|
print(" envelope: " + json.dumps(out["envelope"], default=str))
|
|
tx = out.get("tx") or b""
|
|
print(" tx (%d B): %s" % (len(tx), tx.hex()))
|
|
rx = out.get("reply_raw") or b""
|
|
print(" rx (%d B): %s" % (len(rx), rx.hex()[:400] or "(empty)"))
|
|
if out.get("reply") is not None:
|
|
print(" reply: " + json.dumps(out["reply"], default=str)[:2000])
|
|
soc = extract_soc(out["reply"])
|
|
print(" soc: %s (source=%s)" % (soc["value"], soc["source"]))
|
|
if out.get("error"):
|
|
print(" error: %s" % out["error"])
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|