从xanylabeling标注标签文件中筛选出对应的类别的工具和源码实现
2026/9/15 8:27:51 网站建设 项目流程

目录

1 选择数据,统计类别

2 类别筛选

2.1 筛选方式

2.2 具体操作实例

2.2.1 示例1

2.2.2 示例2

3 源码实现

4 源码和软件下载


该软件主要两个功能:

  • 1)加载xanylabeling标注数据,统计类别数量信息
  • 2)导出指定类别的标签并保存

1 选择数据,统计类别

  1. 加载使用xanylabeling标注数据,统计数据结果,具体操作
  • 选择标注图片目录
  • 选择存放xanylabling格式的xxx.json标签文件目录
  • 选择classes.txt类别文件(必须选择),这个主要是用来筛选类别数据的,会加载为类别筛选复选框
  • 选好之后,点击加载并统计 按钮

在统计总览,可以看到每个类别数据对应的数量:

在类别图表,可以看到两个统计图:

  • 每个类别占百分比的饼状图
  • 每个类别数量的柱状图

2 类别筛选

2.1 筛选方式

  1. 只有勾选的类别才有可能保存下来,没有勾选的类别标签一定不会被保存下来。
  2. 标签文件匹配方式:
  • 必须同时包含所有勾选类别:必须勾选的类别同时存在标签文件中,才会把勾选的类别保存到新的标签目录下,没有后勾选类别不会保存。
  • 包含任意一个勾选类别:只要标签文件含有勾选类别的一个或多个,这些类别都会保存到新的标签目录下,没有勾选类别不会保存。

比如:

  • 0001.json文件中,有A、B、C、D、E种类别标签
  • 0002.json文件中,有A、B、C、D、E,F种类别标签
  • 0003.json文件中,有A、B、C、D、F种类别标签

勾选E、F,匹配方式选择”包含任意一个勾选类别“,此时就是

  • 此时会保存到save_anylabeling/labels/0001.json,标签文件中此时只有E类别(A、B、C没有勾选都删除了)
  • 此时会保存到save_anylabeling/labels/0002.json,标签文件中有E和F两个类别(A、B、C没有勾选都删除了)
  • 此时会保存到save_anylabeling/labels/0003.json,标签文件中此时只有F类别(A、B、C没有勾选都删除了)

勾选E、F,匹配方式选择”必须同时包含所有勾选类别“,此时就是

  • 因为原始标签文件只有E类别,没有F类别,因此不会保存标签文件
  • 因为原始标签文件满足同时有E和F标签,因此会保存到save_anylabeling/labels/0002.json,标签文件中有E和F两个类别(A、B、C没有勾选都删除了)
  • 因为原始标签文件只有F类别,没有E类别,因此不会保存标签文件

2.2 具体操作实例

2.2.1 示例1

  1. 如下:
  • 勾选了smallPig和vomiting类别
  • 标签文件匹配方式为:包含任意一个类别

  1. 导出成功:

  1. 导出结果:

  1. 使用xanylabeling查看导出结果

可以看到,只要包含其中的一个类别标签就会被保存下来

2.2.2 示例2

如下:

  • 勾选了feedHave和vomiting类别
  • 标签文件匹配方式为:必须同时包含所有勾选类别

导出结果:

3 源码实现

源码:

from __future__ import annotations import json import os import shutil import subprocess import sys from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from pathlib import Path from typing import Callable, Iterable IMAGE_EXTENSIONS = { ".jpg", ".jpeg", ".png", ".bmp", ".webp", ".tif", ".tiff", } PREFERRED_SHAPE_TYPES = [ "polygon", "rectangle", "rotation", "quadrilateral", "point", "line", "circle", "linestrip", "cuboid", ] class ClassFileError(ValueError): """Raised when classes.txt is not a readable plain-text class list.""" def _read_protected_file_via_python(path: Path) -> bytes | None: """Ask a trusted system Python to read files hidden from an unsigned EXE. Some Windows DLP/transparent-encryption products expose plaintext to python.exe but encrypted TSD bytes to newly packaged executables. """ helper = ( "import pathlib,sys;" "data=pathlib.Path(sys.argv[1]).read_bytes();" "sys.stdout.buffer.write(data)" ) commands: list[list[str]] = [] py_launcher = shutil.which("py") if py_launcher: commands.append([py_launcher, "-3"]) python_executable = shutil.which("python") if python_executable: try: is_current_executable = Path(python_executable).resolve() == Path(sys.executable).resolve() except OSError: is_current_executable = False if not is_current_executable: commands.append([python_executable]) child_environment = os.environ.copy() child_environment.pop("PYTHONHOME", None) for name in list(child_environment): if name.startswith("_PYI_"): child_environment.pop(name, None) creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) for command in commands: try: completed = subprocess.run( [*command, "-c", helper, str(path)], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=10, check=False, env=child_environment, creationflags=creation_flags, ) except (OSError, subprocess.SubprocessError): continue recovered = completed.stdout if ( completed.returncode == 0 and recovered and len(recovered) <= 10 * 1024 * 1024 and not recovered.startswith(b"%TSD-Header-###%") ): return recovered return None @dataclass(slots=True) class AnnotationRecord: json_path: Path data: dict labels: set[str] image_path: Path | None relative_path: Path @dataclass(slots=True) class DatasetSummary: records: list[AnnotationRecord] = field(default_factory=list) counts: dict[str, Counter] = field(default_factory=dict) shape_types: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) unmatched_images: int = 0 @property def total_shapes(self) -> int: return sum(sum(counter.values()) for counter in self.counts.values()) @dataclass(slots=True) class ExportSummary: exported_files: int = 0 exported_shapes: int = 0 skipped_missing_image: int = 0 errors: list[str] = field(default_factory=list) def _decode_plain_text(path: Path) -> str: raw = path.read_bytes() if raw.startswith(b"%TSD-Header-###%"): recovered = _read_protected_file_via_python(path) if recovered is not None: raw = recovered else: raise ClassFileError( "当前 EXE 只能读取到 classes.txt 的 TSD 加密内容,且未能通过本机 " "Python 获取明文;程序将从 JSON 标签中自动提取类别。" ) if raw.startswith((b"\xff\xfe", b"\xfe\xff")): candidates = ("utf-16",) elif raw.startswith(b"\xef\xbb\xbf"): candidates = ("utf-8-sig",) else: candidates = ("utf-8", "gb18030", "utf-16") for encoding in candidates: try: text = raw.decode(encoding) except UnicodeDecodeError: continue if "\x00" in text: continue printable = sum(character.isprintable() or character in "\r\n\t" for character in text) if text and printable / len(text) < 0.9: continue return text raise ClassFileError( "无法识别 classes.txt 的文本编码;请另存为 UTF-8、GB18030 或 UTF-16 文本。" ) def read_classes(path: str | Path) -> list[str]: """Read UTF-8/GB18030/UTF-16 classes, preserving order and uniqueness.""" result: list[str] = [] seen: set[str] = set() for raw_line in _decode_plain_text(Path(path)).splitlines(): label = raw_line.strip() if label and label not in seen: seen.add(label) result.append(label) return result def _image_candidates( image_dir: Path, wanted_names: set[str], wanted_stems: set[str], progress: Callable[[int, int], None] | None = None, ) -> tuple[dict[str, list[Path]], dict[str, list[Path]]]: """Index only names needed by unresolved annotations using fast os.walk/scandir.""" by_name: dict[str, list[Path]] = defaultdict(list) by_stem: dict[str, list[Path]] = defaultdict(list) scanned = 0 for root, directory_names, file_names in os.walk(image_dir): # Avoid recursively indexing this tool's own default exports. directory_names[:] = [name for name in directory_names if name != "save_xanylabeling"] root_path = Path(root) for filename in file_names: scanned += 1 suffix = Path(filename).suffix.lower() if suffix not in IMAGE_EXTENSIONS: continue folded_name = filename.casefold() folded_stem = Path(filename).stem.casefold() if folded_name in wanted_names or folded_stem in wanted_stems: path = root_path / filename if folded_name in wanted_names: by_name[folded_name].append(path) if folded_stem in wanted_stems: by_stem[folded_stem].append(path) if progress and scanned % 2000 == 0: progress(96, 100) return by_name, by_stem def _resolve_image( data: dict, json_path: Path, image_dir: Path, by_name: dict[str, list[Path]], by_stem: dict[str, list[Path]], ) -> Path | None: raw_image_path = data.get("imagePath") if isinstance(raw_image_path, str) and raw_image_path.strip(): normalized = raw_image_path.replace("\\", "/") direct = image_dir / normalized if direct.is_file(): return direct candidates = by_name.get(Path(normalized).name.casefold(), []) if len(candidates) == 1: return candidates[0] candidates = by_stem.get(json_path.stem.casefold(), []) if len(candidates) == 1: return candidates[0] return None def _quick_resolve_image( data: dict, json_path: Path, image_root: Path, label_root: Path, ) -> Path | None: """Resolve the usual layouts without building a global image index.""" relative_parent = json_path.relative_to(label_root).parent raw_image_path = data.get("imagePath") candidates: list[Path] = [] if isinstance(raw_image_path, str) and raw_image_path.strip(): normalized = raw_image_path.replace("\\", "/") image_name = Path(normalized).name candidates.extend( [ image_root / image_name, image_root / relative_parent / image_name, image_root / normalized, ] ) else: for extension in IMAGE_EXTENSIONS: candidates.append(image_root / f"{json_path.stem}{extension}") candidates.append(image_root / relative_parent / f"{json_path.stem}{extension}") seen: set[str] = set() for candidate in candidates: key = os.path.normcase(os.path.normpath(str(candidate))) if key in seen: continue seen.add(key) if candidate.is_file(): return candidate return None def _read_annotation(json_path: Path, label_root: Path): with json_path.open("r", encoding="utf-8-sig") as stream: data = json.load(stream) if not isinstance(data, dict): raise ValueError("JSON 根节点不是对象") shapes = data.get("shapes", []) if not isinstance(shapes, list): raise ValueError("shapes 字段不是列表") labels: set[str] = set() counts: dict[str, Counter] = defaultdict(Counter) shape_types: set[str] = set() for shape in shapes: if not isinstance(shape, dict): continue label = shape.get("label") if not isinstance(label, str) or not label: continue shape_type = str(shape.get("shape_type") or "unknown") labels.add(label) shape_types.add(shape_type) counts[label][shape_type] += 1 record = AnnotationRecord( json_path=json_path, data=data, labels=labels, image_path=None, relative_path=json_path.relative_to(label_root), ) return record, counts, shape_types def discover_classes_from_json( label_dir: str | Path, progress: Callable[[int, int], None] | None = None, ) -> dict[str, list[str]]: """Quickly recover class names when classes.txt is protected or unreadable.""" label_root = Path(label_dir).resolve() json_files = [ Path(root) / filename for root, _, file_names in os.walk(label_root) for filename in file_names if filename.lower().endswith(".json") ] labels: set[str] = set() errors: list[str] = [] last_reported = -1 worker_count = min(8, max(2, os.cpu_count() or 4)) with ThreadPoolExecutor(max_workers=worker_count, thread_name_prefix="class-reader") as executor: futures = { executor.submit(_read_annotation, json_path, label_root): json_path for json_path in json_files } for completed, future in enumerate(as_completed(futures), start=1): try: record, _, _ = future.result() labels.update(record.labels) except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: errors.append(f"{futures[future]}: {exc}") current = round(completed / max(1, len(json_files)) * 100) if progress and current != last_reported: progress(current, 100) last_reported = current return {"classes": sorted(labels, key=str.casefold), "errors": errors} def load_dataset( image_dir: str | Path, label_dir: str | Path, known_classes: Iterable[str], progress: Callable[[int, int], None] | None = None, ) -> DatasetSummary: image_root = Path(image_dir).resolve() label_root = Path(label_dir).resolve() known = list(known_classes) summary = DatasetSummary(counts={name: Counter() for name in known}) json_files = [ Path(root) / filename for root, _, file_names in os.walk(label_root) for filename in file_names if filename.lower().endswith(".json") ] if progress: progress(3, 100) found_shape_types: set[str] = set() worker_count = min(8, max(2, os.cpu_count() or 4)) last_reported_progress = 3 with ThreadPoolExecutor(max_workers=worker_count, thread_name_prefix="json-reader") as executor: futures = { executor.submit(_read_annotation, json_path, label_root): json_path for json_path in json_files } for completed, future in enumerate(as_completed(futures), start=1): json_path = futures[future] try: record, record_counts, record_shape_types = future.result() summary.records.append(record) found_shape_types.update(record_shape_types) for label, counter in record_counts.items(): summary.counts.setdefault(label, Counter()).update(counter) except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: summary.errors.append(f"{json_path}: {exc}") current_progress = 5 + round(completed / max(1, len(json_files)) * 75) if progress and current_progress != last_reported_progress: progress(current_progress, 100) last_reported_progress = current_progress summary.records.sort(key=lambda record: str(record.relative_path).casefold()) unresolved: list[AnnotationRecord] = [] for index, record in enumerate(summary.records, start=1): record.image_path = _quick_resolve_image( record.data, record.json_path, image_root, label_root ) if record.image_path is None: unresolved.append(record) current_progress = 80 + round(index / max(1, len(summary.records)) * 15) if progress and current_progress != last_reported_progress: progress(current_progress, 100) last_reported_progress = current_progress if unresolved: wanted_names: set[str] = set() wanted_stems: set[str] = set() for record in unresolved: raw_image_path = record.data.get("imagePath") if isinstance(raw_image_path, str) and raw_image_path.strip(): wanted_names.add(Path(raw_image_path.replace("\\", "/")).name.casefold()) else: wanted_stems.add(record.json_path.stem.casefold()) by_name, by_stem = _image_candidates( image_root, wanted_names, wanted_stems, progress ) for record in unresolved: record.image_path = _resolve_image( record.data, record.json_path, image_root, by_name, by_stem ) if record.image_path is None: summary.unmatched_images += 1 extras = sorted(found_shape_types.difference(PREFERRED_SHAPE_TYPES)) summary.shape_types = PREFERRED_SHAPE_TYPES.copy() + extras if progress: progress(100, 100) return summary def record_matches(record: AnnotationRecord, selected: set[str], mode: str) -> bool: if mode == "all": return selected.issubset(record.labels) if mode == "any": return bool(selected.intersection(record.labels)) raise ValueError(f"不支持的筛选模式: {mode}") def export_dataset( summary: DatasetSummary, output_root: str | Path, selected_classes: Iterable[str], mode: str, progress: Callable[[int, int], None] | None = None, ) -> ExportSummary: selected_list = list(dict.fromkeys(selected_classes)) selected = set(selected_list) if not selected: raise ValueError("请至少选择一个类别") output = Path(output_root).resolve() image_output = output / "image" label_output = output / "labels" image_output.mkdir(parents=True, exist_ok=True) label_output.mkdir(parents=True, exist_ok=True) (output / "classes.txt").write_text( "\n".join(selected_list) + "\n", encoding="utf-8" ) result = ExportSummary() matched = [record for record in summary.records if record_matches(record, selected, mode)] for index, record in enumerate(matched, start=1): try: if record.image_path is None or not record.image_path.is_file(): result.skipped_missing_image += 1 continue relative_parent = record.relative_path.parent destination_image = image_output / relative_parent / record.image_path.name destination_json = label_output / record.relative_path destination_image.parent.mkdir(parents=True, exist_ok=True) destination_json.parent.mkdir(parents=True, exist_ok=True) filtered_data = dict(record.data) filtered_shapes = [ shape for shape in record.data.get("shapes", []) if isinstance(shape, dict) and shape.get("label") in selected ] filtered_data["shapes"] = filtered_shapes filtered_data["imagePath"] = os.path.relpath( destination_image, destination_json.parent ).replace("\\", "/") shutil.copy2(record.image_path, destination_image) destination_json.write_text( json.dumps(filtered_data, ensure_ascii=False, indent=2), encoding="utf-8", ) result.exported_files += 1 result.exported_shapes += len(filtered_shapes) except (OSError, ValueError, TypeError) as exc: result.errors.append(f"{record.json_path}: {exc}") if progress: progress(index, len(matched)) return result

4 源码和软件下载

软件下载地址:从xanylabeling标注标签文件中筛选出对应的类别的工具

(https://mbd.pub/o/bread/YZaVlJxvbQ==)

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询