Line data    Source code

       1             : # SPDX-FileCopyrightText: 2025 Pairinteraction Developers
       2             : # SPDX-License-Identifier: LGPL-3.0-or-later
       3             : 
       4           0 : import logging
       5           0 : import os
       6           0 : import signal
       7           0 : from multiprocessing import Process
       8           0 : from types import FrameType
       9           0 : from typing import ClassVar, Optional
      10             : 
      11           0 : from PySide6.QtCore import QObject, QSocketNotifier, QThread, QTimer, Signal
      12           0 : from PySide6.QtWidgets import QApplication
      13             : 
      14           0 : logger = logging.getLogger(__name__)
      15             : 
      16             : 
      17           0 : class MainSignals(QObject):
      18             :     """Signals for the application.
      19             : 
      20             :     We store an instance of this signal class in the Application instance, see app.py.
      21             :     So to access these signals (from anywhere in the application), you can use
      22             :     `Application.instance().signals`.
      23             :     """
      24             : 
      25           0 :     ask_download_database = Signal(str)
      26             : 
      27             : 
      28           0 : class Application(QApplication):
      29             :     """Add some global signals to the QApplication."""
      30             : 
      31           0 :     signals = MainSignals()
      32           0 :     all_processes: ClassVar[set[Process]] = set()
      33           0 :     all_threads: ClassVar[set[QThread]] = set()
      34             : 
      35           0 :     @classmethod
      36           0 :     def instance(cls) -> "Application":  # type: ignore  # overwrite type hints
      37             :         """Return the current instance of the application."""
      38           0 :         return super().instance()  # type: ignore [return-value]
      39             : 
      40           0 :     def allow_ctrl_c(self) -> None:
      41             :         # Create a pipe to communicate between the signal handler and the Qt event loop
      42           0 :         pipe_r, pipe_w = os.pipe()
      43             : 
      44           0 :         def signal_handler(signal: int, frame: Optional[FrameType]) -> None:
      45           0 :             os.write(pipe_w, b"x")  # Write a single byte to the pipe
      46             : 
      47           0 :         signal.signal(signal.SIGINT, signal_handler)
      48             : 
      49           0 :         def handle_signal() -> None:
      50           0 :             os.read(pipe_r, 1)  # Read the byte from the pipe to clear it
      51           0 :             logger.info("Ctrl+C detected in terminal. Shutting down gracefully...")
      52           0 :             self.quit()
      53             : 
      54           0 :         sn = QSocketNotifier(pipe_r, QSocketNotifier.Type.Read, parent=self)
      55           0 :         sn.activated.connect(handle_signal)
      56             : 
      57             :         # Create a timer to ensure the event loop processes events regularly
      58             :         # This makes Ctrl+C work even when the application is idle
      59           0 :         timer = QTimer(self)
      60           0 :         timer.timeout.connect(lambda: None)  # Do nothing, just wake up the event loop
      61           0 :         timer.start(200)
      62             : 
      63           0 :     @staticmethod
      64           0 :     def quit() -> None:
      65             :         """Quit the application."""
      66           0 :         logger.debug("Calling Application.quit().")
      67           0 :         Application.terminate_all_processes()
      68           0 :         Application.terminate_all_threads()
      69           0 :         QApplication.quit()
      70           0 :         logger.debug("Application.quit() done.")
      71             : 
      72           0 :     @staticmethod
      73           0 :     def terminate_all_processes() -> None:
      74             :         """Terminate all processes started by the application."""
      75           0 :         for process in Application.all_processes:
      76           0 :             if process.is_alive():
      77           0 :                 logger.debug("Terminating process %s.", process.pid)
      78           0 :                 process.terminate()
      79           0 :                 process.join(timeout=1)
      80             : 
      81           0 :         Application.all_processes.clear()
      82           0 :         logger.debug("All processes terminated.")
      83             : 
      84           0 :     @staticmethod
      85           0 :     def terminate_all_threads() -> None:
      86             :         """Terminate all threads started by the application."""
      87           0 :         for thread in Application.all_threads:
      88           0 :             if thread.isRunning():
      89           0 :                 logger.debug("Terminating thread %s.", thread)
      90           0 :                 thread.terminate()
      91           0 :                 thread.wait()
      92             : 
      93           0 :         Application.all_threads.clear()
      94           0 :         logger.debug("All threads terminated.")