#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 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)) # 相对深度存在全局 scale/offset 漂移，对齐后误差更能反映结构偏差。 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): # 板端 CPU 弱，savez_compressed 的 DEFLATE 对 float32 收益极小却耗时明显，用未压缩保存。 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()