From 501aded7ccf20c38c7ccc78a4dd7be7cd0a7485f Mon Sep 17 00:00:00 2001 From: dev Date: Sun, 22 Feb 2026 15:40:12 +0000 Subject: [PATCH] feat(controller_status): implement class ScanSlicerStatusMixin --- scanslicer/controller_status.py | 161 ++++++++++++++++++++++++++++++++ 1 file changed, 161 insertions(+) diff --git a/scanslicer/controller_status.py b/scanslicer/controller_status.py index 7f2a03b..1f2829c 100644 --- a/scanslicer/controller_status.py +++ b/scanslicer/controller_status.py @@ -7,3 +7,164 @@ from PySide6.QtWidgets import QApplication # ScanSlicerStatusMixin: class definition for ScanSlicer behavior. +class ScanSlicerStatusMixin: + STATUS_EVENT_PUMP_INTERVAL_SEC = 0.05 + STATUS_LOG_FLUSH_INTERVAL_SEC = 0.5 + STATUS_LOG_BATCH_SIZE = 24 + + # update_status_filename: function definition for ScanSlicer behavior. + def update_status_filename(self, image_path=None): + label = getattr(self, "status_filename_label", None) + if label is None: + return + if image_path: + label.setText(ntpath.basename(str(image_path))) + else: + label.setText("No file") + + # resolve_status_log_path: function definition for ScanSlicer behavior. + def resolve_status_log_path(self): + override = os.environ.get("SCANSLICER_STATUS_LOG_PATH") + if override: + return override + + if os.path.exists("/.dockerenv") and os.path.isdir("/work"): + return os.path.join("/work", ".scanslicer", "status.log") + + 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", "status.log") + + # initialize_status_log: function definition for ScanSlicer behavior. + def initialize_status_log(self): + self._status_log_buffer = [] + self._status_log_last_flush_ts = monotonic() + self.status_log_path = self.resolve_status_log_path() + self._status_log_file = None + try: + directory = os.path.dirname(self.status_log_path) + if directory: + os.makedirs(directory, exist_ok=True) + # Start each app launch with a fresh log file. + self._status_log_file = open(self.status_log_path, "w", encoding="utf-8") + self._status_log_file.write("=== ScanSlicer session ") + self._status_log_file.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) + self._status_log_file.write(" ===\n") + self._status_log_file.flush() + except OSError: + self.status_logging_enabled = False + self._status_log_file = None + + # _flush_status_log_if_due: function definition for ScanSlicer behavior. + def _flush_status_log_if_due(self, force=False): + if not self.status_logging_enabled or not self.status_log_path: + return + + buffer = getattr(self, "_status_log_buffer", None) + if not buffer: + return + + now = monotonic() + last_flush = getattr(self, "_status_log_last_flush_ts", 0.0) + due_by_time = (now - last_flush) >= self.STATUS_LOG_FLUSH_INTERVAL_SEC + due_by_size = len(buffer) >= self.STATUS_LOG_BATCH_SIZE + if not force and not due_by_time and not due_by_size: + return + + payload = "".join(buffer) + try: + if self._status_log_file is None: + self._status_log_file = open(self.status_log_path, "a", encoding="utf-8") + self._status_log_file.write(payload) + self._status_log_file.flush() + except OSError: + self.status_logging_enabled = False + self._status_log_buffer = [] + if self._status_log_file is not None: + try: + self._status_log_file.close() + except OSError: + pass + self._status_log_file = None + return + + self._status_log_buffer = [] + self._status_log_last_flush_ts = now + + # _close_status_log: function definition for ScanSlicer behavior. + def _close_status_log(self): + self._flush_status_log_if_due(force=True) + fh = getattr(self, "_status_log_file", None) + if fh is None: + return + try: + fh.close() + except OSError: + pass + self._status_log_file = None + + # append_status_log: function definition for ScanSlicer behavior. + def append_status_log(self, text): + if not self.status_logging_enabled or not self.status_log_path: + return + + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + line = f"[{timestamp}] {text}\n" + buffer = getattr(self, "_status_log_buffer", None) + if buffer is None: + self._status_log_buffer = [line] + self._status_log_last_flush_ts = monotonic() + else: + buffer.append(line) + self._flush_status_log_if_due(force=False) + + # log_detection_config: function definition for ScanSlicer behavior. + def log_detection_config(self): + preset_name = self.preset_dropdown.currentText() + values = self.params.to_dict() + values_text = ", ".join(f"{name}={values[name]}" for name in sorted(values.keys())) + self.append_status_log(f"Detection config: preset={preset_name}") + self.append_status_log(f"Detection params: {values_text}") + + # _pump_ui_events_if_due: function definition for ScanSlicer behavior. + def _pump_ui_events_if_due(self, force=False): + now = monotonic() + last_pump = getattr(self, "_status_last_event_pump_ts", 0.0) + if force or (now - last_pump) >= self.STATUS_EVENT_PUMP_INTERVAL_SEC: + QApplication.processEvents() + self._status_last_event_pump_ts = now + + # update_statusbar: function definition for ScanSlicer behavior. + def update_statusbar(self, text): + self.status_message_label.setText(text) + self.append_status_log(text) + self._pump_ui_events_if_due() + + # set_processing_busy: function definition for ScanSlicer behavior. + def set_processing_busy(self, busy, status_text=None): + busy = bool(busy) + if self.processing_busy == busy: + if status_text is not None: + self.update_statusbar(status_text) + return + + self.processing_busy = busy + enabled = not busy + + if self.controls_scroll is not None: + self.controls_scroll.setEnabled(enabled) + if self.preview_toolbar is not None: + self.preview_toolbar.setEnabled(enabled) + self.slicing_canvas.setEnabled(enabled) + + if status_text is not None: + self.update_statusbar(status_text) + + if enabled: + self.update_quality_controls_state() + self.update_parameter_visibility() + self.refresh_active_box_toolbar() + + self._pump_ui_events_if_due(force=True) + self._flush_status_log_if_due(force=True)