feat(controller_status): implement class ScanSlicerStatusMixin
This commit is contained in:
parent
598c8a4a25
commit
501aded7cc
1 changed files with 161 additions and 0 deletions
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue