#!/usr/bin/env python3 # -*- coding: utf-8 -*- """滤光片外观质检 · 模块化推理 v4(只依赖 onnxruntime + numpy + pillow,xlsx 需可选的 openpyxl) 一句话契约 ---------- 一个目录 = 一个模型包 = 一个 onnx + 一个 pack.json。 换模型 = 往 models/ 拷一个目录 + 改一行 ACTIVE.json,【不用重新打包程序】。 任何一处配置与实际行为不符,程序在推理开始前带编号退出,绝不产出结果。 常用 ---- filter-predict-v4 --list 看有哪些模型包 filter-predict-v4 -i imgs/ -o r.xlsx 判废 filter-predict-v4 -i imgs/ -o r.xlsx --with-type 判废 + 缺陷类型 filter-predict-v4 -i imgs/ -o r.xlsx --preset 保守 filter-predict-v4 --verify 换模型后先自检 filter-predict-v4 --compare A B -i 回归集/ -o 对拍.xlsx filter-predict-v4 --activate reject=<包id> --note "张三 回归通过" 退出码:0 正常 / 2 用法 / 3 找不到包或歧义 / 4 manifest / 5 文件完整性 / 6 图契约 7 自检 / 8 无阈值 / 9 解码 / 10 session / 11 无图片 / 12 写文件 """ import argparse import os import sys import time import numpy as np from filterpack import pack as PK from filterpack import report as RP from filterpack.errors import (PackError, EXIT_USAGE, EXIT_DISCOVER, EXIT_NOIMG, EXIT_OK) from filterpack.preprocess import PP_ENGINE VERSION = '4.0.0' IMG_EXT = {'.png', '.jpg', '.jpeg', '.bmp', '.tif', '.tiff'} def _need_ort(): try: import onnxruntime as ort except Exception as e: sys.exit(f'导入 onnxruntime 失败: {type(e).__name__}: {e}\n脚本版请先 pip install onnxruntime') return ort def die(e): print(f'\n✘ [退出码{e.code}] {e}\n', file=sys.stderr) sys.exit(e.code) def collect(inp): if os.path.isfile(inp): return [inp] fs = [] for r, _, names in os.walk(inp): for n in sorted(names): if os.path.splitext(n)[1].lower() in IMG_EXT: fs.append(os.path.join(r, n)) return sorted(fs) # ================================================================ 加载辅助 class Loader: """按 onnx 绝对路径缓存 session。 缓存键必须是 onnx 路径而不是包目录:v1 平铺模式下 filter_binary 与 filter_4class 共享同一个目录,按目录缓存会让后加载的 session 冒充前者——判废 stage 会实际拿到 四分类模型,正是这套设计要消灭的那类静默串包。 """ def __init__(self, ort, args): self.ort, self.args = ort, args self.cache = {} self.packs = PK.discover(args.models_dir) self.active, self.active_path = PK.read_active(args.models_dir) def load(self, found): key = os.path.abspath(found.model_path) if key not in self.cache: p = PK.Pack(found, self.ort) if self.args.strict and p.is_v1: raise PackError( f'--strict: 拒绝 v1 降级包 {p.id}({p.dir})。\n' f' v1 包没有 sha256 也没有 golden fixture,无法检测"只换了 onnx"' f'和"norm 配错"这两类静默失败。\n' f' 迁移: python tools/make_pack.py from-v1 <包目录> --id ... --role ...', EXIT_DISCOVER) self.cache[key] = p return self.cache[key] def pick(self, role, explicit): return PK.select_for_role(role, self.packs, self.active, explicit, self.args.models_dir) def run_selftest(p, args): """返回 (st, skipped)。--no-selftest 必须同时设 FILTER_UNSAFE=1。""" if args.no_selftest: return None, True return p.selftest(), False # ================================================================ --list def cmd_list(args, ort): here, bundled = PK.base_dirs() lo = Loader(ort, args) print(f'程序目录 HERE = {here} 内置资源 = {bundled}') print('搜索顺序: ' + ' -> '.join(f'{r}[{s}]' for r, s, _, _ in PK.search_roots(args.models_dir))) print() print(RP.pad('包名', 34) + RP.pad('来源', 24) + RP.pad('role', 8) + RP.pad('size', 6) + RP.pad('norm', 8) + RP.pad('类别', 6) + RP.pad('阈值', 20) + '自检') for f in lo.packs: f.peek() if f.error: print(RP.pad(f.id, 34) + RP.pad(f.source, 24) + RP.pad('-', 8) + RP.pad('-', 6) + RP.pad('-', 8) + RP.pad('-', 6) + RP.pad('-', 20) + f'✘ 退出码{f.error.code}: {str(f.error).splitlines()[0][:60]}') else: m = f.mf thr = m['decision'].get('threshold') npre = len(m['decision'].get('presets') or {}) thr_s = ('【未标定】' if thr is None else f'{float(thr):.4f}') + (f' ({npre}预设)' if npre else '') st_s = '?' try: p = lo.load(f) st, _ = run_selftest(p, args) st_s = '⚠ v1降级:无 fixture' if st is None else f'✔ {st["n"]}/{st["n"]}' except PackError as e: st_s = f'✘ 退出码{e.code}: {str(e).splitlines()[0][:50]}' act = '' if lo.active: for role, pid in lo.active.items(): if pid == f.id: act = f' <- ACTIVE.{role}' print(RP.pad(f.id, 34) + RP.pad(f.source, 24) + RP.pad(m['pack']['role'], 8) + RP.pad(m['input']['size'], 6) + RP.pad(m['input']['norm']['name'], 8) + RP.pad(len(m['output']['classes']), 6) + RP.pad(thr_s, 20) + st_s + act) print(' ' * 34 + f.dir) for src, d in f.shadowed: print(' ' * 36 + f'↓ 被覆盖: 同名包 [{src}] {d}') print() if lo.active: print(f'当前生效: ' + ' '.join(f'{k}={v}' for k, v in lo.active.items() if k in ('reject', 'type')) + f' (来自 {lo.active_path})') else: print(f'当前生效: 没有 {PK.ACTIVE_NAME},每个角色必须恰好只有一个候选包,否则退出码 3') return EXIT_OK # ================================================================ --info / --show-presets def cmd_info(args, ort): import json lo = Loader(ort, args) f = PK.resolve_spec(args.info, lo.packs, args.models_dir) f.peek() if f.error: raise f.error print(f'{f.id} [{f.source}]\n{f.dir}\n') print(json.dumps(f.mf, ensure_ascii=False, indent=1)) return EXIT_OK def cmd_show_presets(args, ort): lo = Loader(ort, args) f = PK.resolve_spec(args.show_presets, lo.packs, args.models_dir) f.peek() if f.error: raise f.error d = f.mf['decision'] print(f'{f.id} role={f.mf["pack"]["role"]} norm={f.mf["input"]["norm"]["name"]}') print(f' decision.threshold = {d.get("threshold")}') ps = d.get('presets') or {} if not ps: print(' 该包没有标定任何预设。') for k, v in ps.items(): rest = ' '.join(f'{a}={b}' for a, b in v.items() if a != 'threshold') print(f' --preset {RP.pad(k, 12)}threshold={float(v["threshold"]):.4f} {rest}') cal = d.get('calibration') or {} if cal: print(' 标定信息:') for k, v in cal.items(): print(f' {k}: {v}') return EXIT_OK # ================================================================ --activate def cmd_activate(args, ort): """原子切换现役模型。 这是整套工具里最危险的一个动作,所以它必须做三件事而不只是写文件: 1. 切换前把【全部闸门】跑一遍(结构/manifest/完整性/图契约/自检)—— 否则可以把产线切到一个自检必然失败的包,故障要等下一次真跑(rc=7)才暴露, 而那多半是夜班交接之后最不合适的时刻。 2. 写到一个 read_active() 一定读得回来、且不会随进程消失的位置。 3. 写完【回读校验】,确认这次切换真的生效。 """ if '=' not in args.activate: raise PackError('--activate 的格式是 role=包id,例如 --activate reject=reject-effb0-imnet-320-v1', EXIT_USAGE) role, pid = args.activate.split('=', 1) if role not in ('reject', 'type'): raise PackError(f'--activate 的 role 只能是 reject 或 type,得到 {role!r}', EXIT_USAGE) lo = Loader(ort, args) f = PK.resolve_spec(pid, lo.packs, args.models_dir) f.peek() if f.error: raise f.error if f.mf['pack']['role'] != role: raise PackError(f'包 {f.id} 的 role 是 {f.mf["pack"]["role"]},不能激活成 {role}', EXIT_USAGE) # ---- 切换前跑全部闸门(等价于 --verify),任一不过就拒绝切换 print(f'切换前校验 {f.dir}/ …') try: p = lo.load(f) # G0~G5 st, skipped = run_selftest(p, args) # G6 except PackError as e: raise PackError(f'拒绝切换:模型包 {f.id} 没通过加载闸门,切上去下一次运行必然失败。\n' f' 原始错误(退出码 {e.code}):\n {e}', e.code) print(' 闸门 ✔ ' + ('(--no-selftest:自检已跳过)' if skipped else f'自检 {st["n"]}/{st["n"]}' if st else 'v1 降级包无 fixture')) if role == 'reject': try: thr, tsrc = p.threshold() print(f' 阈值 ✔ {thr:.4f} ({tsrc})') except PackError as e: # 不拦:阈值可以每次运行用 --threshold 给。但必须让人当场看见。 print(f' ⚠ 该包没有标定过的阈值 —— 切上去以后,不带 --threshold/--preset 直接跑会退出码 8。\n' f' {str(e).splitlines()[0]}', file=sys.stderr) root = PK.active_target(args.models_dir) path, cur, read_back = PK.write_active(root, role, f.id, args.note or '', args.models_dir) print(f'✔ 已写入 {path}(回读校验 ✔ 程序确实会读这一份)') for k, v in cur.items(): print(f' {k}: {v}') print(f' 提示:切换现役模型前应先跑 --compare <旧包> {f.id} -i 回归集/ -o 对拍.xlsx') return EXIT_OK # ================================================================ --verify def cmd_verify(args, ort): lo = Loader(ort, args) targets = ([PK.resolve_spec(args.verify, lo.packs, args.models_dir)] if args.verify is not True else lo.packs) bad, worst = 0, EXIT_OK for f in targets: print(f'\n校验 {f.dir}/') try: f.peek() if f.error: raise f.error print(f' [1/8] 目录结构 ✔ {os.path.basename(f.manifest_path)} 存在' + ('' if f.is_v1 else ',pack.id == 目录名')) m = f.mf print(f' [2/8] manifest ' + ('⚠ v1 降级包(无 schema,按训练真相推断)' if f.is_v1 else f'✔ schema={m["schema"]} ' f'pp_engine={m["runtime"]["pp_engine"]} ' f'role={m["pack"]["role"]} 无未知键')) g = m['input']['geometry'] print(f' [3/8] 注册表 ✔ geom={g["mode"]} interp={g["interpolation"]} ' f'norm={m["input"]["norm"]["name"]} act={m["output"]["activation"]}') p = lo.load(f) hs = p.verify_hashes() for name, want, got, ok in hs: sz = os.path.getsize(os.path.join(p.dir, name)) if want is None: print(f' [4/8] 文件完整性 ⚠ {name} {sz:,} B sha256 {got[:12]}…(v1 包无基准可比)') elif ok: print(f' [4/8] 文件完整性 ✔ {name} {sz:,} B sha256 {got[:12]}… 一致') else: raise PackError(f'{name} 的 sha256 与 manifest 不符\n' f' manifest {want}\n 实际 {got}', 5) print(f' [5/8] 运行时 ✔ onnxruntime {ort.__version__} >= ' f'{(m.get("runtime") or {}).get("min_onnxruntime", "(未声明)")}') print(f' [6/8] onnx 图契约 ✔ in{p.in_shape} -> out{p.out_shape} == ' f'len(classes)={len(p.classes)} good_index={p.good_index}') st, skipped = run_selftest(p, args) if st is None: print(f' [7/8] golden fixture ⚠ ' + ('已用 --no-selftest 跳过' if skipped else 'v1 降级包没有 fixture,无法自动验证预处理')) else: print(f' [7/8] golden fixture ✔ {st["n"]}/{st["n"]}') for r in st['rows']: print(f' {r["name"]:<14} canvas ✔ norm Δ={r["d_norm"]:.1e} ' f'logits Δ={r["d_lg"]:.1e} score Δ={r["d_sc"]:.1e}') if p.role == 'reject': try: thr, src = p.threshold() npre = len(p.decision.get('presets') or {}) print(f' [8/8] 阈值可用 ✔ {thr:.4f} ({src})' + (f' + {npre} 个预设' if npre else '')) except PackError as e: print(f' [8/8] 阈值可用 ✘ {str(e).splitlines()[0]}') raise else: print(f' [8/8] 阈值可用 n/a (role=type 用 argmax,不需要阈值)') if f.is_v1: print('校验完成,但这是 v1 降级包:缺 sha256 与 golden fixture 两道保护,' '产线请先迁移成 v2 包(--strict 会直接拒绝它)。') else: print('校验通过,可以投产。') except PackError as e: bad += 1 worst = e.code # 退出码要如实反映失败在哪一关(8=没标定阈值 ≠ 7=自检没过) print(f' ✘ [退出码{e.code}] {e}', file=sys.stderr) return EXIT_OK if bad == 0 else worst # ================================================================ --compare def cmd_compare(args, ort): if not args.input: raise PackError('--compare 需要 -i 指定对拍用的图片目录', EXIT_USAGE) lo = Loader(ort, args) files = collect(args.input) if not files: raise PackError(f'没找到图片:{args.input}', EXIT_NOIMG) banner = RP.header_lines(VERSION, ort.__version__, PK.base_dirs()[0]) out_lines = list(banner) res = [] for tag, spec in zip('AB', args.compare): f = PK.resolve_spec(spec, lo.packs, args.models_dir) p = lo.load(f) PK.assert_slot(p, 'reject') # 对拍的是判废判定与阈值,类型包没有可比的"不良分" st, skipped = run_selftest(p, args) thr, tsrc = p.threshold() out_lines += RP.pack_lines(tag, p, thr, tsrc, st, '命令行 --compare', skipped) P, s = p.infer(files, args.batch) res.append((p, thr, tsrc, s)) print('\n'.join(out_lines)) (pa, ta, sa_src, sa), (pb, tb, sb_src, sb) = res va, vb = sa >= ta, sb >= tb diff = va != vb n = len(files) lines = [ '', f'A = {pa.id:<32} norm={pa.input["norm"]["name"]:<7} thr={ta:.4f} ({sa_src})', f'B = {pb.id:<32} norm={pb.input["norm"]["name"]:<7} thr={tb:.4f} ({sb_src})', ' 两包各按自己标定的阈值判定。', '', f'共 {n} 张 A 判不良 {int(va.sum())} B 判不良 {int(vb.sum())} ' f'分歧 {int(diff.sum())} 张 ({diff.mean():.2%})', f' A良→B不良 {int((~va & vb).sum())} 张', f' A不良→B良 {int((va & ~vb).sum())} 张', f' |Δscore| max={np.abs(sa - sb).max():.4f} mean={np.abs(sa - sb).mean():.4f} ' f'p99={np.percentile(np.abs(sa - sb), 99):.4f}', ] print('\n'.join(lines)) out_lines += lines header = ['图片名', 'A判定', 'A不良概率', 'B判定', 'B不良概率', '分歧', '相对路径'] rows, dr = [], [] for i, fp in enumerate(files): rel = os.path.relpath(fp, args.input) if os.path.isdir(args.input) else fp r = [os.path.basename(fp), '不良' if va[i] else '良品', round(float(sa[i]), 4), '不良' if vb[i] else '良品', round(float(sb[i]), 4), '★' if diff[i] else '', rel] rows.append(r) if diff[i]: dr.append(r) if args.output: o = RP.write_table(rows, header, args.output, out_lines, '对拍结果', [('分歧', header, dr)]) print(f' 结果 -> {o}') return EXIT_OK # ================================================================ 主流程 def cmd_run(args, ort): files = collect(args.input) if not files: raise PackError(f'没找到图片:{args.input}', EXIT_NOIMG) lo = Loader(ort, args) here, _ = PK.base_dirs() banner = RP.header_lines(VERSION, ort.__version__, here) f, src = lo.pick('reject', args.model) p = lo.load(f) PK.assert_slot(p, 'reject') # 槽位校验:-m 指的必须是判废包,不是类型包 st, skipped = run_selftest(p, args) thr, tsrc = p.threshold(args.threshold, args.preset) banner.append('') banner += RP.pack_lines('判废', p, thr, tsrc, st, src, skipped) tp = None if args.with_type: tf, tsrc2 = lo.pick('type', args.type_model) tp = lo.load(tf) PK.assert_slot(tp, 'type') # 槽位校验:--type-model 指的必须是类型包 tst, tskip = run_selftest(tp, args) banner.append('') banner += RP.pack_lines('类型', tp, None, None, tst, tsrc2, tskip) if skipped: banner.insert(0, RP.UNSAFE_BANNER) banner.append('') banner.append(f'待检 {len(files)} 张') print('\n'.join(banner)) t0 = time.time() P, score = p.infer(files, args.batch) verdict = np.where(score >= thr, '不良', '良品') types = tconf = None if tp is not None: TP, _ = tp.infer(files, args.batch) min_conf = (args.type_min_conf if args.type_min_conf is not None else float((tp.decision.get('type_rule') or {}).get('min_conf', 0.0))) tconf = TP.max(1) types = [tp.classes[i] if c >= min_conf else '不确定' for i, c in zip(TP.argmax(1), tconf)] dt = time.time() - t0 n_bad = int((verdict == '不良').sum()) tail = [f'完成 {len(files)} 张,用时 {dt:.1f}s({len(files) / max(dt, 1e-9):.1f} 张/秒)', f' 良品 {len(files) - n_bad} 张 | 不良 {n_bad} 张({n_bad / len(files):.1%})'] print('\n'.join(tail)) if len(files) == 1 and not args.output: print(f'\n 文件 : {os.path.basename(files[0])}') print(f' 判定 : {verdict[0]}') print(f' 不良概率 : {score[0]:.4f} (阈值 {thr:.4f})') if types: print(f' 缺陷类型 : {types[0]}(置信度 {tconf[0]:.3f})') return EXIT_OK suffix = '(自检已跳过)' if skipped else '' header = ['图片名', '判定', '不良概率'] rows = [] for i, fp in enumerate(files): r = [os.path.basename(fp), str(verdict[i]), round(float(score[i]), 4)] if types: bad = verdict[i] == '不良' r += [types[i] if bad else '', round(float(tconf[i]), 3) if bad else ''] r += [p.id + suffix, round(thr, 4)] r.append(os.path.relpath(fp, args.input) if os.path.isdir(args.input) else fp) rows.append(r) if types: header += ['缺陷类型', '类型置信度'] header += ['模型包', '阈值', '相对路径'] out = RP.write_table(rows, header, args.output or 'result.xlsx', banner + tail) print(f' 结果 -> {out}') return EXIT_OK # ================================================================ CLI def build_parser(): ap = argparse.ArgumentParser(description='滤光片外观质检推理 v4(模型包热插拔版)', formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument('-i', '--input', help='图片文件或目录(递归)') ap.add_argument('-o', '--output', default=None, help='输出 .xlsx / .csv(目录模式必填)') ap.add_argument('-m', '--model', '--pack', dest='model', default=None, help='判废模型包: 包id / 包目录 / .onnx 路径') ap.add_argument('--type-model', '--type-pack', dest='type_model', default=None, help='类型模型包') ap.add_argument('--with-type', action='store_true', help='同时给出缺陷类型') ap.add_argument('--threshold', type=float, default=None, help='判废阈值(不良概率)') ap.add_argument('--preset', default=None, help='用包内标定好的预设,如 保守/平衡/高召回') ap.add_argument('--type-min-conf', type=float, default=None, help='类型置信度低于此值输出「不确定」') ap.add_argument('--batch', type=int, default=16) ap.add_argument('--models-dir', action='append', default=[], help='追加模型搜索根目录,可重复') ap.add_argument('--strict', action='store_true', help='拒绝一切 v1 降级包(产线推荐常开)') ap.add_argument('--no-selftest', action='store_true', help='跳过 golden fixture 自检(必须同时设环境变量 FILTER_UNSAFE=1)') ap.add_argument('--list', action='store_true', help='列出全部模型包(含被覆盖的)') ap.add_argument('--info', metavar='ID', help='打印该包完整 manifest') ap.add_argument('--verify', nargs='?', const=True, metavar='ID', help='全闸门校验,不推理') ap.add_argument('--show-presets', metavar='ID', help='打印预设表') ap.add_argument('--compare', nargs=2, metavar=('A', 'B'), help='两个包在同一批图上对拍') ap.add_argument('--activate', metavar='role=ID', help=f'原子写 models/{PK.ACTIVE_NAME}') ap.add_argument('--note', default=None, help='配合 --activate 写入审计备注') ap.add_argument('--version', action='version', version=f'filter-predict v{VERSION} (像素链引擎 {PP_ENGINE})') return ap def usage(msg): print(f'✘ [退出码{EXIT_USAGE}] {msg}', file=sys.stderr) sys.exit(EXIT_USAGE) def main(): ap = build_parser() args = ap.parse_args() if args.threshold is not None and args.preset is not None: usage('--threshold 与 --preset 只能给一个(阈值来源必须唯一可追溯)') if args.threshold is not None and not (0.0 <= args.threshold <= 1.0 and args.threshold == args.threshold): # 与 manifest 里 decision.threshold 同一把闸。--threshold 1.5 会把全部产品判成良品, # --threshold -1 会全判不良,两者原来都是退出码 0 静默跑完。 usage(f'--threshold {args.threshold} 超出 [0,1]。判废分是不良【概率】,阈值必须落在 0~1。\n' f' --threshold 1.5 会把所有产品判成良品(漏检 100%),-1 则全判不良,\n' f' 这两种都不该是"跑完了、退出码 0"。常见原因是小数点手滑(0.8176 -> 8176)。') if args.type_min_conf is not None and not (0.0 <= args.type_min_conf <= 1.0): usage(f'--type-min-conf {args.type_min_conf} 超出 [0,1](它是 softmax 置信度)。' f'大于 1 会把每一张都判成「不确定」。') # -m/--type-model 在这些子命令下是无意义的,会被静默丢弃 —— 而"我要校验刚拷进来的这个包" # 写成 `-m --verify` 是很自然的手误,静默丢弃的后果是用户看到一串"校验通过,可以投产", # 而他真正关心的那个包可能根本没被扫到,退出码也来自完全不相干的包。 for flag, on in (('--verify', bool(args.verify)), ('--info', bool(args.info)), ('--show-presets', bool(args.show_presets)), ('--activate', bool(args.activate)), ('--compare', bool(args.compare))): if on and (args.model or args.type_model): given = ' '.join(x for x in (f'-m {args.model}' if args.model else '', f'--type-model {args.type_model}' if args.type_model else '') if x) usage(f'{flag} 不接受 -m/--model / --type-model(给了: {given})。\n' f' 这两个开关只在真正推理时决定用哪个包;{flag} 自己带包参数。\n' + (f' 你要的多半是: --verify {args.model or args.type_model}\n' if flag == '--verify' else '') + f' (静默忽略的后果:你会看到一串"校验通过,可以投产",' f'而那可能根本不是你要查的那个包。)') if args.no_selftest and os.environ.get('FILTER_UNSAFE') != '1': usage('--no-selftest 必须同时设环境变量 FILTER_UNSAFE=1。\n' ' 自检是"换模型不重打包"这套机制唯一的安全网,跳过它的结果不得用于出货判定。') ort = _need_ort() try: if args.list: sys.exit(cmd_list(args, ort)) if args.info: sys.exit(cmd_info(args, ort)) if args.show_presets: sys.exit(cmd_show_presets(args, ort)) if args.activate: sys.exit(cmd_activate(args, ort)) if args.verify: sys.exit(cmd_verify(args, ort)) if args.compare: sys.exit(cmd_compare(args, ort)) if not args.input: print('✘ [退出码2] 需要 -i/--input(或用 --list / --verify / --info / --compare)', file=sys.stderr) sys.exit(EXIT_USAGE) sys.exit(cmd_run(args, ort)) except PackError as e: die(e) if __name__ == '__main__': main()