()
| 173 | |
| 174 | |
| 175 | def main(): |
| 176 | args = parse_args() |
| 177 | |
| 178 | if not HAS_CV2: |
| 179 | print("ERROR: opencv-python is not installed.") |
| 180 | print(" Install with: pip install opencv-python") |
| 181 | sys.exit(1) |
| 182 | |
| 183 | use_camera = not args.no_camera |
| 184 | cap = None |
| 185 | |
| 186 | if use_camera: |
| 187 | cap = open_camera(args.camera, args.fps) |
| 188 | if cap is None: |
| 189 | print( |
| 190 | f"WARNING: could not open camera {args.camera}, falling back to synthetic pattern." |
| 191 | ) |
| 192 | use_camera = False |
| 193 | |
| 194 | sender = UDPSender(UDP_HOST, args.port) |
| 195 | interval = 1.0 / args.fps |
| 196 | |
| 197 | print( |
| 198 | f"Camera Telemetry → udp://{UDP_HOST}:{args.port} @ {args.fps:.0f} Hz quality={args.quality}" |
| 199 | ) |
| 200 | print("Source:", f"camera {args.camera}" if use_camera else "synthetic pattern") |
| 201 | print("Press Ctrl+C to stop.\n") |
| 202 | print(f"{'Frame':>8} {'FPS':>6} {'JPEG kB':>8}") |
| 203 | print("-" * 28) |
| 204 | |
| 205 | frame_count = 0 |
| 206 | t_start = time.monotonic() |
| 207 | |
| 208 | try: |
| 209 | while True: |
| 210 | t0 = time.monotonic() |
| 211 | t = t0 - t_start |
| 212 | |
| 213 | if use_camera: |
| 214 | ok, frame_bgr = cap.read() |
| 215 | if not ok: |
| 216 | print("Camera read failed, switching to synthetic pattern.") |
| 217 | use_camera = False |
| 218 | frame_bgr = make_synthetic_frame(t) |
| 219 | else: |
| 220 | frame_bgr = make_synthetic_frame(t) |
| 221 | |
| 222 | frame_count += 1 |
| 223 | |
| 224 | jpeg_bytes = encode_jpeg(frame_bgr, args.quality) |
| 225 | if jpeg_bytes: |
| 226 | sender.send(jpeg_bytes) |
| 227 | |
| 228 | fps_actual = frame_count / max(t, 0.001) |
| 229 | sender.send( |
| 230 | b"\xab\xcd\xef" |
| 231 | + f"{fps_actual:.1f},{frame_count}".encode() |
| 232 | + b"\xfe\xed" |
no test coverage detected