# edited by glg import os import time from PySide6.QtCore import QDateTime, QEvent, QSettings, QTimer, Signal, Qt from PySide6.QtGui import QAction, QKeySequence, QPixmap, QShortcut from PySide6.QtWidgets import ( QHBoxLayout, QLabel, QMainWindow, QMenu, QMessageBox, QPushButton, QSizePolicy, QSpacerItem, QStackedWidget, QToolButton, QVBoxLayout, QWidget, ) from pypos.core.base_view import BaseView from pypos.core.utils.app_state_utils import set_bool from pypos.core.utils.ui_image_asset_utils import load_best_source_pixmap from pypos.core.utils.worker_pool_utils import submit_ui_periodic_task_keyed from pypos.core.utils.path_utils import get_app_data_resource_dir, get_db_path from pypos.core.utils.ui_scale_runtime import register_ui_scale_listener from pypos.core.utils.ui_message_utils import sanitize_ui_message from pypos.modules.customer.views.customer_setup_view import CustomerSetupView from pypos.modules.dashboard.services.dashboard_layout_builder_service import ( DashboardLayoutBuilderService, ) from pypos.modules.dashboard.services.dashboard_window_ui_service import DashboardWindowUiService from pypos.modules.penjualan.controllers.barang_controller import BarangController from pypos.modules.scanner.controllers.scanner_controller import ScannerController from pypos.modules.penjualan.views.pembatalan_view import PembatalanView from pypos.modules.settings.controllers.settings_controller import SettingsController from pypos.modules.settings.views.settings_view import SettingsView from shiboken6 import isValid # edited by glg def _resolve_dashboard_controller_cls(): # Lazy import untuk mengurangi coupling load-time view -> controller. from pypos.modules.dashboard.controllers.dashboard_controller import DashboardController return DashboardController class DashboardWindow(BaseView, QMainWindow): logo_loaded = Signal(object) # edited by glg # Signal hasil probe autosync (thread-safe dispatch ke UI/main thread). auto_sync_update_detected = Signal(object) # edited by glg # Signal hasil probe koneksi agar update status dilakukan di UI thread. koneksi_probe_finished = Signal(bool, float) # edited by glg # Orchestrator status ikon terpadu (WiFi/Printer/Scanner) via satu payload. icon_status_probe_finished = Signal(object) # edited by glg def build_customer_setup_widget(self): return CustomerSetupView() # edited by glg def build_barang_setup_widget(self): controller = BarangController() return controller, controller.get_view() def showEvent(self, event): from pypos.core.utils.window_size_helper import set_window_size_by_screen set_window_size_by_screen(self) super().showEvent(event) def changeEvent(self, event): screen_change = getattr(QEvent.Type, "ScreenChange", None) screen_change_internal = getattr(QEvent.Type, "ScreenChangeInternal", None) dpi_change = getattr(QEvent.Type, "DpiChange", None) dpi_change_internal = getattr(QEvent.Type, "DpiChangeInternal", None) handled_events = { t for t in ( screen_change, screen_change_internal, dpi_change, dpi_change_internal, ) if t is not None } if handled_events and event.type() in handled_events: from pypos.core.utils.window_size_helper import set_window_size_by_screen set_window_size_by_screen(self) if hasattr(self, "sidebar_widget"): self.sidebar_widget.updateGeometry() if hasattr(self, "content_area"): self.content_area.updateGeometry() self._refresh_header_logo_scaled() super().changeEvent(event) def __init__(self, user_info, app_controller, dashboard_controller_cls=None): super().__init__() self.user_info = user_info self.controller = app_controller controller_cls = dashboard_controller_cls or _resolve_dashboard_controller_cls() self.dashboard_controller = controller_cls(self, app_controller) self.scanner_controller = ScannerController() self.settings_controller = SettingsController(scanner_controller=self.scanner_controller) # edited by glg self.auto_sync_update_detected.connect( self.dashboard_controller.on_auto_sync_update_detected, Qt.QueuedConnection, ) # edited by glg # Hasil probe koneksi diproses via queued signal untuk menjaga UI tetap responsif. self.koneksi_probe_finished.connect( self._on_koneksi_probe_finished, Qt.QueuedConnection, ) self.icon_status_probe_finished.connect( self._on_icon_status_probe_finished, Qt.QueuedConnection, ) self._sync_blocked = False # edited by glg self._koneksi_check_in_progress = False self._icon_status_last_signature = None # edited by glg # Cache probe settlement startup agar query status tidak dieksekusi # berulang kali saat user cepat pindah menu. self._settlement_startup_probe_ttl_ms = 1800 self._settlement_startup_last_probe_ms = 0.0 self._settlement_startup_last_probe_result = False # edited by glg # Notifikasi sementara hasil autosync (khusus UI, non-blocking). self._auto_sync_feedback_timer = None self._auto_sync_feedback_text = "" # edited by glg # Status runtime autosync agar user melihat proses sedang berjalan. self._auto_sync_running = False self.dashboard_window_ui_service = DashboardWindowUiService() self._header_logo_original_pixmap = None # Status widgets will be initialized in init_ui() self.status_circle = None self.printer_status_circle = None self.scanner_status_circle = None self.sync_status_label = None self.sync_button = None self.silent_sync_label = None self.waktu_timer = None self.logo_loaded.connect(self._apply_header_logo_bytes, Qt.QueuedConnection) self.setWindowTitle("Dashboard - Aplikasi POS") self.init_ui() register_ui_scale_listener(self._on_runtime_ui_scale_changed) # Timer auto-refresh koneksi self.timer = QTimer(self) self.timer.timeout.connect(self.cek_koneksi_realtime) # edited by glg probe_interval_ms = self.dashboard_controller.get_online_poll_interval_ms() self.timer.start(probe_interval_ms) def cek_koneksi_realtime(self): # edited by glg if self._koneksi_check_in_progress: return self._koneksi_check_in_progress = True started = {"value": False} def _worker(): started["value"] = True self._run_icon_status_probe_worker() try: key = f"dashboard-online-probe:{id(self)}" future = submit_ui_periodic_task_keyed(key, _worker) if future is None: self._koneksi_check_in_progress = False return # edited by glg # Fail-safe antrean penuh: task bisa ter-drop sebelum worker mulai. # Jangan biarkan guard in-progress terkunci. def _on_done(_): if not bool(started.get("value")): self._koneksi_check_in_progress = False try: future.add_done_callback(_on_done) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): if not bool(started.get("value")): self._koneksi_check_in_progress = False except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self._koneksi_check_in_progress = False self.log_warning(f"Gagal start worker probe koneksi: {e}") # edited by glg # Orchestrator status ikon: probe WiFi/Printer/Scanner di satu worker. def _run_icon_status_probe_worker(self): started_at = time.perf_counter() is_online = False scanner_runtime = {} printer_payload = {} errors = [] try: is_online = bool(self.dashboard_controller.check_online_status_for_ui()) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: errors.append(f"wifi:{e}") try: scanner = getattr(self, "scanner_controller", None) if scanner and hasattr(scanner, "get_runtime_status"): runtime = scanner.get_runtime_status() if isinstance(runtime, dict): scanner_runtime = runtime except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: errors.append(f"scanner:{e}") try: printer_widget = getattr(self, "printer_status_circle", None) if printer_widget is not None and hasattr(printer_widget, "get_status_payload_snapshot"): probe = printer_widget.get_status_payload_snapshot() if isinstance(probe, dict): printer_payload = probe except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: errors.append(f"printer:{e}") elapsed_ms = self._elapsed_ms(started_at) try: payload = { "is_online": bool(is_online), "scanner_runtime": scanner_runtime if isinstance(scanner_runtime, dict) else {}, "printer_payload": printer_payload if isinstance(printer_payload, dict) else {}, "elapsed_ms": float(elapsed_ms), "errors": list(errors), } self.icon_status_probe_finished.emit(payload) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): self._koneksi_check_in_progress = False # edited by glg def _on_icon_status_probe_finished(self, payload): self._koneksi_check_in_progress = False data = payload if isinstance(payload, dict) else {} is_online = bool(data.get("is_online")) scanner_runtime = data.get("scanner_runtime") if isinstance(data.get("scanner_runtime"), dict) else {} printer_payload = data.get("printer_payload") if isinstance(data.get("printer_payload"), dict) else {} self.update_koneksi_status( bool(is_online), scanner_runtime_status=scanner_runtime, printer_payload=printer_payload, ) elapsed_ms = float(data.get("elapsed_ms") or 0.0) if elapsed_ms >= 150.0: self.log_info(f"[PERF] dashboard_icon_probe elapsed_ms={elapsed_ms:.1f}") for err in list(data.get("errors") or [])[:3]: self.log_warning(f"Probe status ikon gagal parsial: {str(err)}") # edited by glg def _on_koneksi_probe_finished(self, is_online, elapsed_ms): self._koneksi_check_in_progress = False self.update_koneksi_status(bool(is_online)) if float(elapsed_ms or 0.0) >= 150.0: self.log_info(f"[PERF] dashboard_online_probe elapsed_ms={float(elapsed_ms):.1f}") def update_koneksi_status(self, is_online, scanner_runtime_status=None, printer_payload=None): try: if self.status_circle is None or not isValid(self.status_circle): if self.timer and self.timer.isActive(): self.timer.stop() return if self.printer_status_circle is None or not isValid(self.printer_status_circle): if self.timer and self.timer.isActive(): self.timer.stop() return if self.scanner_status_circle is None or not isValid(self.scanner_status_circle): if self.timer and self.timer.isActive(): self.timer.stop() return except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): return scanner_sig = () if isinstance(scanner_runtime_status, dict): rawinput = scanner_runtime_status.get("rawinput") if isinstance(scanner_runtime_status.get("rawinput"), dict) else {} activity = ( scanner_runtime_status.get("scanner_activity") if isinstance(scanner_runtime_status.get("scanner_activity"), dict) else {} ) scanner_sig = ( int(bool(rawinput.get("active"))), int(bool(rawinput.get("capture_enabled"))), int(bool(activity.get("last_scan_recent"))), str(activity.get("last_scan_source") or ""), ) signature = ( int(bool(is_online)), str((printer_payload or {}).get("printer_name") if isinstance(printer_payload, dict) else ""), int(bool((printer_payload or {}).get("printing_on")) if isinstance(printer_payload, dict) else -1), scanner_sig, ) if signature == self._icon_status_last_signature: return self._icon_status_last_signature = signature self.status_circle.set_online(is_online) # edited by glg # Satu jalur ikon: # - probe printer dikerjakan di worker orchestrator dashboard # - widget printer hanya apply payload final di UI thread. try: if ( self.printer_status_circle is not None and isValid(self.printer_status_circle) and isinstance(printer_payload, dict) and hasattr(self.printer_status_circle, "apply_status_payload") ): self.printer_status_circle.apply_status_payload(printer_payload) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal apply payload status printer: {e}") self.update_scanner_status(runtime_status=scanner_runtime_status) def update_scanner_status(self, runtime_status=None): data = runtime_status if isinstance(runtime_status, dict) else {} if not data: try: if hasattr(self, "scanner_controller") and self.scanner_controller: if hasattr(self.scanner_controller, "get_runtime_status"): payload = self.scanner_controller.get_runtime_status() if isinstance(payload, dict): data = payload except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal baca runtime scanner: {e}") try: if self.scanner_status_circle is not None and isValid(self.scanner_status_circle): self.scanner_status_circle.update_status(data) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal update ikon scanner: {e}") def cek_status_printer_default(self): try: default_printer = self.settings_controller.get_default_printer() if default_printer and default_printer.get("name"): return True except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal cek status printer default: {e}") return False def update_sync_status(self, needs_sync): payload = self._get_dashboard_window_ui_service().build_sync_status_payload(needs_sync) self.sync_status_label.setText(str(payload.get("text") or "")) self.sync_status_label.setStyleSheet(str(payload.get("style") or "")) if bool(payload.get("show_button")): self.sync_button.show() return self.sync_button.hide() def lakukan_sinkronisasi(self): # edited by glg # Jika ada update pending dari probe autosync, tampilkan kembali # dialog konfirmasi update yang sama saat user klik "Sinkron Sekarang". if hasattr(self, "dashboard_controller") and self.dashboard_controller: if self.dashboard_controller.handle_sync_now_request(): return self.log_info("Sinkronisasi dimulai...") self.controller.mulai_sinkronisasi_background() def set_sinkron_status(self, text): if self.silent_sync_label: self.silent_sync_label.setText(text) # edited by glg def _bind_auto_sync_feedback_signal(self): try: service = getattr(self, "auto_sync_service", None) if not service: return service.progress_updated.connect( self._on_auto_sync_progress_feedback, Qt.QueuedConnection, ) service.sync_completed.connect( self._on_auto_sync_completed_feedback, Qt.QueuedConnection, ) service.sync_failed.connect( self._on_auto_sync_failed_feedback, Qt.QueuedConnection, ) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal bind notifikasi autosync: {e}") # edited by glg def _get_dashboard_window_ui_service(self) -> DashboardWindowUiService: service = getattr(self, "dashboard_window_ui_service", None) if service is None: service = DashboardWindowUiService() self.dashboard_window_ui_service = service return service # edited by glg def _format_row_count(self, value): return self._get_dashboard_window_ui_service().format_row_count(value) # edited by glg def _elapsed_ms(self, started_at): return self._get_dashboard_window_ui_service().elapsed_ms(started_at) # edited by glg def _log_perf_metric(self, label, elapsed_ms, threshold_ms=180.0, extra="", always_info=False): service = self._get_dashboard_window_ui_service() message = service.build_perf_log_message(label, elapsed_ms, extra=extra) if service.should_log_info(elapsed_ms, threshold_ms=threshold_ms, always_info=always_info): self.log_info(message) return self.log_debug(message) # edited by glg def _clear_auto_sync_feedback_status(self): try: if ( self.silent_sync_label and str(self.silent_sync_label.text() or "").strip() == str(self._auto_sync_feedback_text or "").strip() ): self.silent_sync_label.setText("") except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_debug(f"Cleanup status autosync diabaikan: {e}") self._auto_sync_feedback_text = "" # edited by glg def _schedule_auto_sync_feedback_clear(self, delay_ms=8000): if self._auto_sync_feedback_timer is None: self._auto_sync_feedback_timer = QTimer(self) self._auto_sync_feedback_timer.setSingleShot(True) self._auto_sync_feedback_timer.timeout.connect(self._clear_auto_sync_feedback_status) elif self._auto_sync_feedback_timer.isActive(): self._auto_sync_feedback_timer.stop() self._auto_sync_feedback_timer.start(max(1000, int(delay_ms or 0))) # edited by glg def _on_auto_sync_progress_feedback(self, percent, status, detail_log): try: progress = max(0, min(100, int(percent or 0))) except (TypeError, ValueError): progress = 0 status_text = str(status or "").strip() or "Sinkronisasi produk" if progress < 100: self._auto_sync_running = True if self.sync_button is not None: self.sync_button.setEnabled(False) self.set_sinkron_status(f"{status_text}... {progress}%") return self._auto_sync_running = False if self.sync_button is not None: self.sync_button.setEnabled(True) # edited by glg def _on_auto_sync_completed_feedback(self, total_rows): self._auto_sync_running = False if self.sync_button is not None: self.sync_button.setEnabled(True) payload = self._get_dashboard_window_ui_service().build_auto_sync_feedback_payload(total_rows) if not payload: return self._auto_sync_feedback_text = str(payload.get("status_text") or "") self.set_sinkron_status(self._auto_sync_feedback_text) self.show_toast( str(payload.get("toast_text") or ""), duration_ms=int(payload.get("toast_duration_ms") or 3600), level="success", ) self._schedule_auto_sync_feedback_clear(delay_ms=int(payload.get("clear_delay_ms") or 9000)) # edited by glg def _on_auto_sync_failed_feedback(self, error_message): self._auto_sync_running = False if self.sync_button is not None: self.sync_button.setEnabled(True) self.set_sinkron_status("Sinkronisasi produk gagal. Periksa koneksi lalu coba lagi.") self.show_toast( "Sinkronisasi produk gagal. Cek koneksi internet/server.", duration_ms=3600, level="warning", ) self._schedule_auto_sync_feedback_clear(delay_ms=10000) def init_ui(self): from pypos.modules.dashboard.widgets.status_circle_widget import StatusCircle from pypos.modules.printer.widgets.printer_status_circle import PrinterStatusCircle from pypos.modules.dashboard.widgets.scanner_status_circle_widget import ScannerStatusCircle layout_cfg = self.dashboard_controller.get_ui_layout_config() logo_size = layout_cfg["logo_size"] logo_width = layout_cfg["logo_width"] # edited by glg # Kompres aturan logo sedikit agar tidak terlihat kepotong pada navbar # ketika tinggi bar/header ketat di beberapa resolusi. try: logo_size = max(24, int(round(float(logo_size) * 0.88))) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): logo_size = 64 try: logo_width = max(72, int(round(float(logo_width) * 0.88))) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): logo_width = 120 sidebar_width = layout_cfg["sidebar_width"] sidebar_btn_height = layout_cfg["sidebar_btn_height"] sidebar_spacing = layout_cfg["sidebar_spacing"] sidebar_margin = layout_cfg["sidebar_margin"] status_icon_box = layout_cfg["status_icon_box"] status_icon_size = layout_cfg["status_icon_size"] self.status_circle = StatusCircle( online=False, box_size=status_icon_box, icon_size=status_icon_size ) self.printer_status_circle = PrinterStatusCircle( box_size=status_icon_box, icon_size=status_icon_size ) # edited by glg # Semua ikon diprobe melalui satu jalur orchestrator dashboard. # Matikan timer internal printer agar tidak terjadi double polling. if hasattr(self.printer_status_circle, "set_polling_enabled"): try: self.printer_status_circle.set_polling_enabled(False) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal matikan polling internal printer: {e}") self.scanner_status_circle = ScannerStatusCircle( box_size=status_icon_box, icon_size=status_icon_size ) self.sync_status_label = QLabel("\u2705 Data Up-to-date") self.sync_button = QPushButton("Sinkron Sekarang") self.sync_button.hide() self.sync_button.clicked.connect(self.lakukan_sinkronisasi) self.silent_sync_label = QLabel("") self.sync_status_label.setStyleSheet("color: white; font-weight: bold;") self.silent_sync_label.setStyleSheet("color: white; font-size: 10px;") main_bar = QWidget() main_bar.setStyleSheet("background-color: #2C3E50; color: white; padding: 6px;") main_bar_layout = QHBoxLayout(main_bar) self.logo_header_label = QLabel() # edited by glg # Batasi ruang logo agar navbar tidak memakan lebar berlebih. # Tinggi tetap mengikuti layout, lebar mengikuti batas maksimum. self.logo_header_label.setMinimumSize(1, logo_size) self.logo_header_label.setMaximumWidth(logo_width) self.logo_header_label.setSizePolicy(QSizePolicy.Maximum, QSizePolicy.Preferred) self.logo_header_label.setAlignment(Qt.AlignLeft | Qt.AlignVCenter) # Scaling pixmap sudah dikendalikan manual di _refresh_header_logo_scaled. self.logo_header_label.setScaledContents(False) self.logo_header_label.setStyleSheet("margin-right: 6px;") self.logo_header_label.hide() self.label_title = QLabel("") self.label_title.setStyleSheet("font-weight: bold; font-size: 18px;") self.label_title.hide() spacer = QSpacerItem(40, 20, QSizePolicy.Expanding, QSizePolicy.Minimum) self.user_menu_button = QToolButton() self.user_menu_button.setText(f"User: {self.user_info['nama']}") self.user_menu_button.setPopupMode(QToolButton.InstantPopup) self.user_menu_button.setStyleSheet( "QToolButton { color: white; background: transparent; border: none; }" "QToolButton::menu-indicator { image: none; }" ) self.user_menu = QMenu(self) self.action_change_password = QAction("Ubah Password", self) self.action_change_password.triggered.connect(self._open_change_password_dialog) self.user_menu.addAction(self.action_change_password) self.user_menu_button.setMenu(self.user_menu) self.label_waktu = QLabel("") main_bar_layout.addWidget(self.logo_header_label) main_bar_layout.addWidget(self.label_title) main_bar_layout.addItem(spacer) main_bar_layout.addWidget(self.label_waktu) main_bar_layout.addSpacing(2) main_bar_layout.addWidget(self.status_circle) main_bar_layout.addSpacing(5) main_bar_layout.addWidget(self.printer_status_circle) main_bar_layout.addSpacing(5) main_bar_layout.addWidget(self.scanner_status_circle) main_bar_layout.addSpacing(10) main_bar_layout.addWidget(self.sync_button) main_bar_layout.addWidget(self.silent_sync_label) main_bar_layout.addSpacing(10) main_bar_layout.addWidget(self.user_menu_button) # Update status awal self.update_koneksi_status(is_online=False) self.dashboard_controller.load_header_logo() self.dashboard_controller.init_auto_sync_master() # edited by glg # Tampilkan notifikasi sementara saat autosync 5 menit benar-benar mengubah data. self._bind_auto_sync_feedback_signal() self.dashboard_controller.init_export_json_batch() # ---------- LAYOUT UTAMA ---------- main_widget = QWidget() layout_utama = QHBoxLayout(main_widget) sidebar_widget = DashboardLayoutBuilderService.setup_sidebar( self, sidebar_btn_height=sidebar_btn_height, sidebar_spacing=sidebar_spacing, sidebar_margin=sidebar_margin, sidebar_width=sidebar_width, ) DashboardLayoutBuilderService.setup_content_area(self) # Layout penggabung layout_utama.addWidget(sidebar_widget) layout_utama.addWidget(self.content_area) # edited by glg # Pastikan konten mengambil sisa lebar, sidebar tetap selebar teks. layout_utama.setStretch(0, 0) layout_utama.setStretch(1, 1) main_layout = QVBoxLayout() main_layout.setContentsMargins(0, 0, 0, 0) main_layout.setSpacing(0) main_layout.addWidget(main_bar) main_layout.addWidget(main_widget) container = QWidget() container.setLayout(main_layout) self.setCentralWidget(container) self._startup_settlement_checked = False QTimer.singleShot(400, self._cek_settlement_awal) DashboardLayoutBuilderService.bind_menu_actions(self) DashboardLayoutBuilderService.bind_shortcuts(self) self._start_waktu_timer() def _get_logo_cache_path(self): logo_dir = os.path.join(get_app_data_resource_dir(), "logo") try: os.makedirs(logo_dir, exist_ok=True) except OSError: return None return os.path.join(logo_dir, "ui_logo_header.png") def _set_header_logo(self, pixmap): if pixmap is None or pixmap.isNull(): return self._header_logo_original_pixmap = QPixmap(pixmap) self._refresh_header_logo_scaled() # edited by glg # Logo header disimpan dalam bentuk asli agar re-render tetap tajam saat skala berubah. def _refresh_header_logo_scaled(self): source = self._header_logo_original_pixmap if source is None or source.isNull(): return target_height = self.logo_header_label.height() or self.logo_header_label.minimumHeight() or 70 max_width = self.logo_header_label.maximumWidth() if max_width <= 0: max_width = max(1, int(source.width())) # edited by glg # Sisakan safe-area kecil agar logo tidak menempel ke tepi label # dan menghindari efek visual "kepotong" pada beberapa DPI. safe_height = max(1, int(round(float(target_height) * 0.90))) safe_width = max(1, int(round(float(max_width) * 0.90))) scaled = source.scaled( safe_width, safe_height, Qt.KeepAspectRatio, Qt.SmoothTransformation, ) self.logo_header_label.setPixmap(scaled) self.logo_header_label.show() # edited by glg def _on_runtime_ui_scale_changed(self, _scale): try: QTimer.singleShot(0, self._refresh_header_logo_scaled) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): pass def _apply_header_logo_path(self, path): if not path or not os.path.isfile(path): return pixmap = load_best_source_pixmap( str(path), prefer_svg=True, svg_render_size=( int(self.logo_header_label.maximumWidth() or 512), int(self.logo_header_label.minimumHeight() or 256), ), ) if pixmap.isNull(): return self._set_header_logo(pixmap) def _apply_header_logo_bytes(self, content_bytes): if not content_bytes: return pixmap = QPixmap() if not pixmap.loadFromData(content_bytes): return self._set_header_logo(pixmap) cache_path = self._get_logo_cache_path() if cache_path: try: pixmap.save(cache_path, "PNG") except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal simpan cache logo: {e}") def _open_change_password_dialog(self): try: from pypos.modules.auth.views.change_password_dialog import ChangePasswordDialog dialog = ChangePasswordDialog(self.user_info, self) dialog.exec() except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal membuka dialog ubah password: {e}") safe_message, _ = sanitize_ui_message("warning", str(e)) QMessageBox.warning(self, "Gagal", f"Gagal membuka dialog ubah password.\n{safe_message}") def navigate_to_page(self, page_name): """Navigate to specific page""" # edited by glg # Instrumentasi latency perpindahan menu untuk baseline performa UI. started_at = time.perf_counter() page_key = str(page_name).lower() status = "unknown_page" try: if self._sync_blocked: status = "blocked_sync" QMessageBox.information( self, "Sinkronisasi Berjalan", "Sinkronisasi sedang berjalan.\n" "Mohon tunggu hingga selesai sebelum berpindah halaman." ) return if page_key != "sinkron" and self._enforce_settlement_navigation_lock(): status = "blocked_settlement_lock" return page_mapping = { "beranda": lambda: self.content_area.setCurrentWidget(self.dashboard_info.get_view()), "penjualan": self.buka_penjualan, "sinkron": self.buka_menu_sinkron_data, "printer": self.buka_pengaturan_printer } action = page_mapping.get(page_key) if action: self.log_debug(f"Navigating to: {page_name}") action() status = f"ok:{page_key}" else: self.log_warning(f"Unknown page: {page_name}") finally: self._log_perf_metric( "navigate_to_page", self._elapsed_ms(started_at), threshold_ms=220.0, extra=f"page={page_key} status={status}", always_info=True, ) def _is_settlement_navigation_blocked(self): controller = getattr(self, "transaksi_controller", None) if controller is None or not hasattr(controller, "is_settlement_blocked"): return False try: return bool(controller.is_settlement_blocked()) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal baca status settlement lock: {e}") return False def _enforce_settlement_navigation_lock(self): if not self._is_settlement_navigation_blocked(): return False current_widget = self.content_area.currentWidget() if hasattr(self, "content_area") else None settlement_container = getattr(self, "settlement_container", None) settlement_view = getattr(self, "settlement_view", None) is_settlement_page = current_widget in (settlement_container, settlement_view) if not is_settlement_page: QMessageBox.information( self, "Settlement Wajib", "Ada transaksi penjualan yang belum diselesaikan.\n" "Selesaikan settlement terlebih dahulu." ) self.buka_page_settlement() return True def update_waktu(self): if not hasattr(self, "label_waktu"): return try: label = self.label_waktu except RuntimeError: if self.waktu_timer and self.waktu_timer.isActive(): self.waktu_timer.stop() return if label is None: return try: if not isValid(label): if self.waktu_timer and self.waktu_timer.isActive(): self.waktu_timer.stop() return except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError): return now = QDateTime.currentDateTime() hari_map = { "Monday": "Senin", "Tuesday": "Selasa", "Wednesday": "Rabu", "Thursday": "Kamis", "Friday": "Jumat", "Saturday": "Sabtu", "Sunday": "Minggu", } bulan_map = { "January": "Januari", "February": "Februari", "March": "Maret", "April": "April", "May": "Mei", "June": "Juni", "July": "Juli", "August": "Agustus", "September": "September", "October": "Oktober", "November": "November", "December": "Desember", } hari_en = now.toString("dddd") bulan_en = now.toString("MMMM") hari_id = hari_map.get(hari_en, hari_en) bulan_id = bulan_map.get(bulan_en, bulan_en) try: label.setText( f"{hari_id}, {now.toString('dd')} {bulan_id} {now.toString('yyyy hh:mm:ss')}" ) except RuntimeError: if self.waktu_timer and self.waktu_timer.isActive(): self.waktu_timer.stop() return def _start_waktu_timer(self): if self.waktu_timer and self.waktu_timer.isActive(): self.waktu_timer.stop() self.waktu_timer = QTimer(self) self.waktu_timer.timeout.connect(self.update_waktu) self.waktu_timer.start(1000) self.update_waktu() def _cek_settlement_awal(self): if getattr(self, "_startup_settlement_checked", False): return self._startup_settlement_checked = True controller = getattr(self, "transaksi_controller", None) if controller: if self._should_check_settlement_startup(): controller.cek_status_settlement(include_today=True, enforce_kasir=True, async_mode=True) def _should_check_settlement_startup(self): settings = QSettings("MayaGrahaKencana", "POS_System") today = QDateTime.currentDateTime().toString("yyyy-MM-dd") current_user_id = "" if self.user_info and self.user_info.get("id") is not None: current_user_id = str(self.user_info.get("id")) last_date = settings.value("settlement/last_check_date", "") last_user_id = settings.value("settlement/last_user_id", "") should_check = (today != last_date) or (current_user_id != last_user_id) if should_check: settings.setValue("settlement/last_check_date", today) settings.setValue("settlement/last_user_id", current_user_id) return True now_ms = float(time.monotonic() * 1000.0) last_probe_ms = float(getattr(self, "_settlement_startup_last_probe_ms", 0.0) or 0.0) cache_ttl_ms = float(getattr(self, "_settlement_startup_probe_ttl_ms", 0.0) or 0.0) if last_probe_ms > 0 and cache_ttl_ms > 0 and (now_ms - last_probe_ms) < cache_ttl_ms: return bool(getattr(self, "_settlement_startup_last_probe_result", False)) # edited by glg # Hindari query SQLite sinkron di UI thread. # Saat cache kedaluwarsa, cukup trigger pengecekan settlement async sekali per TTL. self._settlement_startup_last_probe_ms = now_ms self._settlement_startup_last_probe_result = False return True def buka_penjualan(self): started_at = time.perf_counter() status = "ok" try: if self._sync_blocked: status = "blocked_sync" QMessageBox.information( self, "Sinkronisasi Berjalan", "Transaksi tidak dapat dilakukan saat sinkronisasi berjalan." ) return if not self._ensure_transaksi_controller_initialized(): status = "init_transaksi_failed" return if self._enforce_settlement_navigation_lock(): status = "blocked_settlement_lock" return self.content_area.setCurrentWidget(self.transaksi_view) if self._should_check_settlement_startup(): # edited by glg # Tunda cek settlement ke event-loop berikutnya agar perpindahan # halaman tetap terasa instan. QTimer.singleShot(0, lambda: self.transaksi_controller.cek_status_settlement(async_mode=True)) # Fokus ke barang_input setelah sedikit delay agar UI stabil QTimer.singleShot(100, lambda: self.transaksi_view.barang_input.setFocus()) finally: self._log_perf_metric( "buka_penjualan", self._elapsed_ms(started_at), threshold_ms=220.0, extra=f"status={status}", always_info=True, ) # edited by glg def _ensure_transaksi_controller_initialized(self): controller = getattr(self, "transaksi_controller", None) view = getattr(self, "transaksi_view", None) if controller and view: return True try: controller = self.dashboard_controller.create_transaksi_penjualan_controller( self.user_info, self, scanner_controller=self.scanner_controller, ) view = controller.view self.transaksi_controller = controller self.transaksi_view = view if self.content_area.indexOf(view) < 0: self.content_area.addWidget(view) return True except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_error(f"Gagal inisialisasi modul transaksi: {e}") self.show_error("Modul Penjualan gagal dimuat.", title="Gagal Memuat Penjualan") return False def buka_page_pembatalan(self): if self._enforce_settlement_navigation_lock(): return # edited by glg # PembatalanView mewajibkan tiga controller domain. # Jangan instantiate langsung dengan parent saja karena akan raise ValueError. if not hasattr(self, "pembatalan_view") or self.pembatalan_view is None: self.pembatalan_view = self._build_pembatalan_view() if self.pembatalan_view is None: return container = self._wrap_with_sticky_menu(self.pembatalan_view, "pembatalan_container") self.content_area.setCurrentWidget(container) if hasattr(self.pembatalan_view, "refresh_active_tab"): # edited by glg # Defer refresh tab agar klik menu tidak menunggu query tab aktif. QTimer.singleShot(0, self.pembatalan_view.refresh_active_tab) # edited by glg def _build_pembatalan_view(self): try: from pypos.modules.penjualan.services.controller_factory_service import ( PenjualanControllerFactoryService, ) factory = PenjualanControllerFactoryService() pembatalan_controller = factory.create_pembatalan_transaksi_controller( user_info=self.user_info, db_path=get_db_path(), ) return_controller = factory.create_return_controller() return_history_controller = factory.create_return_history_controller( return_model=getattr(return_controller, "model", None), pembatalan_model=getattr(pembatalan_controller, "model", None), ) return PembatalanView( parent=self, pembatalan_controller=pembatalan_controller, return_controller=return_controller, return_history_controller=return_history_controller, ) except ( TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError, ) as exc: self.log_error(f"Gagal inisialisasi halaman pembatalan: {exc}") self.show_error( "Halaman pembatalan gagal dimuat. Silakan coba lagi.", title="Gagal Membuka Pembatalan", ) return None def buka_page_history(self): started_at = time.perf_counter() status = "ok" created_now = False try: if self._enforce_settlement_navigation_lock(): status = "blocked_settlement_lock" return from datetime import datetime user_id = self.user_info.get("id") if self.user_info else None today = datetime.now().strftime("%Y-%m-%d") if not hasattr(self, "history_controller") or self.history_controller is None: self.history_controller = self.dashboard_controller.create_history_transaksi_controller( self, user_id, today, self.buka_penjualan, ) self.history_view = self.history_controller.view status = "created_controller" created_now = True else: self.history_controller.user_id = user_id status = "reuse_controller" if hasattr(self.history_controller.view, "set_date_inputs"): self.history_controller.view.set_date_inputs( self.history_controller.start_date, self.history_controller.end_date, ) container = self._wrap_with_sticky_menu(self.history_view, "history_container") self.content_area.setCurrentWidget(container) # edited by glg # Defer refresh saat reuse agar perpindahan menu terasa instan. if not created_now: QTimer.singleShot(0, self.history_controller.load_history_transaksi) finally: self._log_perf_metric( "buka_page_history", self._elapsed_ms(started_at), threshold_ms=260.0, extra=f"status={status}", always_info=True, ) def _on_settlement_close_requested(self): # edited by glg def _runner(): try: controller = getattr(self, "transaksi_controller", None) if controller: controller.cek_status_settlement(async_mode=True) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal refresh status settlement setelah close page: {e}") self.buka_penjualan() QTimer.singleShot(0, _runner) def buka_page_settlement(self): started_at = time.perf_counter() status = "ok" created_now = False try: if not hasattr(self, "settlement_controller_page") or self.settlement_controller_page is None: self.settlement_controller_page = self.dashboard_controller.create_settlement_controller( self.user_info, self, self._on_settlement_close_requested, ) self.settlement_view = self.settlement_controller_page.view status = "created_controller" created_now = True else: self.settlement_controller_page.user_info = self.user_info status = "reuse_controller" if hasattr(self.settlement_controller_page, "view") and self.settlement_controller_page.view is not None: self.settlement_controller_page.view._on_close_requested = self._on_settlement_close_requested # edited by glg # Refresh data settlement via controller agar state view/input konsisten saat page dibuka ulang. refresh_settlement_after_navigation = False if hasattr(self.settlement_controller_page, "refresh_page_data"): # edited by glg # Controller baru sudah refresh di constructor, jadi cukup refresh saat reuse. if not created_now: refresh_settlement_after_navigation = True else: refresh_settlement_after_navigation = True container = self._wrap_with_sticky_menu(self.settlement_view, "settlement_container") self.content_area.setCurrentWidget(container) # edited by glg # Defer refresh agar klik menu settlement tidak terhambat query awal. if refresh_settlement_after_navigation: if hasattr(self.settlement_controller_page, "refresh_page_data"): QTimer.singleShot( 0, lambda: self.settlement_controller_page.refresh_page_data(reset_input=True), ) else: QTimer.singleShot(0, self.settlement_controller_page.load_transaksi) QTimer.singleShot( 0, lambda: self.settlement_controller_page.view.tampilkan_history( self.settlement_controller_page.model.get_last_settlements() ), ) finally: self._log_perf_metric( "buka_page_settlement", self._elapsed_ms(started_at), threshold_ms=280.0, extra=f"status={status}", always_info=True, ) def _wrap_with_sticky_menu(self, page_widget, container_attr_name: str): if not hasattr(self, container_attr_name) or getattr(self, container_attr_name) is None: container = QWidget() layout = QVBoxLayout(container) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) layout.setAlignment(Qt.AlignTop) sticky_menu = None if hasattr(self, "transaksi_view") and hasattr(self.transaksi_view, "create_shortcut_table"): sticky_menu = self.transaksi_view.create_shortcut_table() if sticky_menu is not None: layout.addWidget(sticky_menu) if hasattr(page_widget, "setSizePolicy"): page_widget.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) layout.addWidget(page_widget, 1) setattr(self, container_attr_name, container) self.content_area.addWidget(container) return getattr(self, container_attr_name) def buka_menu_sinkron_data(self): self.log_debug('buka_menu_sinkron_data dipanggil') if hasattr(self, "dashboard_controller") and self.dashboard_controller: status = self.dashboard_controller.request_manual_sync() if status == "queued": QMessageBox.information( self, "Sinkronisasi", "Sinkronisasi otomatis sedang berjalan.\n" "Sinkronisasi manual akan dimulai setelah selesai." ) return if self._sync_blocked: QMessageBox.information( self, "Sinkronisasi Berjalan", "Sinkronisasi sedang berjalan.\nMohon tunggu hingga selesai." ) return if self.dashboard_controller.is_any_sync_running(): QMessageBox.information( self, "Sinkronisasi Berjalan", "Sinkronisasi lain sedang berjalan.\nMohon tunggu hingga selesai." ) return confirm = QMessageBox(self) confirm.setIcon(QMessageBox.Warning) confirm.setWindowTitle("Konfirmasi Sinkronisasi") confirm.setText( "PERHATIAN: Sinkronisasi akan menghapus dan mengisi ulang data master.\n\n" "Dampak selama sinkron:\n" "- Proses dapat memakan waktu beberapa menit (tergantung data)\n" "- Anda tidak dapat melakukan transaksi atau berpindah halaman\n" "- Data karyawan/akun akan disinkron ulang dari server\n" "- Password karyawan akan kembali mengikuti data server pusat\n\n" "Pastikan koneksi stabil. Lanjutkan sinkronisasi?" ) confirm.setStandardButtons(QMessageBox.Yes | QMessageBox.No) confirm.setDefaultButton(QMessageBox.No) if confirm.exec() != QMessageBox.Yes: return self._set_sync_blocked(True) from pypos.modules.sinkronisasi.views.sinkron_view import SinkronView self.sinkron_view = SinkronView() from pypos.modules.sinkronisasi.controllers.sync_flow_controller import SyncFlowController self.sync_flow_controller = SyncFlowController( mode='manual', view=self.sinkron_view, user_info=self.user_info, app_controller=self.controller ) self.sync_flow_controller.sync_finished.connect(self.on_sinkron_manual_selesai) # Tambahkan tampilan ke content area self.content_area.addWidget(self.sinkron_view) self.content_area.setCurrentWidget(self.sinkron_view) self.sync_flow_controller.start() def on_sinkron_manual_selesai(self, success): """Handler ketika sinkronisasi manual dari dashboard selesai""" self.log_debug(f"Sinkronisasi manual selesai, success={success}") self._set_sync_blocked(False) # Hapus sinkron_view dari content area if hasattr(self, 'sinkron_view') and self.sinkron_view is not None: self.content_area.removeWidget(self.sinkron_view) self.sinkron_view.deleteLater() self.sinkron_view = None if self.controller and self.user_info: try: if hasattr(self, "dashboard_controller") and self.dashboard_controller: self.dashboard_controller.load_header_logo(force_refresh=True) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal refresh logo header pasca-sinkron: {e}") try: self.settings_controller.refresh_company_logo_assets(force_refresh=True) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal refresh logo company pasca-sinkron: {e}") if success and self.controller and self.user_info: try: set_bool("first_sync_pending", False) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal reset first_sync_pending: {e}") self.controller.show_dashboard(self.user_info) return # Kembali ke halaman beranda dashboard if hasattr(self, 'beranda_view') and self.beranda_view is not None: self.content_area.setCurrentWidget(self.beranda_view) self.log_debug("Kembali ke halaman beranda dashboard") def buka_menu_setting_printer(self): self.buka_pengaturan_printer() def buka_pengaturan_printer(self): if self._sync_blocked: QMessageBox.information( self, "Sinkronisasi Berjalan", "Pengaturan printer tidak dapat diakses saat sinkronisasi berjalan." ) return if self._enforce_settlement_navigation_lock(): return if not hasattr(self, "settings_page") or self.settings_page is None: view = SettingsView(self.settings_controller, self) view.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) view.scanner_settings_saved.connect(self._on_scanner_settings_saved) self.settings_page = view self.content_area.addWidget(self.settings_page) self.content_area.setCurrentWidget(self.settings_page) def _on_scanner_settings_saved(self, payload): if hasattr(self, "transaksi_controller") and self.transaksi_controller: if hasattr(self.transaksi_controller, "reload_scanner_detector_settings"): self.transaksi_controller.reload_scanner_detector_settings(payload) self.update_scanner_status() def logout(self): self.log_info('logout dipanggil') if self._sync_blocked: QMessageBox.information( self, "Sinkronisasi Berjalan", "Logout tidak dapat dilakukan saat sinkronisasi berjalan." ) return # edited by glg # Guard tambahan: auto/background sync juga harus selesai sebelum logout. if hasattr(self, "dashboard_controller") and self.dashboard_controller: try: if self.dashboard_controller.is_any_sync_running(): QMessageBox.information( self, "Sinkronisasi Berjalan", "Logout tidak dapat dilakukan saat sinkronisasi masih berjalan." ) return except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal cek status sinkronisasi saat logout: {e}") return if hasattr(self.controller, 'login_window'): self.controller.login_window.auto_login_attempted = False self.controller.show_login(reset_size=True) def _set_sync_blocked(self, blocked): self._sync_blocked = bool(blocked) if hasattr(self, "menu_buttons"): for btn in self.menu_buttons.values(): try: btn.setEnabled(not self._sync_blocked) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal update enable button: {e}") if hasattr(self, "user_menu_button"): try: self.user_menu_button.setEnabled(not self._sync_blocked) except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal update enable user menu: {e}") def closeEvent(self, event): """ CRITICAL FIX: Stop semua timer dan cleanup resources saat dashboard ditutup Ini adalah ROOT CAUSE dari freeze - timer masih berjalan setelah window close! """ self.log_info("Dashboard closing - stopping timers...") try: # 1. Stop connection check timer (30 detik interval) if hasattr(self, 'timer') and self.timer: if self.timer.isActive(): self.timer.stop() self.log_debug("Connection check timer stopped") self.timer.deleteLater() self.timer = None if self.waktu_timer: if self.waktu_timer.isActive(): self.waktu_timer.stop() self.log_debug("Waktu timer stopped") self.waktu_timer.deleteLater() self.waktu_timer = None if hasattr(self, 'export_json_timer') and self.export_json_timer: if self.export_json_timer.isActive(): self.export_json_timer.stop() self.log_debug("Export JSON timer stopped") self.export_json_timer.deleteLater() self.export_json_timer = None # 2. Stop waktu display timer (1 detik interval) # Cari semua child QTimer for child in self.findChildren(QTimer): if child and child.isActive(): child.stop() self.log_debug(f"Child timer stopped: {child.objectName() or 'unnamed'}") child.deleteLater() if hasattr(self, "scanner_controller") and self.scanner_controller: try: self.scanner_controller.dispose() except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Gagal stop scanner runtime: {e}") # 3. Block signals untuk prevent event firing during cleanup self.blockSignals(True) self.log_info("Dashboard cleanup completed") except (TypeError, ValueError, KeyError, AttributeError, RuntimeError, OSError, LookupError, ArithmeticError, ImportError) as e: self.log_warning(f"Error during dashboard cleanup: {e}") # Accept close event event.accept()