criado script para separação de fotos

This commit is contained in:
Diego Freitas 2026-05-13 19:21:37 -03:00
parent b33b4a2ae5
commit 8c4d10f329
1 changed files with 816 additions and 0 deletions

View File

@ -0,0 +1,816 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
sample_bundle_sorter.py
-------------------------------------------------
Classificador manual para o formato de dataset do módulo multiespectral.
Cada amostra é formada por:
<base>.png
<base>.json
<base>_cam0.bin
<base>_cam1.bin
<base>_cam2.bin # opcional
Fluxo:
- Entrada: uma ou mais pastas-raiz contendo amostras em subpastas.
- O script mostra o preview (.png).
- Você usa teclas 1..9/0 para enviar o conjunto da amostra para uma label.
Estrutura de saída:
out_root/
<label>/
bins/
previews/
metas/
masks/
Regras:
- PNG vai para previews/
- JSON vai para metas/
- BINs vão para bins/
- masks/ é criada vazia
"""
import argparse
import csv
import json
import shutil
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import List, Optional
import tkinter as tk
from tkinter import messagebox, filedialog
try:
from PIL import Image, ImageTk
except Exception:
print("ERROR: Pillow (PIL) is required. Install with: pip install pillow", file=sys.stderr)
raise
PREVIEW_EXT = ".png"
META_EXT = ".json"
@dataclass
class SampleBundle:
sample_dir: Path
preview_path: Path
meta_path: Path
bin_paths: List[Path]
sample_id: str
def find_sample_bundles_in_folder(folder: Path) -> List[SampleBundle]:
"""
Procura amostras em formatos antigos e atuais.
Suporta formato antigo Pi:
<base>.png
<base>.json
<base>_cam0.bin
<base>_cam1.bin
<base>_cam2.bin
Suporta formato OAK-FFC-3:
preview.png
meta.json
CAM_A.bin / CAM_B.bin / CAM_C.bin
Também suporta:
<base>_CAM_A.bin
<base>_CAM_B.bin
<base>_CAM_C.bin
<base>_rgb.bin
<base>_nir.bin
<base>_re.bin
"""
import re
if not folder.is_dir():
return []
bundles: List[SampleBundle] = []
candidate_dirs = [folder]
candidate_dirs.extend([p for p in folder.rglob("*") if p.is_dir()])
# Padrões aceitos
old_cam_re = re.compile(r"^(?P<base>.+)_cam(?P<cam>\d+)\.bin$", re.IGNORECASE)
oak_cam_re = re.compile(r"^(?:(?P<base>.+)_)?(?P<cam>CAM_[ABC])\.bin$", re.IGNORECASE)
role_re = re.compile(r"^(?P<base>.+)_(?P<role>rgb|nir|re)\.bin$", re.IGNORECASE)
role_order = {
"CAM_A": 0,
"CAM_B": 1,
"CAM_C": 2,
"RGB": 0,
"NIR": 1,
"RE": 2,
}
generic_preview_names = {"preview.png", "preview.jpg", "preview.jpeg"}
generic_meta_names = {"meta.json", "metadata.json"}
for d in candidate_dirs:
if not d.is_dir():
continue
files = [p for p in d.iterdir() if p.is_file()]
if not files:
continue
pngs = {}
jsons = {}
bins_by_base = {}
generic_preview = None
generic_meta = None
generic_bins = []
for p in files:
name_lower = p.name.lower()
suffix = p.suffix.lower()
# -----------------------------
# Preview
# -----------------------------
if suffix in (".png", ".jpg", ".jpeg"):
if name_lower in generic_preview_names:
generic_preview = p
else:
pngs[p.stem] = p
continue
# -----------------------------
# Meta
# -----------------------------
if suffix == ".json":
if name_lower in generic_meta_names:
generic_meta = p
else:
jsons[p.stem] = p
continue
# -----------------------------
# BIN
# -----------------------------
if suffix != ".bin":
continue
# Antigo: <base>_cam0.bin
m = old_cam_re.match(p.name)
if m:
base = m.group("base")
cam_idx = int(m.group("cam"))
bins_by_base.setdefault(base, []).append((cam_idx, p))
continue
# Novo: <base>_CAM_A.bin ou CAM_A.bin
m = oak_cam_re.match(p.name)
if m:
base = m.group("base") or d.name
cam = m.group("cam").upper()
order = role_order.get(cam, 99)
bins_by_base.setdefault(base, []).append((order, p))
generic_bins.append((order, p))
continue
# Alternativo: <base>_rgb.bin / _nir.bin / _re.bin
m = role_re.match(p.name)
if m:
base = m.group("base")
role = m.group("role").upper()
order = role_order.get(role, 99)
bins_by_base.setdefault(base, []).append((order, p))
continue
# -------------------------------------------------
# Caso 1: formato OAK em pasta de amostra:
# preview.png + meta.json + CAM_A/B/C.bin
# -------------------------------------------------
if generic_preview and generic_meta and generic_bins:
sorted_bins = [p for _, p in sorted(generic_bins, key=lambda x: x[0])]
sample_id = d.name
bundles.append(
SampleBundle(
sample_dir=d,
preview_path=generic_preview,
meta_path=generic_meta,
bin_paths=sorted_bins,
sample_id=sample_id,
)
)
continue
# -------------------------------------------------
# Caso 2: formato com base comum:
# sample_x.png/json + sample_x_CAM_A.bin
# ou sample_x.png/json + sample_x_cam0.bin
# -------------------------------------------------
valid_bases = sorted(set(pngs.keys()) & set(jsons.keys()) & set(bins_by_base.keys()))
for base in valid_bases:
sorted_bins = [p for _, p in sorted(bins_by_base[base], key=lambda x: x[0])]
if not sorted_bins:
continue
bundles.append(
SampleBundle(
sample_dir=d,
preview_path=pngs[base],
meta_path=jsons[base],
bin_paths=sorted_bins,
sample_id=base,
)
)
# -------------------------------------------------
# Caso 3: meta/preview genéricos, mas bins com base:
# preview.png + meta.json + sample_x_CAM_A.bin...
# -------------------------------------------------
if generic_preview and generic_meta and bins_by_base and not generic_bins:
for base, bin_items in sorted(bins_by_base.items()):
sorted_bins = [p for _, p in sorted(bin_items, key=lambda x: x[0])]
if not sorted_bins:
continue
bundles.append(
SampleBundle(
sample_dir=d,
preview_path=generic_preview,
meta_path=generic_meta,
bin_paths=sorted_bins,
sample_id=base,
)
)
bundles.sort(key=lambda x: (str(x.sample_dir), x.sample_id))
return bundles
def collect_all_bundles(folders: List[Path]) -> List[SampleBundle]:
all_bundles: List[SampleBundle] = []
seen_dirs = set()
for folder in folders:
for bundle in find_sample_bundles_in_folder(folder):
key = str((bundle.sample_dir / bundle.sample_id).resolve())
if key in seen_dirs:
continue
seen_dirs.add(key)
all_bundles.append(bundle)
all_bundles.sort(key=lambda x: (str(x.sample_dir), x.sample_id))
return all_bundles
class ActionLogger:
"""
Loga ações em:
- CSV: sorting_log.csv
- JSONL: sorting_log.jsonl
O resume usa sample_dir como chave.
"""
def __init__(self, out_root: Path):
self.out_root = out_root
self.log_csv = out_root / "sorting_log.csv"
self.log_json = out_root / "sorting_log.jsonl"
self.seen = set()
if self.log_csv.exists():
try:
with self.log_csv.open("r", newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
key = row.get("meta_path", "")
if key:
self.seen.add(key)
except Exception:
pass
def already_logged(self, sample_key: Path) -> bool:
return str(sample_key.resolve()) in self.seen
def log(
self,
action: str,
sample_dir: Path,
preview_path: Path,
meta_path: Path,
bin_paths: List[Path],
label: Optional[str] = None,
dest_sample_dir: Optional[Path] = None,
):
sample_dir_str = str(sample_dir.resolve())
meta_key_str = str(meta_path.resolve())
bin_paths_str = [str(p.resolve()) for p in bin_paths]
new_file = not self.log_csv.exists()
with self.log_csv.open("a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
if new_file:
writer.writerow([
"timestamp",
"action",
"label",
"sample_dir",
"preview_path",
"meta_path",
"bin_paths_json",
"dest_sample_dir",
])
writer.writerow([
datetime.now().isoformat(timespec="seconds"),
action,
label or "",
sample_dir_str,
str(preview_path.resolve()),
str(meta_path.resolve()),
json.dumps(bin_paths_str, ensure_ascii=False),
str(dest_sample_dir.resolve()) if dest_sample_dir else "",
])
with self.log_json.open("a", encoding="utf-8") as f:
rec = {
"ts": datetime.now().isoformat(timespec="seconds"),
"action": action,
"label": label,
"sample_dir": sample_dir_str,
"preview_path": str(preview_path.resolve()),
"meta_path": str(meta_path.resolve()),
"bin_paths": bin_paths_str,
"dest_sample_dir": str(dest_sample_dir.resolve()) if dest_sample_dir else None,
}
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
self.seen.add(str(meta_path.resolve()))
class SampleBundleSorterApp:
def __init__(
self,
bundles: List[SampleBundle],
labels: List[str],
out_root: Path,
move: bool,
resume: bool,
display_height: int = 512,
):
self.all_bundles = bundles
self.labels = labels
self.out_root = out_root
self.move = move
self.display_height = display_height
self.logger = ActionLogger(out_root)
if resume:
self.all_bundles = [
b for b in self.all_bundles if not self.logger.already_logged(b.meta_path)
]
self.idx = 0
self.history = []
for label in self.labels:
label_root = out_root / label
(label_root / "bins").mkdir(parents=True, exist_ok=True)
(label_root / "previews").mkdir(parents=True, exist_ok=True)
(label_root / "metas").mkdir(parents=True, exist_ok=True)
(label_root / "masks").mkdir(parents=True, exist_ok=True)
self.root = tk.Tk()
self.root.title("Agrobotics Sample Bundle Sorter")
self.root.geometry("1100x750")
self.root.bind("<Key>", self.on_key)
self.top_frame = tk.Frame(self.root)
self.top_frame.pack(side=tk.TOP, fill=tk.X)
self.info_label = tk.Label(self.top_frame, text="", font=("Segoe UI", 11))
self.info_label.pack(side=tk.LEFT, padx=10, pady=6)
self.legend_label = tk.Label(self.top_frame, text=self.build_legend_text(), font=("Segoe UI", 10))
self.legend_label.pack(side=tk.RIGHT, padx=10, pady=6)
self.img_frame = tk.Frame(self.root)
self.img_frame.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
self.preview_label = tk.Label(self.img_frame)
self.preview_label.pack(side=tk.LEFT, expand=True, padx=6, pady=6)
self.status_var = tk.StringVar(value="Pronto.")
self.status_label = tk.Label(self.root, textvariable=self.status_var, font=("Segoe UI", 10), anchor="w")
self.status_label.pack(side=tk.BOTTOM, fill=tk.X, padx=8, pady=4)
self.footer = tk.Label(
self.root,
text="1..9/0 = labels | espaço/n/→ = próxima (skip) | p/← = anterior | b = undo | q/Esc = sair",
font=("Segoe UI", 10),
)
self.footer.pack(side=tk.BOTTOM, fill=tk.X, pady=2)
self.render()
def unique_file(self, p: Path) -> Path:
if not p.exists():
return p
stem = p.stem
suffix = p.suffix
k = 1
while True:
cand = p.with_name(f"{stem}__{k}{suffix}")
if not cand.exists():
return cand
k += 1
def build_legend_text(self) -> str:
parts = []
for i, label in enumerate(self.labels, start=1):
key = i if i <= 9 else 0
parts.append(f"[{key}] {label}")
return " | ".join(parts)
def pil_load_preview(self, preview_path: Path) -> ImageTk.PhotoImage:
img = Image.open(preview_path).convert("RGB")
h_target = self.display_height
w, h = img.size
new_w = int(w * (h_target / h))
img = img.resize((new_w, h_target), Image.BILINEAR)
return ImageTk.PhotoImage(img)
def render(self):
if not self.all_bundles:
messagebox.showinfo("Fim", "Não há amostras para exibir.")
self.root.destroy()
return
self.idx = max(0, min(self.idx, len(self.all_bundles) - 1))
bundle = self.all_bundles[self.idx]
try:
tk_img = self.pil_load_preview(bundle.preview_path)
self.preview_tk = tk_img
self.preview_label.configure(image=self.preview_tk)
except Exception as e:
self.info_label.configure(text=f"Erro ao abrir preview: {e}")
return
bins_text = ", ".join(p.name for p in bundle.bin_paths)
self.info_label.configure(
text=(
f"{self.idx+1}/{len(self.all_bundles)} | "
f"sample='{bundle.sample_id}' | "
f"bins={len(bundle.bin_paths)} [{bins_text}]"
)
)
def send_to_label(self, label_index: int):
if label_index < 0 or label_index >= len(self.labels):
return
label = self.labels[label_index]
bundle = self.all_bundles[self.idx]
label_root = self.out_root / label
dst_preview = self.unique_file(label_root / "previews" / bundle.preview_path.name)
dst_meta = self.unique_file(label_root / "metas" / bundle.meta_path.name)
dst_bins = [self.unique_file(label_root / "bins" / p.name) for p in bundle.bin_paths]
try:
if self.move:
shutil.move(str(bundle.preview_path), str(dst_preview))
shutil.move(str(bundle.meta_path), str(dst_meta))
for src_bin, dst_bin in zip(bundle.bin_paths, dst_bins):
shutil.move(str(src_bin), str(dst_bin))
else:
shutil.copy2(bundle.preview_path, dst_preview)
shutil.copy2(bundle.meta_path, dst_meta)
for src_bin, dst_bin in zip(bundle.bin_paths, dst_bins):
shutil.copy2(src_bin, dst_bin)
self.logger.log(
action="assign",
sample_dir=bundle.sample_dir,
preview_path=bundle.preview_path,
meta_path=bundle.meta_path,
bin_paths=bundle.bin_paths,
label=label,
dest_sample_dir=label_root,
)
self.history.append({
"action": "assign",
"label": label,
"sample_src": bundle.sample_dir,
"preview_src": bundle.preview_path,
"meta_src": bundle.meta_path,
"bin_srcs": list(bundle.bin_paths),
"preview_dst": dst_preview,
"meta_dst": dst_meta,
"bin_dsts": dst_bins,
"moved": self.move,
"index": self.idx,
})
self.status_var.set(
f"{'Movido' if self.move else 'Copiado'}'{label}': {bundle.sample_id}"
)
self.idx += 1
if self.idx >= len(self.all_bundles):
messagebox.showinfo("Concluído", "Você chegou ao final da fila!")
self.root.destroy()
return
self.render()
except Exception as e:
messagebox.showerror("Erro", f"Falha ao copiar/mover amostra: {e}")
self.status_var.set(f"ERRO: {e}")
def undo(self):
if not self.history:
return
last = self.history.pop()
if last["action"] != "assign":
return
try:
preview_src = Path(last["preview_src"])
meta_src = Path(last["meta_src"])
bin_srcs = [Path(p) for p in last["bin_srcs"]]
preview_dst = Path(last["preview_dst"])
meta_dst = Path(last["meta_dst"])
bin_dsts = [Path(p) for p in last["bin_dsts"]]
if last["moved"]:
shutil.move(str(preview_dst), str(preview_src))
shutil.move(str(meta_dst), str(meta_src))
for dst_bin, src_bin in zip(bin_dsts, bin_srcs):
shutil.move(str(dst_bin), str(src_bin))
else:
if preview_dst.exists():
preview_dst.unlink()
if meta_dst.exists():
meta_dst.unlink()
for dst_bin in bin_dsts:
if dst_bin.exists():
dst_bin.unlink()
self.logger.log(
action="undo",
sample_dir=Path(last["sample_src"]),
preview_path=preview_src,
meta_path=meta_src,
bin_paths=bin_srcs,
label=last["label"],
dest_sample_dir=self.out_root / last["label"],
)
self.idx = max(0, min(last.get("index", self.idx), len(self.all_bundles) - 1))
self.status_var.set(f"Desfeito ← '{last['label']}': {preview_src.stem}")
self.render()
except Exception as e:
messagebox.showerror("Erro", f"Falha no undo: {e}")
self.status_var.set(f"ERRO: {e}")
def on_key(self, event):
ch = event.keysym.lower()
if ch in [str(i) for i in range(1, 10)] or ch == "0":
label_idx = 9 if ch == "0" else int(ch) - 1
self.send_to_label(label_idx)
return
if ch in ("space", "s", "n", "right"):
bundle = self.all_bundles[self.idx]
self.logger.log(
action="skip",
sample_dir=bundle.sample_dir,
preview_path=bundle.preview_path,
meta_path=bundle.meta_path,
bin_paths=bundle.bin_paths,
label=None,
dest_sample_dir=None,
)
self.status_var.set(f"Pulada → {bundle.sample_id}")
self.idx += 1
if self.idx >= len(self.all_bundles):
messagebox.showinfo("Concluído", "Você chegou ao final da fila!")
self.root.destroy()
return
self.render()
return
if ch in ("p", "left"):
self.idx = max(0, self.idx - 1)
self.status_var.set("Voltou uma amostra.")
self.render()
return
if ch == "b":
self.undo()
return
if ch in ("q", "escape"):
self.root.destroy()
return
def run(self):
self.root.mainloop()
class SetupWindow:
def __init__(self):
self.root = tk.Tk()
self.root.title("Configurar - Sample Bundle Sorter")
self.root.geometry("720x520")
frm_in = tk.LabelFrame(self.root, text="Pastas de entrada (raízes com subpastas de amostras)")
frm_in.pack(fill=tk.BOTH, expand=False, padx=10, pady=8)
self.inputs_listbox = tk.Listbox(frm_in, height=6)
self.inputs_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=6, pady=6)
btns_in = tk.Frame(frm_in)
btns_in.pack(side=tk.RIGHT, fill=tk.Y, padx=6, pady=6)
tk.Button(btns_in, text="Adicionar pasta...", command=self.add_input).pack(fill=tk.X, pady=2)
tk.Button(btns_in, text="Remover selecionada", command=self.remove_selected).pack(fill=tk.X, pady=2)
tk.Button(btns_in, text="Limpar lista", command=self.clear_inputs).pack(fill=tk.X, pady=2)
frm_out = tk.LabelFrame(self.root, text="Pasta de saída (raiz do dataset rotulado)")
frm_out.pack(fill=tk.X, expand=False, padx=10, pady=8)
self.out_root_var = tk.StringVar(value="")
tk.Entry(frm_out, textvariable=self.out_root_var).pack(side=tk.LEFT, fill=tk.X, expand=True, padx=6, pady=6)
tk.Button(frm_out, text="Escolher...", command=self.choose_out_root).pack(side=tk.RIGHT, padx=6, pady=6)
frm_labels = tk.LabelFrame(self.root, text="Labels (classes) separadas por vírgula")
frm_labels.pack(fill=tk.X, expand=False, padx=10, pady=8)
self.labels_var = tk.StringVar(value="chao,chao_cana,chao_erva,chao_cana_erva,cana,cana_erva,erva")
tk.Entry(frm_labels, textvariable=self.labels_var).pack(fill=tk.X, padx=6, pady=6)
frm_opts = tk.LabelFrame(self.root, text="Opções")
frm_opts.pack(fill=tk.X, expand=False, padx=10, pady=8)
self.move_var = tk.BooleanVar(value=False)
self.resume_var = tk.BooleanVar(value=True)
self.height_var = tk.IntVar(value=512)
tk.Checkbutton(frm_opts, text="Mover em vez de copiar", variable=self.move_var).pack(anchor="w", padx=6)
tk.Checkbutton(frm_opts, text="Retomar (pular itens já logados)", variable=self.resume_var).pack(anchor="w", padx=6)
frm_height = tk.Frame(frm_opts)
frm_height.pack(fill=tk.X, padx=6, pady=6)
tk.Label(frm_height, text="Altura de exibição (px):").pack(side=tk.LEFT)
tk.Entry(frm_height, textvariable=self.height_var, width=6).pack(side=tk.LEFT, padx=6)
tk.Button(self.root, text="Iniciar classificação", command=self.start).pack(pady=10)
tk.Label(
self.root,
text="Teclas: 1..9 (0=10ª), espaço/n/→=pular, p/←=anterior, b=undo, q/Esc=sair",
fg="#555",
).pack(pady=4)
self.result = None
def add_input(self):
p = filedialog.askdirectory(title="Selecione a pasta raiz contendo as amostras")
if p:
self.inputs_listbox.insert(tk.END, p)
def remove_selected(self):
sel = list(self.inputs_listbox.curselection())
for idx in reversed(sel):
self.inputs_listbox.delete(idx)
def clear_inputs(self):
self.inputs_listbox.delete(0, tk.END)
def choose_out_root(self):
p = filedialog.askdirectory(title="Selecione a pasta raiz de saída")
if p:
self.out_root_var.set(p)
def start(self):
inputs = [self.inputs_listbox.get(i) for i in range(self.inputs_listbox.size())]
out_root = self.out_root_var.get().strip()
labels_raw = self.labels_var.get().strip()
labels = [s.strip() for s in labels_raw.split(",") if s.strip()]
if not inputs:
messagebox.showwarning("Faltando entradas", "Adicione pelo menos uma pasta de entrada.")
return
if not out_root:
messagebox.showwarning("Faltando saída", "Escolha a pasta raiz de saída.")
return
if not labels:
messagebox.showwarning("Faltando labels", "Informe pelo menos uma label (classe).")
return
self.result = (
[Path(p) for p in inputs],
labels,
Path(out_root),
self.move_var.get(),
self.resume_var.get(),
self.height_var.get(),
)
self.root.destroy()
def run(self):
self.root.mainloop()
return self.result
def run_with_gui_setup():
setup = SetupWindow()
res = setup.run()
if not res:
return
input_folders, labels, out_root, move, resume, height = res
bundles = collect_all_bundles(input_folders)
if not bundles:
messagebox.showinfo(
"Sem amostras",
"Nenhuma amostra válida encontrada.\n"
"Cada amostra precisa ter preview/meta e os .bin das câmeras "
"(CAM_A/CAM_B/CAM_C, _CAM_A/_CAM_B/_CAM_C ou _cam0/_cam1/_cam2).",
)
return
app = SampleBundleSorterApp(
bundles=bundles,
labels=labels,
out_root=out_root,
move=move,
resume=resume,
display_height=height,
)
app.run()
def main():
parser = argparse.ArgumentParser(
description="Classificar manualmente amostras no novo formato do módulo multiespectral."
)
parser.add_argument("--inputs", nargs="+", help="Pastas-raiz contendo subpastas de amostras")
parser.add_argument("--labels", nargs="+", help="Labels (classes) mapeadas para teclas 1..9/0")
parser.add_argument("--out-root", help="Pasta raiz de saída")
parser.add_argument("--move", action="store_true", help="Mover em vez de copiar")
parser.add_argument("--resume", action="store_true", help="Pular itens já presentes no sorting_log.csv")
parser.add_argument("--display-height", type=int, default=512, help="Altura de exibição do preview (px)")
parser.add_argument("--no-gui-setup", action="store_true", help="Não abrir a GUI de setup")
args = parser.parse_args()
if (not args.inputs or not args.labels or not args.out_root) and not args.no_gui_setup:
run_with_gui_setup()
return
if not args.inputs or not args.labels or not args.out_root:
print("ERRO: informe --inputs, --labels e --out-root ou use a GUI.")
sys.exit(1)
input_folders = [Path(p) for p in args.inputs]
labels = args.labels
out_root = Path(args.out_root)
bundles = collect_all_bundles(input_folders)
if not bundles:
print(
"Nenhuma amostra válida encontrada. "
"Formatos aceitos: preview.png + meta.json + CAM_A/B/C.bin, "
"ou <base>.png/json + <base>_CAM_A/B/C.bin, "
"ou formato antigo <base>_cam0/cam1/cam2.bin."
)
sys.exit(1)
app = SampleBundleSorterApp(
bundles=bundles,
labels=labels,
out_root=out_root,
move=args.move,
resume=args.resume,
display_height=args.display_height,
)
app.run()
if __name__ == "__main__":
main()