import json import math import os from datetime import datetime import cv2 import numpy as np 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, }