From b1bd67d4d2b581f143eb5c8814408262bb9a5c14 Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 21:08:15 +0800 Subject: [PATCH] test: report gb32960 live command diagnostics --- tools/test_vehicle_ingest_live_verify.py | 52 +++++++++++ tools/vehicle_ingest_live_verify.py | 112 +++++++++++++++++++++++ 2 files changed, 164 insertions(+) diff --git a/tools/test_vehicle_ingest_live_verify.py b/tools/test_vehicle_ingest_live_verify.py index e9184551..d2f300e3 100644 --- a/tools/test_vehicle_ingest_live_verify.py +++ b/tools/test_vehicle_ingest_live_verify.py @@ -73,6 +73,58 @@ class VehicleIngestLiveVerifyTest(unittest.TestCase): self.assertIn("peer LIKE '115.29.187.%'", sql) + def test_gb32960_command_counts_sql_uses_same_peer_filter(self): + sql = live_verify.gb32960_command_count_sql( + date_from="2026-06-29 00:00:00", + date_to="2026-06-29 23:59:59", + peer_like="115.29.187.205", + ) + + self.assertIn("protocol = 'GB32960'", sql) + self.assertIn("peer LIKE '%115.29.187.205%'", sql) + self.assertIn("GROUP BY message_id", sql) + + def test_gb32960_command_count_rows_include_command_names(self): + response = { + "code": 0, + "data": [ + [2, 9], + [5, 3], + ], + } + + rows = live_verify.command_count_rows(response) + + self.assertEqual(rows, [ + {"messageId": 2, "command": "REALTIME_REPORT", "count": 9}, + {"messageId": 5, "command": "PLATFORM_LOGIN", "count": 3}, + ]) + + def test_gb32960_recent_frame_rows_surface_peer_and_raw_uri(self): + response = { + "code": 0, + "data": [[ + "2026-06-29T12:26:21.385Z", + 5, + "", + "unknown:GB32960:f729d788b943630e", + "115.29.187.205:59630", + "archive://frame.bin", + ]], + } + + rows = live_verify.recent_frame_rows(response) + + self.assertEqual(rows, [{ + "ts": "2026-06-29T12:26:21.385Z", + "messageId": 5, + "command": "PLATFORM_LOGIN", + "vin": "", + "vehicleKey": "unknown:GB32960:f729d788b943630e", + "peer": "115.29.187.205:59630", + "rawUri": "archive://frame.bin", + }]) + def test_jt808_location_sql_uses_dedicated_location_stable_without_protocol_filter(self): sql = live_verify.jt808_location_count_sql( date_from="2026-06-29 00:00:00", diff --git a/tools/vehicle_ingest_live_verify.py b/tools/vehicle_ingest_live_verify.py index e2e469a4..531701fb 100755 --- a/tools/vehicle_ingest_live_verify.py +++ b/tools/vehicle_ingest_live_verify.py @@ -87,6 +87,40 @@ def latest_gb32960_platform_raw_uri_sql(date_from: str, date_to: str, peer_like: return "SELECT raw_uri FROM raw_frames WHERE " + " AND ".join(where) + " ORDER BY ts DESC LIMIT 1" +def gb32960_command_count_sql(date_from: str, date_to: str, peer_like: str = "") -> str: + filters = ["protocol = 'GB32960'"] + if peer_like: + filters.append("peer LIKE " + tdengine_smoke.sql_literal(peer_like_pattern(peer_like))) + where = list(filters) + if date_from: + where.append("ts >= " + tdengine_smoke.sql_literal(date_from)) + if date_to: + where.append("ts <= " + tdengine_smoke.sql_literal(date_to)) + return ( + "SELECT message_id,COUNT(*) FROM raw_frames WHERE " + + " AND ".join(where) + + " GROUP BY message_id ORDER BY message_id" + ) + + +def gb32960_recent_frame_sql(date_from: str, date_to: str, peer_like: str = "", limit: int = 10) -> str: + filters = ["protocol = 'GB32960'"] + if peer_like: + filters.append("peer LIKE " + tdengine_smoke.sql_literal(peer_like_pattern(peer_like))) + where = list(filters) + if date_from: + where.append("ts >= " + tdengine_smoke.sql_literal(date_from)) + if date_to: + where.append("ts <= " + tdengine_smoke.sql_literal(date_to)) + safe_limit = max(1, min(limit, 100)) + return ( + "SELECT ts,message_id,vin,vehicle_key,peer,raw_uri FROM raw_frames WHERE " + + " AND ".join(where) + + " ORDER BY ts DESC LIMIT " + + str(safe_limit) + ) + + def peer_like_pattern(value: str) -> str: value = value.strip() if "%" in value or "_" in value: @@ -94,6 +128,53 @@ def peer_like_pattern(value: str) -> str: return "%" + value + "%" +def command_name(message_id: int) -> str: + return { + 1: "VEHICLE_LOGIN", + 2: "REALTIME_REPORT", + 3: "RESEND_REPORT", + 4: "VEHICLE_LOGOUT", + 5: "PLATFORM_LOGIN", + 6: "PLATFORM_LOGOUT", + 7: "HEARTBEAT", + 8: "TIME_CALIBRATION", + }.get(message_id, "UNKNOWN_0x" + format(message_id, "02X")) + + +def response_rows(response: dict[str, Any]) -> list[list[Any]]: + if response.get("code") != 0: + raise RuntimeError(f"tdengine query failed: {response}") + return response.get("data") or [] + + +def command_count_rows(response: dict[str, Any]) -> list[dict[str, Any]]: + rows = [] + for item in response_rows(response): + message_id = int(item[0]) + rows.append({ + "messageId": message_id, + "command": command_name(message_id), + "count": int(item[1]), + }) + return rows + + +def recent_frame_rows(response: dict[str, Any]) -> list[dict[str, Any]]: + rows = [] + for item in response_rows(response): + message_id = int(item[1]) + rows.append({ + "ts": item[0], + "messageId": message_id, + "command": command_name(message_id), + "vin": item[2], + "vehicleKey": item[3], + "peer": item[4], + "rawUri": item[5], + }) + return rows + + def evaluate_counts(counts: dict[str, int], thresholds: Thresholds) -> list[Check]: checks = [ threshold_check( @@ -209,6 +290,32 @@ def query_latest_platform_raw_uri(args: argparse.Namespace) -> str: return value if isinstance(value, str) else "" +def query_gb32960_command_counts(args: argparse.Namespace) -> list[dict[str, Any]]: + response = query_json( + args.tdengine_rest_url, + args.tdengine_username, + args.tdengine_password, + gb32960_command_count_sql(args.date_from, args.date_to, args.gb32960_peer_like), + args.http_timeout, + ) + return command_count_rows(response) + + +def query_gb32960_recent_frames(args: argparse.Namespace) -> list[dict[str, Any]]: + response = query_json( + args.tdengine_rest_url, + args.tdengine_username, + args.tdengine_password, + gb32960_recent_frame_sql( + args.date_from, + args.date_to, + args.gb32960_peer_like, + args.gb32960_recent_limit), + args.http_timeout, + ) + return recent_frame_rows(response) + + def decode_frame(history_base_url: str, raw_uri: str, timeout: float) -> dict[str, Any]: if not raw_uri: return {} @@ -233,6 +340,8 @@ def run(args: argparse.Namespace) -> dict[str, Any]: checks = evaluate_counts(counts, thresholds) latest_raw_uri = "" decoded_platform_frame: dict[str, Any] = {} + command_counts = query_gb32960_command_counts(args) + recent_frames = query_gb32960_recent_frames(args) if args.history_base_url: latest_raw_uri = query_latest_platform_raw_uri(args) decoded_platform_frame = decode_frame(args.history_base_url, latest_raw_uri, args.http_timeout) @@ -242,6 +351,8 @@ def run(args: argparse.Namespace) -> dict[str, Any]: "dateTo": args.date_to, "counts": counts, "checks": [asdict(check) for check in checks], + "gb32960CommandCounts": command_counts, + "gb32960RecentFrames": recent_frames, "gb32960LatestPlatformRawUri": latest_raw_uri, "gb32960LatestPlatformCommand": decoded_platform_frame.get("command"), } @@ -262,6 +373,7 @@ def parser() -> argparse.ArgumentParser: p.add_argument("--min-gb32960-platform-logins", type=int, default=1) p.add_argument("--min-gb32960-vehicle-realtime-frames", type=int, default=1) p.add_argument("--require-gb32960-vehicle-realtime", action="store_true") + p.add_argument("--gb32960-recent-limit", type=int, default=10) p.add_argument("--http-timeout", type=float, default=5) return p