feat(autoslicer_feedback): implement class AutoslicerFeedbackMixin
This commit is contained in:
parent
1da5f212cb
commit
acd8fd1fb1
1 changed files with 740 additions and 0 deletions
|
|
@ -8,3 +8,743 @@ import numpy as np
|
||||||
|
|
||||||
|
|
||||||
# AutoslicerFeedbackMixin: class definition for ScanSlicer behavior.
|
# AutoslicerFeedbackMixin: class definition for ScanSlicer behavior.
|
||||||
|
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,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
# autotune_profile_names: function definition for ScanSlicer behavior.
|
||||||
|
def autotune_profile_names(self):
|
||||||
|
return list(self.AUTOTUNE_PROFILE_ORDER)
|
||||||
|
|
||||||
|
# default_autotune_profile_name: function definition for ScanSlicer behavior.
|
||||||
|
def default_autotune_profile_name(self):
|
||||||
|
return self.DEFAULT_AUTOTUNE_PROFILE
|
||||||
|
|
||||||
|
# normalize_autotune_profile: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# get_autotune_profile_config: function definition for ScanSlicer behavior.
|
||||||
|
def get_autotune_profile_config(self, profile_name):
|
||||||
|
normalized = self.normalize_autotune_profile(profile_name)
|
||||||
|
return normalized, dict(self.AUTOTUNE_PROFILES[normalized])
|
||||||
|
|
||||||
|
# sample_candidates_evenly: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# feedback_path: function definition for ScanSlicer behavior.
|
||||||
|
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")
|
||||||
|
|
||||||
|
# load_feedback_entries: function definition for ScanSlicer behavior.
|
||||||
|
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:]
|
||||||
|
|
||||||
|
# save_feedback_entries: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# clear_feedback_entries: function definition for ScanSlicer behavior.
|
||||||
|
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,
|
||||||
|
}
|
||||||
|
|
||||||
|
# get_param_range_map: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# compute_image_features: function definition for ScanSlicer behavior.
|
||||||
|
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),
|
||||||
|
}
|
||||||
|
|
||||||
|
# normalize_boxes_for_storage: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# boxes_signature: function definition for ScanSlicer behavior.
|
||||||
|
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)
|
||||||
|
|
||||||
|
# record_feedback: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# image_similarity: function definition for ScanSlicer behavior.
|
||||||
|
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))
|
||||||
|
|
||||||
|
# param_similarity: function definition for ScanSlicer behavior.
|
||||||
|
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))
|
||||||
|
|
||||||
|
# feedback_score_bonus: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# score_detected_boxes: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# build_autotune_candidates: function definition for ScanSlicer behavior.
|
||||||
|
def build_autotune_candidates(self, base_values, profile_name=None):
|
||||||
|
_, profile = self.get_autotune_profile_config(profile_name)
|
||||||
|
base = dict(base_values)
|
||||||
|
candidates = []
|
||||||
|
|
||||||
|
# add: function definition for ScanSlicer behavior.
|
||||||
|
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)
|
||||||
|
|
||||||
|
# rank_autotune_candidates: function definition for ScanSlicer behavior.
|
||||||
|
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]
|
||||||
|
|
||||||
|
# autodetect_candidate_boxes: function definition for ScanSlicer behavior.
|
||||||
|
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
|
||||||
|
|
||||||
|
# autotune_for_current_image: function definition for ScanSlicer behavior.
|
||||||
|
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,
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue