Line data    Source code

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