ScanSlicer/scanslicer/controller_status.py

from datetime import datetime
import os
import ntpath
from time import monotonic

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)