| |
| |
| """ |
| DepthART 相对深度 axmodel 板端推理（AX650 + axengine）。 |
| |
| 随板部署文件（建议放在同一目录）： |
| relative_s_448_default_ax650.axmodel |
| depthart_infer_axmodel.py |
| 待测图片或图片目录 |
| |
| 模型接口（以运行时 get_inputs/get_outputs 实际返回为准）： |
| input [1, 3, 448, 448] uint8 NCHW RGB |
| depth [1, 1, 448, 448] float32 仿射不变相对深度，数值越大表示越近 |
| |
| 预处理按编译配置 config_axonnx.json 的 input_processors 执行：tensor_format=RGB、 |
| src_dtype=U8、mean/std 为 0-255 域的 ImageNet 值，归一化由编译前端完成。因此板端 |
| 送 0-255 原始 RGB 像素即可，不要再做 /255 或减均值。若 axmodel 输入为 FP32（编译 |
| 时未配置 input_processors），脚本默认改为 /255 + ImageNet mean/std，与 ONNX 基准 |
| 一致；可用 --input-norm 显式覆盖。 |
| |
| 深度还原使用与 ONNX 脚本一致的 bilinear + align_corners=True（linspace + remap）， |
| 保证板端与开发机基准可以逐像素对比。 |
| |
| 板端计时分为 NPU（session.run）和 CPU（读取、预处理、输出提取/校验、后处理、 |
| 可视化、写盘），两者分别统计，避免弱 CPU 上的后处理掩盖真实 NPU 性能。 |
| |
| 单图模式：打印深度统计、NPU/CPU 各阶段耗时，保存并排可视化，不产出评测文件。 |
| 数据集模式：tqdm 进度 + npu_timing.json；默认快速模式只统计读取/预处理/NPU， |
| --full-output 才做深度还原、可视化、原始 NPZ 和图片写盘。 |
| |
| 板端用法（所有路径均为必填）： |
| # 单图验证 |
| python3 depthart_infer_axmodel.py --axmodel relative_s_448_default_ax650.axmodel \ |
| --image ./host.jpg --output ./results/ax_single |
| |
| # 数据集快速性能模式 |
| python3 depthart_infer_axmodel.py --axmodel relative_s_448_default_ax650.axmodel \ |
| --dataset ./images --output ./results/board |
| |
| # 数据集完整输出（原始深度 NPZ、可视化和大图深度） |
| python3 depthart_infer_axmodel.py --axmodel relative_s_448_default_ax650.axmodel \ |
| --dataset ../../test_depth --output ./results/board_full --full-output |
| |
| # 与开发机 ONNX 基准对比（--ref-npz 指向 ONNX 脚本的汇总 NPZ） |
| python3 depthart_infer_axmodel.py --axmodel relative_s_448_default_ax650.axmodel \ |
| --dataset ../../test_depth --output ./results/board_check --ref-npz ./onnx_npz |
| |
| 开发机用法（仅用于脚本语法/参数检查，实际推理需要板端 axengine）： |
| ../../venv/bin/python src/infer/depthart_infer_axmodel.py \ |
| --axmodel output_axonnx3/relative_s_448_default_ax650.axmodel \ |
| --dataset /path/images --output results/axmodel_board --limit 50 |
| |
| 诊断： |
| --debug 打印输入/输出 tensor 的 shape、dtype、min、max、mean、nan 计数 |
| --dump-npz 单图时为 .npz 文件路径；数据集时为汇总 .npz 文件路径， |
| 保存 names/depths/original_sizes，不再每张图一个小文件 |
| --npz-chunk 数据集模式下每 N 张切一个文件（xxx_0000.npz 等）， |
| 0 表示全部写入同一个文件 |
| --ref-npz 单图时为参考 .npz，数据集时为参考汇总 .npz（按 names 匹配）， |
| 报告 max/mean/p99 绝对误差、cosine 以及 scale-shift 对齐后误差 |
| """ |
|
|
| import argparse |
| import json |
| import logging |
| import time |
| from pathlib import Path |
|
|
| import cv2 |
| import numpy as np |
| from tqdm import tqdm |
|
|
| MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32) |
| STD = np.array([0.229, 0.224, 0.225], dtype=np.float32) |
| EXTENSIONS = {'.jpg', '.jpeg', '.png', '.bmp', '.tif', '.tiff'} |
| STAGES = ('read', 'preprocess', 'npu', 'diagnostics', |
| 'postprocess', 'visualize', 'write', 'total') |
|
|
|
|
| def meta_type(meta): |
| value = getattr(meta, 'type', None) |
| if value is None: |
| value = getattr(meta, 'dtype', '') |
| return str(value).lower() |
|
|
|
|
| def input_spec(meta): |
| """从运行时输入元信息推导尺寸、layout 和 dtype，不依赖固定名称。""" |
| try: |
| shape = [int(v) for v in meta.shape] |
| except (TypeError, ValueError): |
| raise ValueError(f'输入必须是静态 4D NCHW/NHWC: {meta.name} {meta.shape}') |
| if len(shape) != 4 or any(v <= 0 for v in shape): |
| raise ValueError(f'输入必须是静态 4D NCHW/NHWC: {meta.name} {meta.shape}') |
| if shape[0] != 1: |
| raise ValueError(f'仅支持 batch=1: {meta.name} {meta.shape}') |
| if shape[1] == 3: |
| layout, size = 'NCHW', (shape[2], shape[3]) |
| elif shape[3] == 3: |
| layout, size = 'NHWC', (shape[1], shape[2]) |
| else: |
| raise ValueError(f'输入必须含 3 通道: {meta.name} {meta.shape}') |
| type_name = meta_type(meta) |
| if 'float16' in type_name or 'fp16' in type_name: |
| dtype = np.float16 |
| elif 'float' in type_name or 'fp32' in type_name: |
| dtype = np.float32 |
| elif 'uint8' in type_name or type_name.endswith('u8'): |
| dtype = np.uint8 |
| elif 'int8' in type_name or type_name.endswith('s8'): |
| dtype = np.int8 |
| else: |
| raise ValueError(f'不支持的输入类型: {meta.name} {type_name!r}') |
| return size, layout, dtype |
|
|
|
|
| def preprocess(image, size, color_order, layout, dtype, norm_mode): |
| """按编译契约生成输入 tensor。 |
| |
| uint8 直送路径沿用官方 axmodel 流程，在 uint8 域直接 resize；ImageNet |
| 归一化路径严格按 ONNX 基准的顺序（float 域 /255 -> resize -> mean/std）。 |
| """ |
| target_h, target_w = size |
| if color_order == 'rgb': |
| image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) |
| if norm_mode == 'imagenet': |
| canvas = image.astype(np.float32) / 255.0 |
| canvas = cv2.resize(canvas, (target_w, target_h), |
| interpolation=cv2.INTER_CUBIC) |
| canvas = (canvas - MEAN) / STD |
| elif dtype != np.uint8: |
| canvas = cv2.resize(image.astype(np.float32), (target_w, target_h), |
| interpolation=cv2.INTER_CUBIC) |
| else: |
| canvas = cv2.resize(image, (target_w, target_h), |
| interpolation=cv2.INTER_CUBIC).astype(np.float32) |
| if dtype == np.uint8: |
| canvas = np.clip(np.rint(canvas), 0, 255).astype(np.uint8) |
| elif dtype == np.int8: |
| raise ValueError('当前脚本不支持 INT8 输入：缺少反量化 scale/zero-point') |
| else: |
| canvas = canvas.astype(dtype) |
| tensor = canvas.transpose(2, 0, 1) if layout == 'NCHW' else canvas |
| return np.ascontiguousarray(tensor[None]) |
|
|
|
|
| def select_output(outputs, metas, input_size): |
| """axmodel 输出名可能被编译器改写：先按名字，再按 shape 回退。""" |
| if not isinstance(outputs, (list, tuple)): |
| outputs = [outputs] |
| if len(outputs) != len(metas): |
| raise ValueError(f'输出数量不一致: run 返回 {len(outputs)}，' |
| f'接口声明 {len(metas)}') |
| for index, meta in enumerate(metas): |
| if 'depth' in str(meta.name).lower(): |
| return outputs[index], str(meta.name) |
| candidates = [] |
| for index, array in enumerate(outputs): |
| shape = np.asarray(array).shape |
| if len(shape) == 4 and shape[1] == 1 and tuple(shape[2:]) == tuple(input_size): |
| candidates.append(index) |
| elif len(shape) == 3 and tuple(shape[1:]) == tuple(input_size): |
| candidates.append(index) |
| if len(candidates) != 1: |
| actual = [(m.name, np.asarray(o).shape) for m, o in zip(metas, outputs)] |
| raise ValueError(f'无法按 name 或 shape 唯一匹配 depth 输出；实际输出: {actual}') |
| return outputs[candidates[0]], str(metas[candidates[0]].name) |
|
|
|
|
| def read_depth(raw, input_size): |
| array = np.asarray(raw) |
| if array.ndim == 4 and array.shape[1] == 1: |
| array = array[0] |
| if array.ndim == 3 and array.shape[0] == 1: |
| array = array[0] |
| if array.shape != tuple(input_size): |
| raise ValueError(f'depth 输出无效: shape={array.shape}，' |
| f'期望 {tuple(input_size)}') |
| if not np.issubdtype(array.dtype, np.floating): |
| raise ValueError(f'depth 输出应为浮点，实际 dtype={array.dtype}，' |
| '整数输出需要已知反量化参数') |
| array = array.astype(np.float32, copy=False) |
| if not np.isfinite(array).all(): |
| raise ValueError('depth 输出包含 nan/inf') |
| if float(array.std()) < 1e-6: |
| logging.warning('depth 输出近似常数（std=%.3g）：请检查输入归一化/颜色顺序' |
| '是否与编译配置的 input_processors 一致。', float(array.std())) |
| return np.ascontiguousarray(array) |
|
|
|
|
| def restore_depth(depth, original_size): |
| """按官方推理的 bilinear + align_corners=True 还原到原图尺寸。""" |
| height, width = original_size |
| src_h, src_w = depth.shape |
| if (src_h, src_w) == (height, width): |
| return depth.copy() |
| xs = np.linspace(0.0, src_w - 1, width, dtype=np.float32) |
| ys = np.linspace(0.0, src_h - 1, height, dtype=np.float32) |
| map_x, map_y = np.meshgrid(xs, ys) |
| return cv2.remap(depth, map_x, map_y, interpolation=cv2.INTER_LINEAR, |
| borderMode=cv2.BORDER_REPLICATE) |
|
|
|
|
| def visualize(image, depth): |
| finite = depth[np.isfinite(depth)] |
| low, high = np.percentile(finite, (2, 98)) |
| value = np.clip((depth - low) / max(float(high - low), 1e-6), 0, 1) |
| colored = cv2.applyColorMap((value * 255).astype(np.uint8), cv2.COLORMAP_INFERNO) |
| panel = np.concatenate((image, colored), axis=1) |
| legend = np.full((56, panel.shape[1], 3), 255, dtype=np.uint8) |
| ramp = np.tile(np.linspace(0, 255, max(panel.shape[1] - 20, 1), |
| dtype=np.uint8), (12, 1)) |
| legend[5:17, 10:10 + ramp.shape[1]] = cv2.applyColorMap(ramp, cv2.COLORMAP_INFERNO) |
| cv2.putText(legend, f'Image | Relative depth (2-98%): {low:.3f} -> {high:.3f}', |
| (10, 42), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (30, 30, 30), 1, cv2.LINE_AA) |
| return np.concatenate((panel, legend), axis=0) |
|
|
|
|
| def log_tensor(name, value): |
| logging.info('%s: shape=%s dtype=%s min=%.6g max=%.6g mean=%.6g nan=%d inf=%d', |
| name, value.shape, value.dtype, float(np.min(value)), |
| float(np.max(value)), float(np.mean(value)), |
| int(np.isnan(value).sum()), int(np.isinf(value).sum())) |
|
|
|
|
| def compare_arrays(depth, expected, source='', preprocessing=''): |
| expected = np.asarray(expected) |
| if expected.ndim == 4 and expected.shape[1] == 1: |
| expected = expected[0] |
| if expected.shape != depth.shape: |
| raise ValueError(f'参考 depth shape 不符: {expected.shape} vs {depth.shape} ({source})') |
| if not np.isfinite(expected).all(): |
| raise ValueError(f'参考 depth 含 nan/inf: {source}') |
| actual64 = depth.astype(np.float64).ravel() |
| expected64 = expected.astype(np.float64).ravel() |
| diff = np.abs(actual64 - expected64) |
| cosine = float(np.dot(actual64, expected64) |
| / (np.linalg.norm(actual64) * np.linalg.norm(expected64) + 1e-12)) |
| |
| slope, offset = np.polyfit(expected64, actual64, 1) |
| aligned = np.abs(actual64 - (slope * expected64 + offset)) |
| metrics = {'cosine': cosine, |
| 'max_abs': float(diff.max()), |
| 'mean_abs': float(diff.mean()), |
| 'p99_abs': float(np.percentile(diff, 99)), |
| 'scale': float(slope), 'offset': float(offset), |
| 'aligned_mean_abs': float(aligned.mean()), |
| 'aligned_p99_abs': float(np.percentile(aligned, 99)), |
| 'reference_preprocessing': preprocessing} |
| logging.info('参考对比: %s', metrics) |
| return metrics |
|
|
|
|
| def compare_reference(depth, reference_path): |
| with np.load(reference_path, allow_pickle=False) as reference: |
| if 'depth' not in reference: |
| raise ValueError(f'参考 NPZ 缺少 depth: {reference_path}') |
| expected = np.asarray(reference['depth']) |
| preprocessing = (str(reference['preprocessing']) |
| if 'preprocessing' in reference else '') |
| return compare_arrays(depth, expected, str(reference_path), preprocessing) |
|
|
|
|
| def load_reference_dataset(path): |
| """读取汇总 NPZ，返回 {name: depth}。""" |
| path = Path(path) |
| if path.is_file(): |
| with np.load(path, allow_pickle=False) as ref: |
| if 'depths' not in ref or 'names' not in ref: |
| raise ValueError(f'汇总 NPZ 需要 names/depths: {path}') |
| return {str(n): d for n, d in zip(ref['names'], ref['depths'])} |
| raise ValueError(f'数据集模式 --ref-npz 需要汇总 NPZ 文件: {path}') |
|
|
|
|
| def preprocessing_description(color_order, norm_mode): |
| norm = ('imagenet' if norm_mode == 'imagenet' else 'raw-0-255') |
| return f'color={color_order}; resize=stretch; norm={norm}' |
|
|
|
|
| def save_raw_npz(path, depth, original_size, input_size, color_order, norm_mode): |
| |
| np.savez(path, depth=depth, |
| original_size=np.array(original_size, dtype=np.int32), |
| input_size=np.array(input_size, dtype=np.int32), |
| preprocessing=preprocessing_description(color_order, norm_mode)) |
|
|
|
|
| def flush_dataset_npz(path, records, chunk_index, input_size, color_order, norm_mode): |
| """把缓冲区写成汇总 NPZ：names / depths / original_sizes。 |
| |
| 板端弱 CPU 上 DEFLATE 对 float32 收益极小却耗时明显，这里保持未压缩。 |
| chunk_index 为 None 时写入 path 本身，否则按 chunk 编号追加后缀。 |
| """ |
| if not records: |
| return |
| target = path if chunk_index is None else path.with_name( |
| f'{path.stem}_{chunk_index:04d}{path.suffix}') |
| target.parent.mkdir(parents=True, exist_ok=True) |
| np.savez( |
| target, |
| names=np.array([r[0] for r in records]), |
| depths=np.stack([r[1] for r in records]), |
| original_sizes=np.array([r[2] for r in records], dtype=np.int32), |
| input_size=np.array(input_size, dtype=np.int32), |
| preprocessing=preprocessing_description(color_order, norm_mode)) |
|
|
|
|
| def collect_images(dataset): |
| paths = sorted(p for p in dataset.rglob('*') |
| if p.is_file() and p.suffix.lower() in EXTENSIONS) |
| if not paths: |
| raise ValueError(f'目录中没有图片: {dataset}') |
| return paths |
|
|
|
|
| def summarize(samples): |
| result = {} |
| for stage in STAGES: |
| values = np.array([sample[stage] for sample in samples], dtype=np.float64) |
| result[stage] = {'mean_ms': float(values.mean()), |
| 'p50_ms': float(np.percentile(values, 50)), |
| 'p90_ms': float(np.percentile(values, 90)), |
| 'p99_ms': float(np.percentile(values, 99))} |
| return result |
|
|
|
|
| def run_single(session, input_meta, output_metas, input_size, layout, dtype, |
| norm_mode, args): |
| stem = args.image.stem |
| timings = dict.fromkeys(STAGES, 0.0) |
| total_start = time.perf_counter() |
|
|
| start = time.perf_counter() |
| image = cv2.imread(str(args.image), cv2.IMREAD_COLOR) |
| if image is None: |
| raise ValueError(f'读取图片失败: {args.image}') |
| timings['read'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| tensor = preprocess(image, input_size, args.color_order, layout, dtype, norm_mode) |
| timings['preprocess'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| outputs = session.run(None, {input_meta.name: tensor}) |
| timings['npu'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| raw, output_name = select_output(outputs, output_metas, input_size) |
| depth = read_depth(raw, input_size) |
| comparison = None |
| if args.debug: |
| log_tensor(input_meta.name, tensor) |
| log_tensor('depth', depth) |
| if args.ref_npz: |
| comparison = compare_reference(depth, args.ref_npz) |
| timings['diagnostics'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| restored = restore_depth(depth, image.shape[:2]) |
| timings['postprocess'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| panel = visualize(image, restored) |
| timings['visualize'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| args.output.mkdir(parents=True, exist_ok=True) |
| image_path = args.output / f'{stem}_depth.jpg' |
| if not cv2.imwrite(str(image_path), panel, [cv2.IMWRITE_JPEG_QUALITY, 90]): |
| raise OSError(f'保存可视化失败: {image_path}') |
| np.save(str(args.output / f'{stem}_depth.npy'), restored) |
| npz_path = None |
| if args.dump_npz: |
| npz_path = Path(args.dump_npz) |
| npz_path.parent.mkdir(parents=True, exist_ok=True) |
| save_raw_npz(npz_path, depth, image.shape[:2], input_size, |
| args.color_order, norm_mode) |
| timings['write'] = (time.perf_counter() - start) * 1000 |
| timings['total'] = (time.perf_counter() - total_start) * 1000 |
|
|
| cpu_ms = (timings['read'] + timings['preprocess'] + timings['diagnostics'] |
| + timings['postprocess'] + timings['visualize'] + timings['write']) |
| logging.info('深度（模型分辨率）: min=%.4f max=%.4f mean=%.4f p50=%.4f', |
| float(depth.min()), float(depth.max()), float(depth.mean()), |
| float(np.percentile(depth, 50))) |
| logging.info('深度（原图分辨率）: min=%.4f max=%.4f mean=%.4f', |
| float(restored.min()), float(restored.max()), float(restored.mean())) |
| logging.info('输出绑定: %s', output_name) |
| logging.info('NPU 推理: %.2f ms；CPU 处理: %.2f ms；端到端: %.2f ms' |
| '（单图为首次推理，含运行时首次执行开销；性能请用数据集模式）', |
| timings['npu'], cpu_ms, timings['total']) |
| logging.info('各阶段耗时: %s', |
| ' '.join(f'{key}={value:.2f} ms' for key, value in timings.items())) |
| if comparison is not None: |
| logging.info('参考对比结果: %s', comparison) |
| logging.info('结果目录: %s（可视化 %s%s）', args.output, image_path.name, |
| f'，原始 NPZ {npz_path.name}' if npz_path else '') |
|
|
|
|
| def run_dataset(session, input_meta, output_metas, input_size, layout, dtype, |
| norm_mode, runtime_version, args): |
| images = collect_images(args.dataset) |
| if args.limit is not None: |
| images = images[:args.limit] |
| args.output.mkdir(parents=True, exist_ok=True) |
| full_output = args.full_output |
| if full_output: |
| logging.warning('--full-output: CPU 侧深度还原、可视化与写盘会拉高端到端耗时，' |
| '不能只看 NPU 时间。') |
| else: |
| logging.info('数据集快速模式：只统计 read/preprocess/npu，' |
| '跳过深度还原、可视化和结果写盘。') |
| ref_map = load_reference_dataset(args.ref_npz) if args.ref_npz else None |
| preprocessing = preprocessing_description(args.color_order, norm_mode) |
| dump_path = Path(args.dump_npz) if args.dump_npz else None |
| if dump_path is not None and dump_path.is_dir(): |
| raise ValueError('数据集模式 --dump-npz 请传 .npz 文件路径，' |
| '不再按目录为每张图生成小文件') |
| dump_records, dump_chunk = [], 0 |
|
|
| first = cv2.imread(str(images[0]), cv2.IMREAD_COLOR) |
| if first is None: |
| raise ValueError(f'读取图片失败: {images[0]}') |
| warm = preprocess(first, input_size, args.color_order, layout, dtype, norm_mode) |
| warmup_ms = 0.0 |
| if args.warmup: |
| start = time.perf_counter() |
| for _ in range(args.warmup): |
| session.run(None, {input_meta.name: warm}) |
| warmup_ms = (time.perf_counter() - start) * 1000 |
|
|
| samples = [] |
| sample_info = [] |
| for path in tqdm(images, desc='AX 相对深度', unit='image'): |
| relative = path.relative_to(args.dataset) |
| timings = dict.fromkeys(STAGES, 0.0) |
| total_start = time.perf_counter() |
|
|
| start = time.perf_counter() |
| image = cv2.imread(str(path), cv2.IMREAD_COLOR) |
| if image is None: |
| raise ValueError(f'读取图片失败: {path}') |
| timings['read'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| tensor = preprocess(image, input_size, args.color_order, layout, dtype, norm_mode) |
| timings['preprocess'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| outputs = session.run(None, {input_meta.name: tensor}) |
| timings['npu'] = (time.perf_counter() - start) * 1000 |
|
|
| start = time.perf_counter() |
| raw, output_name = select_output(outputs, output_metas, input_size) |
| depth = read_depth(raw, input_size) |
| comparison = None |
| if args.debug and path == images[0]: |
| log_tensor(input_meta.name, tensor) |
| log_tensor('depth', depth) |
| if dump_path is not None: |
| dump_records.append((str(relative), depth, |
| np.array(image.shape[:2], dtype=np.int32))) |
| if args.npz_chunk and len(dump_records) >= args.npz_chunk: |
| flush_dataset_npz(dump_path, dump_records, dump_chunk, |
| input_size, args.color_order, norm_mode) |
| dump_records.clear() |
| dump_chunk += 1 |
| if ref_map is not None: |
| key = str(relative) |
| if key not in ref_map: |
| raise ValueError(f'参考 NPZ 缺少样本: {key}') |
| comparison = compare_arrays(depth, ref_map[key], |
| f'{args.ref_npz}:{key}', preprocessing) |
| timings['diagnostics'] = (time.perf_counter() - start) * 1000 |
|
|
| restored = None |
| if full_output: |
| start = time.perf_counter() |
| restored = restore_depth(depth, image.shape[:2]) |
| timings['postprocess'] = (time.perf_counter() - start) * 1000 |
| start = time.perf_counter() |
| panel = visualize(image, restored) |
| timings['visualize'] = (time.perf_counter() - start) * 1000 |
| start = time.perf_counter() |
| prefix = args.output / relative |
| prefix.parent.mkdir(parents=True, exist_ok=True) |
| if not cv2.imwrite(str(prefix) + '.depth.jpg', panel, |
| [cv2.IMWRITE_JPEG_QUALITY, 90]): |
| raise OSError(f'保存可视化失败: {prefix}.depth.jpg') |
| timings['write'] = (time.perf_counter() - start) * 1000 |
| timings['total'] = (time.perf_counter() - total_start) * 1000 |
|
|
| samples.append(timings) |
| sample_info.append({'name': str(relative), |
| 'original_size': list(image.shape[:2]), |
| 'depth_min': float((restored if restored is not None |
| else depth).min()), |
| 'depth_max': float((restored if restored is not None |
| else depth).max()), |
| 'timing_ms': timings, |
| 'reference_comparison': comparison}) |
|
|
| if dump_path is not None: |
| flush_dataset_npz(dump_path, dump_records, |
| None if not args.npz_chunk else dump_chunk, |
| input_size, args.color_order, norm_mode) |
|
|
| stats = summarize(samples) |
| total_ms = sum(sample['total'] for sample in samples) |
| cpu_stages = ('read', 'preprocess', 'diagnostics', 'postprocess', 'visualize', 'write') |
| cpu_total_ms = sum(stats[stage]['mean_ms'] for stage in cpu_stages) |
| report = {'model': str(Path(args.axmodel).resolve()), |
| 'runtime': runtime_version, |
| 'task': 'relative-depth', |
| 'input_name': input_meta.name, 'output_name': output_name, |
| 'input_size': list(input_size), 'input_layout': layout, |
| 'input_dtype': np.dtype(dtype).name, |
| 'color_order': args.color_order, |
| 'preprocessing': preprocessing_description(args.color_order, norm_mode), |
| 'batch_size': 1, 'sample_count': len(samples), |
| 'full_output': full_output, 'warmup_count': args.warmup, |
| 'warmup_ms': warmup_ms, 'total_ms': total_ms, |
| 'fps': len(samples) * 1000 / total_ms, |
| 'npu_mean_ms': stats['npu']['mean_ms'], |
| 'cpu_total_mean_ms': cpu_total_ms, |
| 'timing_scope': 'npu=仅 session.run；cpu=读取+预处理+输出提取/校验+' |
| '诊断+后处理+可视化+写盘；total=单样本端到端。排除' |
| '模型加载、session 初始化和 warmup。快速模式的 ' |
| 'postprocess/visualize/write 记为 0。', |
| 'stages': stats, 'samples': sample_info} |
| with (args.output / 'npu_timing.json').open('w', encoding='utf-8') as file: |
| json.dump(report, file, ensure_ascii=False, indent=2, allow_nan=False) |
| for stage, values in stats.items(): |
| logging.info('%s: mean=%.2f p50=%.2f p90=%.2f p99=%.2f ms', |
| stage, values['mean_ms'], values['p50_ms'], |
| values['p90_ms'], values['p99_ms']) |
| logging.info('%d 张图片，NPU mean=%.2f ms，CPU mean=%.2f ms，' |
| 'total=%.2f s，FPS=%.3f，结果目录: %s', |
| len(samples), stats['npu']['mean_ms'], cpu_total_ms, |
| total_ms / 1000, report['fps'], args.output) |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser( |
| description='DepthART 相对深度 axmodel 板端推理（AX650）', |
| formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__) |
| parser.add_argument('--axmodel', type=Path, required=True, help='编译后的 axmodel') |
| source = parser.add_mutually_exclusive_group(required=True) |
| source.add_argument('--image', type=Path, help='单张图片') |
| source.add_argument('--dataset', type=Path, help='图片目录（递归）') |
| parser.add_argument('--output', type=Path, required=True, help='结果目录') |
| parser.add_argument('--color-order', choices=('rgb', 'bgr'), default='rgb', |
| help='rgb=与官方预处理一致（BGR->RGB）；bgr=按编译配置' |
| 'input_processors.src_format 直送') |
| parser.add_argument('--input-norm', choices=('auto', 'raw', 'imagenet'), |
| default='auto', |
| help='auto=U8 送 0-255、浮点用 ImageNet mean/std；' |
| 'raw=直送 0-255；imagenet=/255+mean/std（仅浮点输入）') |
| parser.add_argument('--warmup', type=int, default=3, |
| help='数据集模式 warmup 次数，默认 3') |
| parser.add_argument('--limit', type=int, help='数据集模式只处理前 N 张') |
| parser.add_argument('--full-output', action='store_true', |
| help='数据集模式执行完整后处理、可视化和结果写盘（含 NPZ）') |
| parser.add_argument('--debug', action='store_true', help='打印输入/输出统计') |
| parser.add_argument('--dump-npz', |
| help='单图：NPZ 文件；数据集：汇总 NPZ 文件（names/depths/original_sizes）') |
| parser.add_argument('--npz-chunk', type=int, default=0, |
| help='数据集模式每 N 张切一个 NPZ 文件，0 表示全部写入一个文件') |
| parser.add_argument('--ref-npz', type=Path, help='参考 NPZ 文件或目录') |
| args = parser.parse_args() |
| logging.basicConfig(level=logging.INFO, format='%(message)s') |
| if args.warmup < 0: |
| parser.error('--warmup 不能为负数') |
| if args.npz_chunk < 0: |
| parser.error('--npz-chunk 不能为负数') |
| if args.limit is not None and args.limit < 1: |
| parser.error('--limit 必须为正数') |
| if args.image: |
| if not args.image.is_file(): |
| parser.error(f'--image 不是文件: {args.image}') |
| if args.ref_npz and not args.ref_npz.is_file(): |
| parser.error('单图模式 --ref-npz 必须是文件') |
| if args.limit is not None or args.full_output: |
| logging.warning('单图模式忽略 --limit/--full-output') |
| else: |
| if not args.dataset.is_dir(): |
| parser.error(f'--dataset 不是目录: {args.dataset}') |
|
|
| try: |
| import axengine |
| except ImportError as error: |
| raise SystemExit('未找到 axengine：该脚本需在板端运行（pip 安装 axengine/' |
| 'pyaxengine 后重试）') from error |
|
|
| session = axengine.InferenceSession(str(args.axmodel)) |
| inputs = session.get_inputs() |
| output_metas = session.get_outputs() |
| logging.info('模型输入: %s', [(m.name, m.shape, meta_type(m)) for m in inputs]) |
| logging.info('模型输出: %s', [(m.name, m.shape, meta_type(m)) for m in output_metas]) |
| if len(inputs) != 1: |
| raise ValueError(f'模型应有 1 个输入，实际 ' |
| f'{[(m.name, m.shape) for m in inputs]}') |
| input_meta = inputs[0] |
| input_size, layout, dtype = input_spec(input_meta) |
| if dtype == np.int8: |
| raise ValueError('当前脚本不支持 INT8 输入（缺少反量化参数）；请把编译配置的 ' |
| 'input_processors.src_dtype 设为 U8 后重编译') |
|
|
| if args.input_norm == 'auto': |
| norm_mode = 'raw' if dtype == np.uint8 else 'imagenet' |
| else: |
| norm_mode = args.input_norm |
| if dtype == np.uint8 and norm_mode == 'imagenet': |
| raise ValueError('输入类型为 uint8，不能在推理时做 ImageNet 归一化；' |
| '归一化必须由编译前端 input_processors 或校准口径保证') |
| if dtype != np.uint8 and norm_mode == 'raw': |
| logging.warning('输入为浮点但按 0-255 直送（--input-norm raw），' |
| '仅用于诊断，需与编译前端/校准口径一致。') |
|
|
| logging.info('batch=1，输入尺寸: %s，layout=%s，dtype=%s，颜色顺序=%s，' |
| '预处理=%s，resize=stretch', input_size, layout, |
| np.dtype(dtype).name, args.color_order, |
| 'ImageNet mean/std (/255)' if norm_mode == 'imagenet' |
| else '0-255 原始像素') |
| if dtype == np.uint8 and norm_mode == 'raw': |
| logging.info('提示：uint8 输入按编译契约送 0-255；ImageNet 归一化应由 ' |
| 'config_axonnx.json 的 input_processors 在模型内完成。') |
|
|
| runtime_version = getattr(axengine, '__version__', 'unknown') |
| if args.image: |
| run_single(session, input_meta, output_metas, input_size, layout, dtype, |
| norm_mode, args) |
| else: |
| run_dataset(session, input_meta, output_metas, input_size, layout, dtype, |
| norm_mode, runtime_version, args) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
