Line data    Source code

       1             : # SPDX-FileCopyrightText: 2025 Pairinteraction Developers
       2             : # SPDX-License-Identifier: LGPL-3.0-or-later
       3             : 
       4           1 : import logging
       5           1 : import os
       6           1 : import signal
       7           1 : from types import FrameType
       8           1 : from typing import Optional
       9             : 
      10           1 : from PySide6.QtCore import QObject, QSocketNotifier, QTimer, Signal
      11           1 : from PySide6.QtWidgets import QApplication
      12             : 
      13           1 : from pairinteraction_gui.worker import MultiProcessWorker, MultiThreadWorker
      14             : 
      15           1 : logger = logging.getLogger(__name__)
      16             : 
      17             : 
      18           1 : class MainSignals(QObject):
      19             :     """Signals for the application.
      20             : 
      21             :     We store an instance of this signal class in the Application instance, see app.py.
      22             :     So to access these signals (from anywhere in the application), you can use
      23             :     `Application.instance().signals`.
      24             :     """
      25             : 
      26           1 :     ask_download_database = Signal(str)
      27             : 
      28             : 
      29           1 : class Application(QApplication):
      30             :     """Add some global signals to the QApplication."""
      31             : 
      32           1 :     signals = MainSignals()
      33             : 
      34           1 :     @classmethod
      35           1 :     def instance(cls) -> "Application":
      36             :         """Return the current instance of the application."""
      37           1 :         return super().instance()  # type: ignore [return-value]
      38             : 
      39           1 :     def allow_ctrl_c(self) -> None:
      40             :         # Create a pipe to communicate between the signal handler and the Qt event loop
      41           0 :         pipe_r, pipe_w = os.pipe()
      42             : 
      43           0 :         def signal_handler(signal: int, frame: Optional[FrameType]) -> None:
      44           0 :             os.write(pipe_w, b"x")  # Write a single byte to the pipe
      45             : 
      46           0 :         signal.signal(signal.SIGINT, signal_handler)
      47             : 
      48           0 :         def handle_signal() -> None:
      49           0 :             os.read(pipe_r, 1)  # Read the byte from the pipe to clear it
      50           0 :             logger.info("Ctrl+C detected in terminal. Shutting down gracefully...")
      51           0 :             self.quit()
      52             : 
      53           0 :         sn = QSocketNotifier(pipe_r, QSocketNotifier.Type.Read, parent=self)
      54           0 :         sn.activated.connect(handle_signal)
      55             : 
      56             :         # Create a timer to ensure the event loop processes events regularly
      57             :         # This makes Ctrl+C work even when the application is idle
      58           0 :         timer = QTimer(self)
      59           0 :         timer.timeout.connect(lambda: None)  # Do nothing, just wake up the event loop
      60           0 :         timer.start(200)
      61             : 
      62           1 :     @staticmethod
      63           1 :     def quit() -> None:
      64             :         """Quit the application."""
      65           1 :         logger.debug("Calling Application.quit().")
      66           1 :         MultiProcessWorker.terminate_all(create_new_pool=False)
      67           1 :         MultiThreadWorker.terminate_all()
      68           1 :         QApplication.quit()
      69           1 :         logger.debug("Application.quit() done.")