(args: argparse.Namespace)
| 195 | |
| 196 | |
| 197 | def run_demo(args: argparse.Namespace) -> int: |
| 198 | camera = create_camera(args) |
| 199 | selected_info = camera.info |
| 200 | backend = selected_info.get("backend", camera.backend) |
| 201 | print(f"Selected {backend}: {selected_info.get('name', 'unknown')}") |
| 202 | for key in ("serial", "uid", "physical_port", "usb_id", "product_id", "firmware"): |
| 203 | if selected_info.get(key): |
| 204 | print(f" {key}={selected_info[key]}") |
| 205 | |
| 206 | output_dir: Path | None = None |
| 207 | if args.output_dir: |
| 208 | output_dir = Path(args.output_dir) |
| 209 | elif not args.show_display and not args.show_viser: |
| 210 | output_dir = Path("live_demo_output") / datetime.now().strftime("%Y%m%d_%H%M%S") |
| 211 | if output_dir: |
| 212 | output_dir.mkdir(parents=True, exist_ok=True) |
| 213 | print(f"Writing latest frames to: {output_dir}") |
| 214 | |
| 215 | refiner = TensorRTRefiner(args) |
| 216 | |
| 217 | video_writer: cv2.VideoWriter | None = None |
| 218 | pointcloud_viewer: ViserPointCloudViewer | None = None |
| 219 | camera_started = False |
| 220 | last_report = time.perf_counter() |
| 221 | frames = 0 |
| 222 | saved = 0 |
| 223 | try: |
| 224 | camera.start() |
| 225 | camera_started = True |
| 226 | selected_info = camera.info |
| 227 | color_intrinsics = camera.intrinsics |
| 228 | if output_dir: |
| 229 | write_run_metadata(args, output_dir, selected_info) |
| 230 | if args.show_viser: |
| 231 | pointcloud_viewer = ViserPointCloudViewer(args, color_intrinsics) |
| 232 | |
| 233 | for _ in range(args.warmup): |
| 234 | camera.wait_frame(args.timeout_ms) |
| 235 | |
| 236 | while True: |
| 237 | rgbd = camera.wait_frame(args.timeout_ms) |
| 238 | if rgbd is None: |
| 239 | continue |
| 240 | |
| 241 | color_bgr = rgbd.color_bgr |
| 242 | depth_m = rgbd.depth_m |
| 243 | |
| 244 | refined, infer_s = refiner.refine(color_bgr, depth_m) |
| 245 | comparison = make_comparison(depth_m, refined, infer_s * 1000.0) |
| 246 | frames += 1 |
| 247 | if pointcloud_viewer is not None: |
| 248 | pointcloud_viewer.update(color_bgr, depth_m, refined, frames, infer_s * 1000.0) |
| 249 | |
| 250 | if video_writer is None: |
| 251 | h, w = comparison.shape[:2] |
| 252 | video_writer = open_video_writer(args, (w, h)) |
| 253 | if video_writer is not None: |
| 254 | video_writer.write(comparison) |
no test coverage detected