dan/ScanSlicer
dan/ScanSlicer
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions
ScanSlicer/scanslicer/autoslicer_feedback.py

727 lines
28 KiB
Python
Raw Normal View History

feat(autoslicer_feedback): scaffold module imports and constants
2026-02-20 11:47:13 +00:00
import json
import math
import os
from datetime import datetime
import cv2
import numpy as np
feat(autoslicer_feedback): implement class AutoslicerFeedbackMixin
2026-02-20 11:52:13 +00:00
class AutoslicerFeedbackMixin:
AUTOTUNE_PROFILE_ORDER = ("Fast", "Balanced", "Thorough", "Exhaustive")
DEFAULT_AUTOTUNE_PROFILE = "Balanced"
AUTOTUNE_PROFILES = {
"Fast": {
"thresholds": [160, 180, 200],
"morph_kernels": [(2, 18), (1, 14), (4, 10)],
"adaptive_blocks": [61, 101],
"adaptive_cs": [2],
"bbox_windows": [(10, 40), (8, 50), (5, 70)],
"aspect_windows": [(115, 315), (100, 450)],
"invert_modes_simple": [0],
"max_candidates_expected": 40,
"max_candidates_default": 56,
"seed_limit": 10,
"early_stop_min_expected": 16,
"early_stop_min_default": 22,
"early_stop_streak_expected": 10,
"early_stop_streak_default": 14,
"finalists_expected": 6,
"finalists_default": 8,
},
"Balanced": {
"thresholds": [160, 170, 180, 190, 200],
"morph_kernels": [(2, 25), (2, 18), (1, 20), (4, 12)],
"adaptive_blocks": [61, 81, 101, 121],
"adaptive_cs": [2],
"bbox_windows": [(10, 40), (8, 50), (5, 70), (2, 90)],
"aspect_windows": [(115, 315), (100, 450), (100, 600)],
"invert_modes_simple": [0],
"max_candidates_expected": 64,
"max_candidates_default": 80,
"seed_limit": 16,
"early_stop_min_expected": 20,
"early_stop_min_default": 28,
"early_stop_streak_expected": 14,
"early_stop_streak_default": 20,
"finalists_expected": 8,
"finalists_default": 12,
},
"Thorough": {
"thresholds": [140, 150, 160, 170, 180, 190, 200, 210, 220],
"morph_kernels": [(2, 30), (2, 22), (2, 16), (1, 18), (4, 14), (3, 12)],
"adaptive_blocks": [41, 61, 81, 101, 121, 141],
"adaptive_cs": [0, 2, 4],
"bbox_windows": [(12, 35), (10, 40), (8, 50), (5, 70), (2, 90)],
"aspect_windows": [(100, 300), (115, 315), (100, 450), (100, 600), (80, 800)],
"invert_modes_simple": [0, 2],
"max_candidates_expected": 120,
"max_candidates_default": 160,
"seed_limit": 24,
"early_stop_min_expected": 32,
"early_stop_min_default": 44,
"early_stop_streak_expected": 20,
"early_stop_streak_default": 28,
"finalists_expected": 12,
"finalists_default": 16,
},
"Exhaustive": {
"thresholds": [130, 140, 150, 160, 170, 180, 190, 200, 210, 220, 230],
"morph_kernels": [(2, 35), (2, 28), (2, 22), (2, 16), (1, 24), (1, 16), (4, 16), (3, 14)],
"adaptive_blocks": [31, 41, 61, 81, 101, 121, 141, 161],
"adaptive_cs": [-2, 0, 2, 4, 6],
"bbox_windows": [(15, 30), (12, 35), (10, 40), (8, 50), (5, 70), (2, 90)],
"aspect_windows": [(90, 280), (100, 300), (115, 315), (100, 450), (100, 600), (80, 900)],
"invert_modes_simple": [0, 2, 1],
"max_candidates_expected": 220,
"max_candidates_default": 280,
"seed_limit": 36,
"early_stop_min_expected": 56,
"early_stop_min_default": 72,
"early_stop_streak_expected": 36,
"early_stop_streak_default": 48,
"finalists_expected": 18,
"finalists_default": 24,
},
}
def autotune_profile_names(self):
return list(self.AUTOTUNE_PROFILE_ORDER)
def default_autotune_profile_name(self):
return self.DEFAULT_AUTOTUNE_PROFILE
def normalize_autotune_profile(self, profile_name):
text = str(profile_name).strip().lower() if profile_name is not None else ""
for name in self.AUTOTUNE_PROFILE_ORDER:
if text == name.lower():
return name
return self.DEFAULT_AUTOTUNE_PROFILE
def get_autotune_profile_config(self, profile_name):
normalized = self.normalize_autotune_profile(profile_name)
return normalized, dict(self.AUTOTUNE_PROFILES[normalized])
def sample_candidates_evenly(self, candidates, max_candidates):
if max_candidates <= 0 or len(candidates) <= max_candidates:
return list(candidates)
if max_candidates == 1:
return [candidates[0]]
selected = []
selected_indexes = set()
span = float(len(candidates) - 1)
for idx in range(max_candidates):
pick = int(round((idx * span) / float(max_candidates - 1)))
pick = min(len(candidates) - 1, max(0, pick))
if pick in selected_indexes:
continue
selected_indexes.add(pick)
selected.append(candidates[pick])
if len(selected) < max_candidates:
for idx, candidate in enumerate(candidates):
if idx in selected_indexes:
continue
selected.append(candidate)
if len(selected) >= max_candidates:
break
return selected
def feedback_path(self):
override = os.environ.get("SCANSLICER_FEEDBACK_PATH")
if override:
return override
if os.path.exists("/.dockerenv") and os.path.isdir("/work"):
return os.path.join("/work", ".scanslicer", "autotune-feedback.json")
config_home = os.environ.get("XDG_CONFIG_HOME")
if not config_home:
config_home = os.path.join(os.path.expanduser("~"), ".config")
return os.path.join(config_home, "scanslicer", "autotune-feedback.json")
def load_feedback_entries(self):
path = self.feedback_path()
if not os.path.isfile(path):
return []
try:
with open(path, "r", encoding="utf-8") as fh:
loaded = json.load(fh)
except (OSError, json.JSONDecodeError, TypeError, ValueError):
return []
if not isinstance(loaded, list):
return []
cleaned = []
for entry in loaded:
if not isinstance(entry, dict):
continue
params = entry.get("params")
final_boxes = entry.get("final_boxes")
features = entry.get("image_features")
final_count = entry.get("final_count")
if not isinstance(params, dict):
continue
if not isinstance(final_boxes, list):
continue
if not isinstance(features, dict):
continue
if not isinstance(final_count, int):
continue
cleaned.append(entry)
return cleaned[-self.MAX_FEEDBACK_ENTRIES:]
def save_feedback_entries(self):
path = self.feedback_path()
directory = os.path.dirname(path)
if directory:
os.makedirs(directory, exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
json.dump(self.feedback_entries[-self.MAX_FEEDBACK_ENTRIES:], fh, indent=2, sort_keys=True)
return path
def clear_feedback_entries(self):
cleared_count = len(self.feedback_entries)
self.feedback_entries = []
path = self.feedback_path()
removed_file = False
if os.path.isfile(path):
os.remove(path)
removed_file = True
return {
"cleared_count": int(cleared_count),
"path": path,
"removed_file": removed_file,
}
def get_param_range_map(self):
range_map = {}
for spec in self.params.SPECS:
if len(spec) < 4:
continue
name = spec[0]
min_value = spec[2]
max_value = spec[3]
range_map[name] = max(1.0, float(max_value - min_value))
return range_map
def compute_image_features(self, image_gray=None):
gray = image_gray if image_gray is not None else self.image_gray
if gray is None:
return None
p10, p50, p90 = np.percentile(gray, [10, 50, 90])
edges = cv2.Canny(gray, 80, 180)
edge_density = float(np.count_nonzero(edges)) / float(max(1, gray.shape[0] * gray.shape[1]))
return {
"mean": float(np.mean(gray)),
"std": float(np.std(gray)),
"p10": float(p10),
"p50": float(p50),
"p90": float(p90),
"edge_density": float(edge_density),
}
def normalize_boxes_for_storage(self, boxes):
normalized = []
for box in boxes:
arr = np.array(box, dtype=np.float64).reshape(4, 2)
normalized.append([[float(arr[i, 0]), float(arr[i, 1])] for i in range(4)])
return normalized
def boxes_signature(self, boxes):
if not boxes:
return ""
normalized = self.normalize_boxes_for_storage(boxes)
centroids = []
for box in normalized:
xs = [p[0] for p in box]
ys = [p[1] for p in box]
centroids.append((float(sum(xs) / 4.0), float(sum(ys) / 4.0), box))
centroids.sort(key=lambda item: (round(item[1], 3), round(item[0], 3)))
rounded = [
[[round(p[0], 1), round(p[1], 1)] for p in box]
for _, _, box in centroids
]
return json.dumps(rounded, sort_keys=True)
def record_feedback(self, image_path, final_boxes, detected_boxes=None):
if not image_path or not final_boxes:
return False
features = self.compute_image_features()
if features is None:
return False
final_boxes_stored = self.normalize_boxes_for_storage(final_boxes)
detected_boxes_stored = self.normalize_boxes_for_storage(detected_boxes or [])
entry = {
"timestamp": datetime.now().isoformat(timespec="seconds"),
"image_path": str(image_path),
"params": self.params.to_dict(),
"image_features": features,
"detected_boxes": detected_boxes_stored,
"final_boxes": final_boxes_stored,
"final_count": int(len(final_boxes_stored)),
"final_signature": self.boxes_signature(final_boxes_stored),
}
replaced = False
for idx in range(len(self.feedback_entries) - 1, -1, -1):
existing = self.feedback_entries[idx]
if (
existing.get("image_path") == entry["image_path"]
and existing.get("final_signature") == entry["final_signature"]
):
self.feedback_entries[idx] = entry
replaced = True
break
if not replaced:
self.feedback_entries.append(entry)
if len(self.feedback_entries) > self.MAX_FEEDBACK_ENTRIES:
self.feedback_entries = self.feedback_entries[-self.MAX_FEEDBACK_ENTRIES:]
self.save_feedback_entries()
return True
def image_similarity(self, a, b):
if not a or not b:
return 0.0
dist = 0.0
dist += abs(float(a.get("mean", 0.0)) - float(b.get("mean", 0.0))) / 64.0
dist += abs(float(a.get("std", 0.0)) - float(b.get("std", 0.0))) / 64.0
dist += abs(float(a.get("p50", 0.0)) - float(b.get("p50", 0.0))) / 64.0
dist += abs(float(a.get("edge_density", 0.0)) - float(b.get("edge_density", 0.0))) / 0.2
return float(math.exp(-dist))
def param_similarity(self, candidate_values, reference_values, range_map):
distances = []
for key in self.AUTOTUNE_PARAM_KEYS:
if key not in candidate_values or key not in reference_values:
continue
try:
a = float(candidate_values[key])
b = float(reference_values[key])
except (TypeError, ValueError):
continue
denom = range_map.get(key, 1.0)
distances.append(abs(a - b) / denom)
if not distances:
return 0.0
mean_dist = float(np.mean(distances))
return max(0.0, 1.0 - min(1.0, mean_dist))
def feedback_score_bonus(self, candidate_values, box_count, image_features):
if not self.feedback_entries or not image_features:
return 0.0
range_map = self.get_param_range_map()
scored = []
for entry in self.feedback_entries:
sim = self.image_similarity(image_features, entry.get("image_features"))
if sim <= 0.01:
continue
scored.append((sim, entry))
if not scored:
return 0.0
scored.sort(key=lambda item: item[0], reverse=True)
top_scored = scored[:20]
bonus = 0.0
for image_sim, entry in top_scored:
reference_params = entry.get("params", {})
final_count = int(entry.get("final_count", 0))
param_sim = self.param_similarity(candidate_values, reference_params, range_map)
if final_count > 0:
count_delta = abs(int(box_count) - final_count)
count_sim = max(0.0, 1.0 - (count_delta / max(1.0, float(final_count))))
else:
count_sim = 0.0
entry_bonus = image_sim * ((0.7 * param_sim) + (0.3 * count_sim))
bonus += entry_bonus
# Keep this as a moderate bias, not an override.
return bonus * 280.0
def score_detected_boxes(self, boxes, candidate_values=None, image_features=None, box_area_props=None):
if not boxes:
return -1e9
if box_area_props is None:
h, w, _ = self.image.shape
img_area = float(max(1, h * w))
areas = []
for box in boxes:
area = float(cv2.contourArea(np.array(box, dtype=np.float32)))
if area > 0:
areas.append(area / img_area)
else:
areas = [float(a) for a in box_area_props if float(a) > 0.0]
if not areas:
return -1e9
count = len(areas)
total_prop = float(np.sum(areas))
mean_prop = float(np.mean(areas))
std_prop = float(np.std(areas))
uniformity = 1.0 - min(1.0, std_prop / (mean_prop + 1e-9))
tiny_count = sum(1 for p in areas if p < 0.003)
expected_count = max(0, int(self.params.expected_count.get()))
if expected_count > 0:
plausible_count = 1 if max(1, expected_count - 2) <= count <= expected_count + 4 else 0
else:
plausible_count = 1 if 1 <= count <= 12 else 0
multi_bonus = 1 if count >= 2 else 0
score = 0.0
score += plausible_count * 900.0
score += multi_bonus * 180.0
score += min(count, 8) * 60.0
score -= max(0, count - 12) * 70.0
score += min(total_prop, 0.95) * 140.0
score -= abs(total_prop - 0.35) * 90.0
score += uniformity * 80.0
score -= tiny_count * 45.0
if expected_count > 0:
count_delta = abs(count - expected_count)
score += 350.0 if count_delta == 0 else 0.0
score -= count_delta * 220.0
candidate = candidate_values if isinstance(candidate_values, dict) else self.params.to_dict()
score += self.feedback_score_bonus(candidate, count, image_features)
return score
def build_autotune_candidates(self, base_values, profile_name=None):
_, profile = self.get_autotune_profile_config(profile_name)
base = dict(base_values)
candidates = []
def add(overrides):
c = dict(base)
c.update(overrides)
candidates.append(c)
# Always evaluate the current parameter set as a baseline.
add({})
# White-background dominant candidates (threshold + invert mode variations).
thresholds = profile["thresholds"]
morph_kernels = profile["morph_kernels"]
bbox_windows = profile["bbox_windows"]
aspect_windows = profile["aspect_windows"]
for thresh in thresholds:
for morph_op, kernel in morph_kernels:
for invert_mode in profile["invert_modes_simple"]:
add(
{
"detect_pipeline": 0,
"bw_method": 0,
"bw_thresh": thresh,
"invert_mode": invert_mode,
"morph_op": morph_op,
"dilate_kernel": kernel,
"contour_approx_pct": 0,
"clahe_clip": 0,
"gaussian": 0,
}
)
# More adaptive candidates for uneven scans.
for block in profile["adaptive_blocks"]:
for adaptive_c in profile["adaptive_cs"]:
add(
{
"detect_pipeline": 0,
"bw_method": 1,
"bw_gauss": block,
"bw_adaptive_c": adaptive_c,
"invert_mode": 0,
"morph_op": 4,
"dilate_kernel": 12,
"clahe_clip": 10,
"gaussian": 1,
}
)
# Otsu candidates.
for morph_op, kernel in morph_kernels:
add(
{
"detect_pipeline": 0,
"bw_method": 2,
"invert_mode": 0,
"morph_op": morph_op,
"dilate_kernel": kernel,
"clahe_clip": 0,
"gaussian": 0,
}
)
# Apply bbox/aspect windows to a bounded subset of seeds.
seed_limit = max(1, int(profile["seed_limit"]))
seed_candidates = list(candidates[:seed_limit]) if len(candidates) >= seed_limit else list(candidates)
for seed in seed_candidates:
for min_sz, max_sz in bbox_windows:
for min_ar, max_ar in aspect_windows:
add(
{
**seed,
"bbox_min_size": min_sz,
"bbox_max_size": max_sz,
"bbox_fill_thresh": 4,
"aspect_ratio_min": min_ar,
"aspect_ratio_max": max_ar,
"iou_merge_thresh": 3,
"border_ignore": 0,
}
)
# Deduplicate candidate dicts.
dedup = {}
for candidate in candidates:
key = tuple(sorted(candidate.items()))
dedup[key] = candidate
deduped_candidates = list(dedup.values())
# Keep bounded runtime.
expected_count = max(0, int(base.get("expected_count", 0)))
max_candidates = (
int(profile["max_candidates_expected"])
if expected_count > 0
else int(profile["max_candidates_default"])
)
return self.sample_candidates_evenly(deduped_candidates, max_candidates)
def rank_autotune_candidates(self, candidates, base_values, image_features):
if not candidates:
return []
range_map = self.get_param_range_map()
expected_count = max(0, int(base_values.get("expected_count", 0)))
scored_feedback = []
if image_features and self.feedback_entries:
for entry in self.feedback_entries:
sim = self.image_similarity(image_features, entry.get("image_features"))
if sim <= 0.01:
continue
scored_feedback.append((sim, entry))
scored_feedback.sort(key=lambda item: item[0], reverse=True)
scored_feedback = scored_feedback[:10]
ranked = []
for idx, candidate in enumerate(candidates):
score = 0.0
# Slightly prefer candidates near current values.
score += 0.25 * self.param_similarity(candidate, base_values, range_map)
# Prefer candidates that look similar to known-good feedback for
# similar-looking images.
for image_sim, entry in scored_feedback:
entry_params = entry.get("params", {})
param_sim = self.param_similarity(candidate, entry_params, range_map)
score += image_sim * param_sim
if expected_count > 0:
final_count = int(entry.get("final_count", 0))
if final_count > 0:
count_sim = max(
0.0,
1.0 - abs(expected_count - final_count) / max(1.0, float(expected_count)),
)
score += 0.2 * image_sim * count_sim
# Preserve deterministic ordering when scores tie.
score += (len(candidates) - idx) * 1e-6
ranked.append((score, candidate))
ranked.sort(key=lambda item: item[0], reverse=True)
return [candidate for _, candidate in ranked]
def autodetect_candidate_boxes(self, candidate_values, fast_mode=False, downscale_max_dim=None):
original_image = self.image
original_gray = self.image_gray
scaled = False
if (
downscale_max_dim is not None
and original_image is not None
and original_gray is not None
and downscale_max_dim > 0
):
src_h, src_w = original_gray.shape[:2]
longest = float(max(src_h, src_w))
if longest > float(downscale_max_dim):
scale = float(downscale_max_dim) / longest
dst_w = max(1, int(round(src_w * scale)))
dst_h = max(1, int(round(src_h * scale)))
self.image = cv2.resize(original_image, (dst_w, dst_h), interpolation=cv2.INTER_AREA)
self.image_gray = cv2.resize(original_gray, (dst_w, dst_h), interpolation=cv2.INTER_AREA)
scaled = True
try:
self.params.apply_values(candidate_values)
boxes, _ = self.autodetect_slices(
update_status_callback=None,
preview_filter_output=False,
fast_mode=bool(fast_mode),
)
h, w, _ = self.image.shape
img_area = float(max(1, h * w))
area_props = []
for box in boxes:
area = float(cv2.contourArea(np.array(box, dtype=np.float32)))
if area > 0:
area_props.append(area / img_area)
return boxes, area_props
finally:
if scaled:
self.image = original_image
self.image_gray = original_gray
def autotune_for_current_image(self, update_status_callback=None, profile_name=None):
if not self.image_loaded():
raise RuntimeError("No image loaded for auto-tune.")
resolved_profile, profile = self.get_autotune_profile_config(profile_name)
original_values = self.params.to_dict()
image_features = self.compute_image_features()
candidates = self.build_autotune_candidates(original_values, profile_name=resolved_profile)
candidates = self.rank_autotune_candidates(candidates, original_values, image_features)
if len(candidates) == 0:
raise RuntimeError("No auto-tune candidates available.")
expected_count = max(0, int(original_values.get("expected_count", 0)))
# Coarse pass: evaluate candidates with downscaled image and fast detection.
coarse_results = []
coarse_evaluated = 0
no_improve_streak = 0
early_stop_min_runs = (
int(profile["early_stop_min_expected"])
if expected_count > 0
else int(profile["early_stop_min_default"])
)
early_stop_streak = (
int(profile["early_stop_streak_expected"])
if expected_count > 0
else int(profile["early_stop_streak_default"])
)
best_coarse_score = -1e12
best_coarse_count = 0
if update_status_callback is not None:
update_status_callback(
f"Auto-tune profile {resolved_profile}: {len(candidates)} candidates queued."
)
for idx, candidate in enumerate(candidates, start=1):
if self.abort_flag:
break
boxes, area_props = self.autodetect_candidate_boxes(
candidate_values=candidate,
fast_mode=True,
downscale_max_dim=1400,
)
score = self.score_detected_boxes(
boxes,
candidate_values=candidate,
image_features=image_features,
box_area_props=area_props,
)
coarse_results.append((score, len(boxes), dict(candidate)))
coarse_evaluated = idx
if update_status_callback is not None and (idx == 1 or idx % 10 == 0 or idx == len(candidates)):
update_status_callback(
f"Auto-tune coarse {idx}/{len(candidates)}: {len(boxes)} box(es), score={score:.1f}"
)
if score > best_coarse_score:
best_coarse_score = score
best_coarse_count = len(boxes)
no_improve_streak = 0
else:
no_improve_streak += 1
# Early-stop once search clearly converges.
if idx >= early_stop_min_runs and no_improve_streak >= early_stop_streak:
# If expected count is set, only stop when the best candidate is
# reasonably close to that target.
if expected_count <= 0 or abs(best_coarse_count - expected_count) <= 1:
break
if len(coarse_results) == 0:
self.params.apply_values(original_values)
return {
"profile": resolved_profile,
"count": 0,
"score": -1e9,
"candidates": coarse_evaluated,
"candidates_total": len(candidates),
"refined": 0,
}
coarse_results.sort(key=lambda item: item[0], reverse=True)
finalists_count = (
int(profile["finalists_expected"])
if expected_count > 0
else int(profile["finalists_default"])
)
finalists = coarse_results[:finalists_count]
# Fine pass: re-run only top coarse candidates at full quality.
best_score = -1e12
best_count = 0
best_values = None
refined_evaluated = 0
for idx, (_, _, candidate) in enumerate(finalists, start=1):
if self.abort_flag:
break
boxes, area_props = self.autodetect_candidate_boxes(
candidate_values=candidate,
fast_mode=False,
downscale_max_dim=None,
)
score = self.score_detected_boxes(
boxes,
candidate_values=candidate,
image_features=image_features,
box_area_props=area_props,
)
refined_evaluated = idx
if update_status_callback is not None and (idx == 1 or idx == len(finalists) or idx % 4 == 0):
update_status_callback(
f"Auto-tune refine {idx}/{len(finalists)}: {len(boxes)} box(es), score={score:.1f}"
)
if score > best_score:
best_score = score
best_count = len(boxes)
best_values = dict(candidate)
if best_values is None:
best_score, best_count, best_values = coarse_results[0]
self.params.apply_values(best_values)
return {
"profile": resolved_profile,
"count": int(best_count),
"score": float(best_score),
"candidates": coarse_evaluated,
"candidates_total": len(candidates),
"refined": refined_evaluated,
}
Reference in a new issue Copy permalink