feat(controller_detection): implement class ScanSlicerDetectionMixin
This commit is contained in:
dev
parent
a25e4620d7
commit
5400b936de
1 changed files with 225 additions and 0 deletions
|
|
@ -69,3 +69,228 @@ def rgb_array_to_qimage(image):
|
|||
h, w, c = image.shape
|
||||
bytes_per_line = c * w
|
||||
return QImage(image.data, w, h, bytes_per_line, QImage.Format_RGB888).copy()
|
||||
|
||||
|
||||
# ScanSlicerDetectionMixin: class definition for ScanSlicer behavior.
|
||||
class ScanSlicerDetectionMixin:
|
||||
# _get_worker_pool: function definition for ScanSlicer behavior.
|
||||
def _get_worker_pool(self):
|
||||
pool = getattr(self, "_worker_pool", None)
|
||||
if pool is None:
|
||||
pool = QThreadPool(self)
|
||||
pool.setMaxThreadCount(1)
|
||||
pool.setExpiryTimeout(-1)
|
||||
self._worker_pool = pool
|
||||
return pool
|
||||
|
||||
# invalidate_preview_cache: function definition for ScanSlicer behavior.
|
||||
def invalidate_preview_cache(self):
|
||||
self.last_detected_boxes = None
|
||||
self.last_preview_normal_image = None
|
||||
self.last_preview_filter_image = None
|
||||
|
||||
# has_preview_cache: function definition for ScanSlicer behavior.
|
||||
def has_preview_cache(self):
|
||||
return (
|
||||
self.last_detected_boxes is not None
|
||||
and self.last_preview_normal_image is not None
|
||||
and self.last_preview_filter_image is not None
|
||||
)
|
||||
|
||||
# render_cached_preview: function definition for ScanSlicer behavior.
|
||||
def render_cached_preview(self, new_image=False):
|
||||
if not self.has_preview_cache():
|
||||
return False
|
||||
image = self.last_preview_filter_image if self.preview_filter_enabled else self.last_preview_normal_image
|
||||
self.slicing_canvas.set_image(rgb_array_to_qimage(image), new_image)
|
||||
return True
|
||||
|
||||
# update_preview: function definition for ScanSlicer behavior.
|
||||
def update_preview(self, new_image=False, run_detection=True):
|
||||
if self.processing_busy:
|
||||
return
|
||||
if not self.autoslicer.image_loaded():
|
||||
self.update_statusbar("No image loaded.")
|
||||
return
|
||||
|
||||
if not run_detection:
|
||||
if self.render_cached_preview(new_image=new_image):
|
||||
return
|
||||
run_detection = True
|
||||
|
||||
if run_detection:
|
||||
self._start_detection_worker(new_image=bool(new_image))
|
||||
|
||||
# _start_detection_worker: function definition for ScanSlicer behavior.
|
||||
def _start_detection_worker(self, new_image=False):
|
||||
if getattr(self, "_is_closing", False):
|
||||
return
|
||||
if getattr(self, "_detection_running", False):
|
||||
return
|
||||
|
||||
self._detection_task_id += 1
|
||||
task_id = int(self._detection_task_id)
|
||||
self.log_detection_config()
|
||||
self._pending_detection_new_image = bool(new_image)
|
||||
self._active_detection_task_id = task_id
|
||||
self._detection_running = True
|
||||
self.set_processing_busy(True, "Processing image...")
|
||||
|
||||
task = DetectionTask(self.autoslicer)
|
||||
task.signals.status.connect(lambda text, tid=task_id: self._on_detection_status(tid, text))
|
||||
task.signals.finished.connect(
|
||||
lambda bbxs, filter_preview_image, normal_preview_image, tid=task_id: self._on_detection_finished(
|
||||
tid, bbxs, filter_preview_image, normal_preview_image
|
||||
)
|
||||
)
|
||||
task.signals.failed.connect(lambda error_text, tid=task_id: self._on_detection_failed(tid, error_text))
|
||||
self._detection_task = task
|
||||
|
||||
self._get_worker_pool().start(task)
|
||||
|
||||
# _is_active_detection_task: function definition for ScanSlicer behavior.
|
||||
def _is_active_detection_task(self, task_id):
|
||||
return (
|
||||
task_id is not None
|
||||
and int(task_id) == int(getattr(self, "_active_detection_task_id", -1))
|
||||
and bool(getattr(self, "_detection_running", False))
|
||||
)
|
||||
|
||||
# _on_detection_status: function definition for ScanSlicer behavior.
|
||||
def _on_detection_status(self, task_id, text):
|
||||
if getattr(self, "_is_closing", False):
|
||||
return
|
||||
if not self._is_active_detection_task(task_id):
|
||||
return
|
||||
self.update_statusbar(text)
|
||||
|
||||
# _on_detection_finished: function definition for ScanSlicer behavior.
|
||||
def _on_detection_finished(self, task_id, bbxs, filter_preview_image, normal_preview_image):
|
||||
if not self._is_active_detection_task(task_id):
|
||||
return
|
||||
try:
|
||||
if getattr(self, "_is_closing", False):
|
||||
return
|
||||
self.last_detected_boxes = bbxs
|
||||
self.last_preview_normal_image = normal_preview_image
|
||||
self.last_preview_filter_image = filter_preview_image
|
||||
|
||||
image_to_show = (
|
||||
self.last_preview_filter_image if self.preview_filter_enabled else self.last_preview_normal_image
|
||||
)
|
||||
self.slicing_canvas.set_image(
|
||||
rgb_array_to_qimage(image_to_show),
|
||||
getattr(self, "_pending_detection_new_image", False),
|
||||
)
|
||||
self.slicing_canvas.update_bboxes(bbxs)
|
||||
self.refresh_active_box_toolbar()
|
||||
if len(bbxs) == 0:
|
||||
self.update_statusbar("No pictures detected. Try adjusting parameters.")
|
||||
else:
|
||||
self.update_statusbar(f"Detected {len(bbxs)} picture(s).")
|
||||
finally:
|
||||
self._detection_running = False
|
||||
self._detection_task = None
|
||||
self._active_detection_task_id = None
|
||||
self._pending_detection_new_image = False
|
||||
if not getattr(self, "_is_closing", False):
|
||||
self.set_processing_busy(False)
|
||||
|
||||
# _on_detection_failed: function definition for ScanSlicer behavior.
|
||||
def _on_detection_failed(self, task_id, error_text):
|
||||
if not self._is_active_detection_task(task_id):
|
||||
return
|
||||
self._detection_running = False
|
||||
self._detection_task = None
|
||||
self._active_detection_task_id = None
|
||||
self._pending_detection_new_image = False
|
||||
if not getattr(self, "_is_closing", False):
|
||||
QMessageBox.critical(self, "Detection error", error_text)
|
||||
self.update_statusbar("Detection failed.")
|
||||
self.set_processing_busy(False)
|
||||
|
||||
# _start_autotune_worker: function definition for ScanSlicer behavior.
|
||||
def _start_autotune_worker(self):
|
||||
if getattr(self, "_is_closing", False):
|
||||
return
|
||||
if getattr(self, "_autotune_running", False):
|
||||
return
|
||||
|
||||
profile_name = self.autoslicer.normalize_autotune_profile(
|
||||
getattr(self, "autotune_profile", self.autoslicer.default_autotune_profile_name())
|
||||
)
|
||||
self.autotune_profile = profile_name
|
||||
|
||||
self._autotune_task_id += 1
|
||||
task_id = int(self._autotune_task_id)
|
||||
self._active_autotune_task_id = task_id
|
||||
self._autotune_running = True
|
||||
self.set_processing_busy(True, f"Auto-tuning parameters ({profile_name})...")
|
||||
|
||||
task = AutotuneTask(self.autoslicer, profile_name)
|
||||
task.signals.status.connect(lambda text, tid=task_id: self._on_autotune_status(tid, text))
|
||||
task.signals.finished.connect(lambda result, tid=task_id: self._on_autotune_finished(tid, result))
|
||||
task.signals.failed.connect(lambda error_text, tid=task_id: self._on_autotune_failed(tid, error_text))
|
||||
self._autotune_task = task
|
||||
|
||||
self._get_worker_pool().start(task)
|
||||
|
||||
# _is_active_autotune_task: function definition for ScanSlicer behavior.
|
||||
def _is_active_autotune_task(self, task_id):
|
||||
return (
|
||||
task_id is not None
|
||||
and int(task_id) == int(getattr(self, "_active_autotune_task_id", -1))
|
||||
and bool(getattr(self, "_autotune_running", False))
|
||||
)
|
||||
|
||||
# _on_autotune_status: function definition for ScanSlicer behavior.
|
||||
def _on_autotune_status(self, task_id, text):
|
||||
if getattr(self, "_is_closing", False):
|
||||
return
|
||||
if not self._is_active_autotune_task(task_id):
|
||||
return
|
||||
self.update_statusbar(text)
|
||||
|
||||
# _on_autotune_finished: function definition for ScanSlicer behavior.
|
||||
def _on_autotune_finished(self, task_id, result):
|
||||
if not self._is_active_autotune_task(task_id):
|
||||
return
|
||||
self._autotune_running = False
|
||||
self._autotune_task = None
|
||||
self._active_autotune_task_id = None
|
||||
if getattr(self, "_is_closing", False):
|
||||
return
|
||||
self.set_processing_busy(False)
|
||||
self.sync_param_controls_from_model()
|
||||
self.set_preset_dropdown_custom()
|
||||
profile_name = str(result.get("profile", getattr(self, "autotune_profile", "Balanced")))
|
||||
tested = int(result.get("candidates", 0))
|
||||
total = int(result.get("candidates_total", tested))
|
||||
refined = int(result.get("refined", 0))
|
||||
candidate_text = f"{tested}/{total}" if total > tested else str(tested)
|
||||
refine_text = f", refined={refined}" if refined > 0 else ""
|
||||
self.append_status_log(
|
||||
"Auto-tune result: "
|
||||
+ f"profile={profile_name}, "
|
||||
+ f"candidates={candidate_text}, "
|
||||
+ f"refined={refined}, "
|
||||
+ f"best_count={result['count']}, "
|
||||
+ f"best_score={result['score']:.1f}"
|
||||
)
|
||||
self.update_statusbar(
|
||||
f"Auto-tune ({profile_name}) applied best candidate "
|
||||
+ f"({result['count']} boxes from {candidate_text} candidates{refine_text})."
|
||||
)
|
||||
self.update_preview(new_image=False, run_detection=True)
|
||||
|
||||
# _on_autotune_failed: function definition for ScanSlicer behavior.
|
||||
def _on_autotune_failed(self, task_id, error_text):
|
||||
if not self._is_active_autotune_task(task_id):
|
||||
return
|
||||
self._autotune_running = False
|
||||
self._autotune_task = None
|
||||
self._active_autotune_task_id = None
|
||||
if not getattr(self, "_is_closing", False):
|
||||
QMessageBox.critical(self, "Auto-tune failed", error_text)
|
||||
self.update_statusbar("Auto-tune failed.")
|
||||
self.set_processing_busy(False)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue