Monitor Serial Studio in real-time: connection status, frame counters, and (optionally) raw incoming bytes.
(api: SerialStudioAPI, args)
| 1086 | |
| 1087 | |
| 1088 | def cmd_monitor(api: SerialStudioAPI, args) -> int: |
| 1089 | """Monitor Serial Studio in real-time: connection status, frame counters, |
| 1090 | and (optionally) raw incoming bytes.""" |
| 1091 | print("=" * 60) |
| 1092 | print("Serial Studio Live Monitor") |
| 1093 | print("=" * 60) |
| 1094 | print("Press Ctrl+C to exit\n") |
| 1095 | |
| 1096 | update_interval = args.interval / 1000.0 |
| 1097 | frame_count = 0 |
| 1098 | raw_data_count = 0 |
| 1099 | last_status: dict = {} |
| 1100 | last_update_time = time.time() |
| 1101 | |
| 1102 | try: |
| 1103 | while True: |
| 1104 | current_time = time.time() |
| 1105 | |
| 1106 | # Drain push notifications (frames + raw data) |
| 1107 | while api.has_data_available(timeout=0.0): |
| 1108 | msg = api.recv_message(timeout=0.1) |
| 1109 | if not msg: |
| 1110 | continue |
| 1111 | |
| 1112 | if "data" in msg and "frames" not in msg: |
| 1113 | raw_data_count += 1 |
| 1114 | if args.show_raw_data: |
| 1115 | try: |
| 1116 | decoded = base64.b64decode(msg["data"]).decode( |
| 1117 | "utf-8", errors="replace" |
| 1118 | ) |
| 1119 | display_text = decoded.replace("\n", "\\n").replace( |
| 1120 | "\r", "\\r" |
| 1121 | ) |
| 1122 | if len(display_text) > 100: |
| 1123 | display_text = display_text[:97] + "..." |
| 1124 | print( |
| 1125 | f"\rRaw [{raw_data_count}]: " |
| 1126 | f"{display_text}".ljust(120), |
| 1127 | end="", |
| 1128 | flush=True, |
| 1129 | ) |
| 1130 | except Exception: |
| 1131 | pass |
| 1132 | elif "frames" in msg: |
| 1133 | frame_count += len(msg.get("frames", [])) |
| 1134 | |
| 1135 | if current_time - last_update_time >= update_interval: |
| 1136 | last_update_time = current_time |
| 1137 | |
| 1138 | response = api.send_command("io.getStatus") |
| 1139 | if not response or not response.get("success"): |
| 1140 | if api.verbose or args.compact: |
| 1141 | print("[ERROR] Failed to get status (will retry)") |
| 1142 | time.sleep(0.01) |
| 1143 | continue |
| 1144 | |
| 1145 | status = response.get("result", {}) |
no test coverage detected