PyQt - how to detect and close UI if it's alre

Question

I'm starting the UI from within Maya. If the UI hasn't been closed, running the UI again will completely freeze Maya (with the error "Event Loop is already running")

Manually closing the UI before re-running the script will prevent it from freezing up. But I guess that's not really practical.

Is there a way to detect if the UI I'm trying to run already exists? And possible force close it?

Answer 1

My solution is this:

import sys

from PyQt5.QtCore import QLockFile
from PyQt5.QtWidgets import QApplication
from PyQt5.QtWidgets import QMessageBox

from window import MainWindow


if __name__ == "__main__":
    try:
        app_object = QApplication(sys.argv)
        lock_file = QLockFile("app.lock")

        if lock_file.tryLock():
            window = MainWindow()
            window.show()

            app_object.exec()
        else:
            error_message = QMessageBox()
            error_message.setIcon(QMessageBox.Warning)
            error_message.setWindowTitle("Error")
            error_message.setText("The application is already running!")
            error_message.setStandardButtons(QMessageBox.Ok)
            error_message.exec()
    finally:
        lock_file.unlock()

Answer 2

In case if someone want to run @ekhumoro solution with python3 there's need to make few adjustments to string operations, i'll share my copy where it was working python 3.

import sys

from PyQt4 import QtGui, QtCore, QtNetwork

class SingleApplication(QtGui.QApplication):
    def __init__(self, argv, key):
        QtGui.QApplication.__init__(self, argv)
        self._memory = QtCore.QSharedMemory(self)
        self._memory.setKey(key)
        if self._memory.attach():
            self._running = True
        else:
            self._running = False
            if not self._memory.create(1):
                raise RuntimeError( self._memory.errorString() )

    def isRunning(self):
        return self._running

class SingleApplicationWithMessaging(SingleApplication):
    def __init__(self, argv, key):
        SingleApplication.__init__(self, argv, key)
        self._key = key
        self._timeout = 1000
        self._server = QtNetwork.QLocalServer(self)

        if not self.isRunning():
            self._server.newConnection.connect(self.handleMessage)
            self._server.listen(self._key)

    def handleMessage(self):
        socket = self._server.nextPendingConnection()
        if socket.waitForReadyRead(self._timeout):
            self.emit(QtCore.SIGNAL('messageAvailable'), bytes(socket.readAll().data()).decode('utf-8') )
            socket.disconnectFromServer()
        else:
            QtCore.qDebug(socket.errorString())

    def sendMessage(self, message):
        if self.isRunning():
            socket = QtNetwork.QLocalSocket(self)
            socket.connectToServer(self._key, QtCore.QIODevice.WriteOnly)
            if not socket.waitForConnected(self._timeout):
                print(socket.errorString())
                return False
            socket.write(str(message).encode('utf-8'))
            if not socket.waitForBytesWritten(self._timeout):
                print(socket.errorString())
                return False
            socket.disconnectFromServer()
            return True
        return False

class Window(QtGui.QWidget):
    def __init__(self):
        QtGui.QWidget.__init__(self)
        self.edit = QtGui.QLineEdit(self)
        self.edit.setMinimumWidth(300)
        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.edit)

    def handleMessage(self, message):
        self.edit.setText(message)

if __name__ == '__main__':

    key = 'foobar'

    # if parameter no. 1 was set then we'll use messaging between app instances
    if len(sys.argv) > 1:
        app = SingleApplicationWithMessaging(sys.argv, key)
        if app.isRunning():
            msg = ''
            # checking if custom message was passed as cli argument
            if len(sys.argv) > 2:
                msg = sys.argv[2]
            else:
                msg = 'APP ALREADY RUNNING'
            app.sendMessage( msg )
            print( "app is already running, sent following message: \n\"{0}\"".format( msg ) )
            sys.exit(1)
    else:
        app = SingleApplication(sys.argv, key)
        if app.isRunning():
            print('app is already running, no message has been sent')
            sys.exit(1)

    window = Window()
    app.connect(app, QtCore.SIGNAL('messageAvailable'), window.handleMessage)
    window.show()

    sys.exit(app.exec_())

Example cli calls, assuming that your script name is "SingleInstanceApp.py":

python SingleInstanceApp.py 1
python SingleInstanceApp.py 1 "test"
python SingleInstanceApp.py 1 "foo bar baz"
python SingleInstanceApp.py 1 "utf8 test FOO ßÄÖÜ ßäöü łąćźżóń ŁĄĆŹŻÓŃ etc"

(and here is call wihout first parameter, so message simply will not be sent)

python SingleInstanceApp.py

Hope that it will help someone.

Answer 3

Here is a very simple PyQt5 solution using QLockFile:

from PyQt5 import QtCore, QtWidgets

lockfile = QtCore.QLockFile(QtCore.QDir.tempPath() + '/my_app_name.lock')

if lockfile.tryLock(100):
    app = QtWidgets.QApplication([])
    win = QtWidgets.QWidget()
    win.setGeometry(50, 50, 100, 100)
    win.show()
    app.exec()
else:
    print('app is already running')

---

There were a couple of fairly straightforward C++ solutions given on the Qt Wiki which no longer seem to exist. I ported one of them to PyQt, and have provided a sample script below. The original C++ solution has been split into two classes, because the messaging facility may not be needed.

PyQt5:

from PyQt5 import QtWidgets, QtCore, QtNetwork

class SingleApplication(QtWidgets.QApplication):
    messageAvailable = QtCore.pyqtSignal(object)

    def __init__(self, argv, key):
        super().__init__(argv)
        # cleanup (only needed for unix)
        QtCore.QSharedMemory(key).attach()
        self._memory = QtCore.QSharedMemory(self)
        self._memory.setKey(key)
        if self._memory.attach():
            self._running = True
        else:
            self._running = False
            if not self._memory.create(1):
                raise RuntimeError(self._memory.errorString())

    def isRunning(self):
        return self._running

class SingleApplicationWithMessaging(SingleApplication):
    def __init__(self, argv, key):
        super().__init__(argv, key)
        self._key = key
        self._timeout = 1000
        self._server = QtNetwork.QLocalServer(self)
        if not self.isRunning():
            self._server.newConnection.connect(self.handleMessage)
            self._server.listen(self._key)

    def handleMessage(self):
        socket = self._server.nextPendingConnection()
        if socket.waitForReadyRead(self._timeout):
            self.messageAvailable.emit(
                socket.readAll().data().decode('utf-8'))
            socket.disconnectFromServer()
        else:
            QtCore.qDebug(socket.errorString())

    def sendMessage(self, message):
        if self.isRunning():
            socket = QtNetwork.QLocalSocket(self)
            socket.connectToServer(self._key, QtCore.QIODevice.WriteOnly)
            if not socket.waitForConnected(self._timeout):
                print(socket.errorString())
                return False
            if not isinstance(message, bytes):
                message = message.encode('utf-8')
            socket.write(message)
            if not socket.waitForBytesWritten(self._timeout):
                print(socket.errorString())
                return False
            socket.disconnectFromServer()
            return True
        return False

class Window(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.edit = QtWidgets.QLineEdit(self)
        self.edit.setMinimumWidth(300)
        layout = QtWidgets.QVBoxLayout(self)
        layout.addWidget(self.edit)

    def handleMessage(self, message):
        self.edit.setText(message)

if __name__ == '__main__':

    import sys

    key = 'app-name'

    # send commandline args as message
    if len(sys.argv) > 1:
        app = SingleApplicationWithMessaging(sys.argv, key)
        if app.isRunning():
            print('app is already running')
            app.sendMessage(' '.join(sys.argv[1:]))
            sys.exit(1)
    else:
        app = SingleApplication(sys.argv, key)
        if app.isRunning():
            print('app is already running')
            sys.exit(1)

    window = Window()
    app.messageAvailable.connect(window.handleMessage)
    window.show()

    sys.exit(app.exec_())

PyQt4:

# only needed for python2
import sip
sip.setapi('QString', 2)

from PyQt4 import QtGui, QtCore, QtNetwork

class SingleApplication(QtGui.QApplication):
    messageAvailable = QtCore.pyqtSignal(object)

    def __init__(self, argv, key):
        QtGui.QApplication.__init__(self, argv)
        # cleanup (only needed for unix)
        QtCore.QSharedMemory(key).attach()
        self._memory = QtCore.QSharedMemory(self)
        self._memory.setKey(key)
        if self._memory.attach():
            self._running = True
        else:
            self._running = False
            if not self._memory.create(1):
                raise RuntimeError(self._memory.errorString())

    def isRunning(self):
        return self._running

class SingleApplicationWithMessaging(SingleApplication):
    def __init__(self, argv, key):
        SingleApplication.__init__(self, argv, key)
        self._key = key
        self._timeout = 1000
        self._server = QtNetwork.QLocalServer(self)
        if not self.isRunning():
            self._server.newConnection.connect(self.handleMessage)
            self._server.listen(self._key)

    def handleMessage(self):
        socket = self._server.nextPendingConnection()
        if socket.waitForReadyRead(self._timeout):
            self.messageAvailable.emit(
                socket.readAll().data().decode('utf-8'))
            socket.disconnectFromServer()
        else:
            QtCore.qDebug(socket.errorString())

    def sendMessage(self, message):
        if self.isRunning():
            socket = QtNetwork.QLocalSocket(self)
            socket.connectToServer(self._key, QtCore.QIODevice.WriteOnly)
            if not socket.waitForConnected(self._timeout):
                print(socket.errorString())
                return False
            if not isinstance(message, bytes):
                message = message.encode('utf-8')
            socket.write(message)
            if not socket.waitForBytesWritten(self._timeout):
                print(socket.errorString())
                return False
            socket.disconnectFromServer()
            return True
        return False

class Window(QtGui.QWidget):
    def __init__(self):
        QtGui.QWidget.__init__(self)
        self.edit = QtGui.QLineEdit(self)
        self.edit.setMinimumWidth(300)
        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.edit)

    def handleMessage(self, message):
        self.edit.setText(message)

if __name__ == '__main__':

    import sys

    key = 'app-name'

    # send commandline args as message
    if len(sys.argv) > 1:
        app = SingleApplicationWithMessaging(sys.argv, key)
        if app.isRunning():
            print('app is already running')
            app.sendMessage(' '.join(sys.argv[1:]))
            sys.exit(1)
    else:
        app = SingleApplication(sys.argv, key)
        if app.isRunning():
            print('app is already running')
            sys.exit(1)

    window = Window()
    app.messageAvailable.connect(window.handleMessage)
    window.show()

    sys.exit(app.exec_())