import argparseimport jsonimport osfrom dataclasses import dataclassfrom typing import Dict, List, Tupleimport numpy as npfrom PIL import ImageSEP # txt中图片路径和标签路径使用三个空格分隔dataclassclass Sample:一行样本记录。image_path: txt左侧图片路径mask_path: txt右侧标签路径source_txt: 样本来源txt文件line_no: 行号便于定位异常image_path: strmask_path: strsource_txt: strline_no: intdef _normalize_path(path: str, case_insensitive: bool) - str:# 统一路径分隔符去掉重复斜杠可选大小写不敏感s path.strip().replace(\\, /)while // in s:s s.replace(//, /)return s.lower() if case_insensitive else sdef _make_key(path: str, match_mode: str, case_insensitive: bool) - str:# fullpath: 用完整路径匹配basename: 用文件名匹配n _normalize_path(path, case_insensitive)if match_mode basename:return os.path.basename(n)return ndef parse_txt(txt_path: str) - List[Sample]:读取一个数据txt。预期格式每行 image_path mask_path三个空格分隔。samples: List[Sample] []with open(txt_path, r, encodingutf-8) as f:for i, raw in enumerate(f, start1):line raw.strip()if not line:continueparts line.split(SEP)if len(parts) ! 2:raise ValueError(fInvalid line format in {txt_path}:{i}. Expect img{SEP}mask, got: {line})samples.append(Sample(parts[0].strip(), parts[1].strip(), txt_path, i))return samplesdef load_mask(path: str) - np.ndarray:# 标签统一读为单通道uint8若是多通道则取第1个通道mask Image.open(path)if len(mask.getbands()) 1:mask mask.split()[0]return np.array(mask, dtypenp.uint8)def make_output_mask_path(fsd_mask_path: str, suffix: str) - str:# 例a_gt.png - a_gt_luyan_mirrorfold.pngbase, ext os.path.splitext(fsd_mask_path)return f{base}{suffix}{ext}def write_txt(path: str, lines: List[Tuple[str, str]]):# 按项目规范回写txt图片路径 三空格 标签路径parent os.path.dirname(path)if parent:os.makedirs(parent, exist_okTrue)with open(path, w, encodingutf-8) as f:for img, mask in lines:f.write(f{img}{SEP}{mask}\n)def build_rm_index(rm_txts: List[str], match_mode: str, case_insensitive: bool) - Dict[str, List[Sample]]:# key - [rm样本列表]保留一对多后续按策略处理默认取第一个idx: Dict[str, List[Sample]] {}for txt in rm_txts:for s in parse_txt(txt):key _make_key(s.image_path, match_mode, case_insensitive)idx.setdefault(key, []).append(s)return idxdef merge_one_fsd_txt(fsd_txt: str,rm_index: Dict[str, List[Sample]],args: argparse.Namespace,):处理一个FSD txt并输出两份可训练txt。- 全量txt所有FSD样本都保留命中则指向融合后mask未命中保持原mask- 命中子集txt仅保留成功命中RM并生成融合mask的样本fsd_samples parse_txt(fsd_txt)all_lines: List[Tuple[str, str]] []hit_only_lines: List[Tuple[str, str]] []# 你要求的核心统计 1个排障统计多命中total len(fsd_samples)hit_rm 0wrote_luyan 0size_mismatch 0multi_match 0merged_mask_written 0for fsd in fsd_samples:key _make_key(fsd.image_path, args.match_mode, args.case_insensitive)cands rm_index.get(key, [])# 默认不融合保留原始mask路径out_mask_path fsd.mask_pathif cands:hit_rm 1if len(cands) 1:multi_match 1rm cands[0] # 多命中策略取第一个按rm txt读取顺序try:fsd_mask load_mask(fsd.mask_path)rm_mask load_mask(rm.mask_path)except Exception as e:# 单样本读取失败时回退原始样本不中断整批处理print(f[WARN] read mask failed: fsd{fsd.mask_path}, rm{rm.mask_path}, err{e})all_lines.append((fsd.image_path, out_mask_path))continueif fsd_mask.shape ! rm_mask.shape:# 按计划尺寸不一致直接跳过不做resize避免引入噪声size_mismatch 1all_lines.append((fsd.image_path, out_mask_path))continue# 融合规则RM路沿像素值 - FSD路沿类别idcurb_area rm_mask args.rm_curb_valuemerged fsd_mask.copy()if np.any(curb_area):merged[curb_area] args.fsd_curb_classwrote_luyan 1# 输出融合mask原目录下文件名追加后缀out_mask_path make_output_mask_path(fsd.mask_path, args.mask_suffix)out_dir os.path.dirname(out_mask_path)if out_dir:os.makedirs(out_dir, exist_okTrue)# overwriteFalse 且目标已存在时复用已有融合maskif args.overwrite or (not os.path.exists(out_mask_path)):Image.fromarray(merged).save(out_mask_path)merged_mask_written 1# 命中子集只收录命中RM并生成融合mask的样本hit_only_lines.append((fsd.image_path, out_mask_path))all_lines.append((fsd.image_path, out_mask_path))base os.path.splitext(os.path.basename(fsd_txt))[0]out_all_txt os.path.join(args.out_dir, f{base}{args.txt_suffix}.txt)out_hit_txt os.path.join(args.out_dir, f{base}{args.txt_suffix}_hit_rm_only.txt)write_txt(out_all_txt, all_lines)write_txt(out_hit_txt, hit_only_lines)summary {fsd_txt: fsd_txt,output_all_txt: out_all_txt,output_hit_txt: out_hit_txt,total_fsd_samples: total,hit_rm_samples: hit_rm,wrote_luyan_pixel_samples: wrote_luyan,size_mismatch_samples: size_mismatch,multi_match_samples: multi_match,merged_mask_written: merged_mask_written,hit_only_dataset_size: len(hit_only_lines),}return summarydef main():命令行入口。示例python tools/build_fsd_luyan_from_rm.py ^--fsd-txt fsd_train.txt fsd_val.txt ^--rm-txt rm_train.txt rm_val.txt ^--match-mode fullpath ^--rm-curb-value 233 ^--fsd-curb-class 2 ^--out-dir merged_txt ^--overwriteparser argparse.ArgumentParser(description离线融合RM路沿标签到FSD标签输出训练可用txt全量 命中RM子集)parser.add_argument(--fsd-txt, nargs, requiredTrue, helpFSD txt列表如train/val)parser.add_argument(--rm-txt, nargs, requiredTrue, helpRM txt列表如train/val)parser.add_argument(--match-mode,choices[fullpath, basename],defaultfullpath,helpFSD与RM图片匹配方式完整路径或文件名,)parser.add_argument(--case-insensitive,actionstore_true,help路径匹配大小写不敏感默认false,)parser.add_argument(--rm-curb-value,typeint,default233,helpRM中路沿像素值默认233,)parser.add_argument(--fsd-curb-class,typeint,default2,helpFSD中路沿类别id默认2,)parser.add_argument(--mask-suffix,default_luyan_mirrorfold,help融合后mask后缀保留原扩展名,)parser.add_argument(--txt-suffix,default_luyan_mirrorfold,help输出txt文件名后缀,)parser.add_argument(--out-dir,requiredTrue,help输出txt目录会生成全量txt和hit_rm_only txt,)parser.add_argument(--overwrite,actionstore_true,help若目标融合mask已存在则覆盖,)args parser.parse_args()# 1) 构建RM索引rm_index build_rm_index(args.rm_txt, args.match_mode, args.case_insensitive)# 2) 逐个处理FSD txt并输出统计all_summary []for fsd_txt in args.fsd_txt:summary merge_one_fsd_txt(fsd_txt, rm_index, args)all_summary.append(summary)print(json.dumps(summary, ensure_asciiFalse, indent2))# 3) 汇总统计跨多个FSD txttotal_summary {num_fsd_txt: len(all_summary),total_fsd_samples: sum(x[total_fsd_samples] for x in all_summary),hit_rm_samples: sum(x[hit_rm_samples] for x in all_summary),wrote_luyan_pixel_samples: sum(x[wrote_luyan_pixel_samples] for x in all_summary),size_mismatch_samples: sum(x[size_mismatch_samples] for x in all_summary),multi_match_samples: sum(x[multi_match_samples] for x in all_summary),merged_mask_written: sum(x[merged_mask_written] for x in all_summary),hit_only_dataset_size: sum(x[hit_only_dataset_size] for x in all_summary),}print( TOTAL SUMMARY )print(json.dumps(total_summary, ensure_asciiFalse, indent2))if __name__ __main__:main()你这个需求很实用我已经做了两件事详细解释脚本 build_fsd_luyan_from_rm.py 的设计与流程已在脚本里补充了更详细的中文注释函数级 关键步骤级方便你后续维护脚本整体在做什么它是一个“离线标签融合器”输入FSD txttrain/val RM txttrain/val匹配FSD图片在RM里找同图支持 fullpath 和 basename融合如果命中且尺寸一致把 RM 中 233 的像素写到 FSD 标签类别 2输出新 mask原文件名后加 _luyan_mirrorfold全量训练txt含命中和未命中命中子集txt只含命中RM的样本统计你要求的四项多命中统计关键实现点你可以重点看这些函数1) 文本解析parse_txt()按三空格分隔每一行解析成 Sample(image_path, mask_path, source_txt, line_no)格式错误会抛异常避免悄悄吃错行。2) 路径匹配键_normalize_path()统一斜杠可选大小写忽略_make_key()根据模式生成键fullpath完整路径键basename文件名键3) 构建 RM 索引build_rm_index()构成 key - [Sample, Sample, ...]保留一对多后续按策略处理现在是取第一个。4) 融合核心merge_one_fsd_txt()对每个 FSD 样本找 RM 候选读取 mask尺寸不一致则跳过并计数curb_area (rm_mask 233)merged[curb_area] 2保存为 xxx_gt_luyan_mirrorfold.png记录到全量txt和命中子集txt5) 主流程main()解析参数 → 构建 RM 索引 → 处理每个 FSD txt → 打印每个txt统计和总统计。你要求的统计项是否都有都有字段名如下total_fsd_sampleshit_rm_sampleswrote_luyan_pixel_samplessize_mismatch_samples另外还有multi_match_samplesmerged_mask_writtenhit_only_dataset_size我补的中文注释内容已在脚本中补了这些层级的注释Sample 数据结构说明路径标准化/匹配模式说明txt格式约束说明融合规则与失败回退策略说明尺寸不一致为什么跳过输出命名与 overwrite 行为说明主流程分步骤注释索引、融合、汇总如果你愿意我还可以再加一版“日志更友好”的输出比如每个 FSD txt 单独打印“命中率、写入率、尺寸不一致率”百分比便于你直接放进汇报材料。