296 lines
12 KiB
Python
296 lines
12 KiB
Python
import traceback
|
|
|
|
import cv2
|
|
from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal
|
|
from PySide6.QtGui import QImage
|
|
from PySide6.QtWidgets import QMessageBox
|
|
|
|
|
|
# DetectionTaskSignals: class definition for ScanSlicer behavior.
|
|
class DetectionTaskSignals(QObject):
|
|
status = Signal(str)
|
|
finished = Signal(object, object, object)
|
|
failed = Signal(str)
|
|
|
|
|
|
# DetectionTask: class definition for ScanSlicer behavior.
|
|
class DetectionTask(QRunnable):
|
|
# __init__: function definition for ScanSlicer behavior.
|
|
def __init__(self, autoslicer):
|
|
super().__init__()
|
|
self.autoslicer = autoslicer
|
|
self.signals = DetectionTaskSignals()
|
|
|
|
# run: function definition for ScanSlicer behavior.
|
|
def run(self):
|
|
try:
|
|
boxes, filter_preview_image = self.autoslicer.autodetect_slices(
|
|
self.signals.status.emit,
|
|
preview_filter_output=True,
|
|
)
|
|
normal_preview_image = cv2.cvtColor(self.autoslicer.image, cv2.COLOR_BGR2RGB)
|
|
self.signals.finished.emit(boxes, filter_preview_image, normal_preview_image)
|
|
except Exception as exc:
|
|
details = f"{exc}\n\n{traceback.format_exc()}"
|
|
self.signals.failed.emit(details)
|
|
|
|
|
|
# AutotuneTaskSignals: class definition for ScanSlicer behavior.
|
|
class AutotuneTaskSignals(QObject):
|
|
status = Signal(str)
|
|
finished = Signal(object)
|
|
failed = Signal(str)
|
|
|
|
|
|
# AutotuneTask: class definition for ScanSlicer behavior.
|
|
class AutotuneTask(QRunnable):
|
|
# __init__: function definition for ScanSlicer behavior.
|
|
def __init__(self, autoslicer, profile_name):
|
|
super().__init__()
|
|
self.autoslicer = autoslicer
|
|
self.profile_name = profile_name
|
|
self.signals = AutotuneTaskSignals()
|
|
|
|
# run: function definition for ScanSlicer behavior.
|
|
def run(self):
|
|
try:
|
|
result = self.autoslicer.autotune_for_current_image(
|
|
self.signals.status.emit,
|
|
profile_name=self.profile_name,
|
|
)
|
|
self.signals.finished.emit(result)
|
|
except Exception as exc:
|
|
details = f"{exc}\n\n{traceback.format_exc()}"
|
|
self.signals.failed.emit(details)
|
|
|
|
|
|
# rgb_array_to_qimage: function definition for ScanSlicer behavior.
|
|
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)
|