test: report gb32960 live command diagnostics

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