python launch editor externally and get texts

Question

I'm writing a PyQt programe where I'd like to allow the user to launch their preferred editor to fill in a TextEdit field.

So the goal is to launch an editor (say vim) externally on a tmp file, and upon editor closing, get its contexts into a python variable.

I've found a few similar questions like Opening vi from Python, call up an EDITOR (vim) from a python script, invoke an editor ( vim ) in python. But they are all in a "blocking" manner that works like the git commit command. What I am after is a "non-blocking" manner (because it is a GUI), something like the "Edit Source" function in zimwiki.

My current attempt:

import os
import tempfile
import threading
import subprocess

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        proc = subprocess.Popen(popenArgs)
        # this immediately finishes OPENING vim.
        rec=proc.wait()
        print('# <runInThread>: rec=', rec)
        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

def openEditor():

    fd, filepath=tempfile.mkstemp()
    print('filepath=',filepath)

    def cb(tmppath):
        print('# <cb>: cb tmppath=',tmppath)
        with open(tmppath, 'r') as tmp:
            lines=tmp.readlines()
            for ii in lines:
                print('# <cb>: ii',ii)
        return

    with os.fdopen(fd, 'w') as tmp:

        cmdflag='--'
        editor_cmd='vim'
        cmd=[os.environ['TERMCMD'], cmdflag, editor_cmd, filepath]
        print('#cmd = ',cmd)

        popenAndCall(cb, cmd)
        print('done')

    return


if __name__=='__main__':

    openEditor()

I think it failed because the Popen.wait() only waits until the editor is opened, not until its closing. So it captures nothing from the editor.

Any idea how to solve this? Thanks!

EDIT:

I found this answer which I guess is related. I'm messing around trying to let os wait for the process group, but it's still not working. Code below:

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        proc = subprocess.Popen(popenArgs, preexec_fn=os.setsid)
        pid=proc.pid
        gid=os.getpgid(pid)
        #rec=proc.wait()
        rec=os.waitid(os.P_PGID, gid, os.WEXITED | os.WSTOPPED)
        print('# <runInThread>: rec=', rec, 'pid=',pid, 'gid=',gid)

        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

I assume this gid=os.getpgid(pid) gives me the id of the group, and os.waitid() wait for the group. I also tried os.waitpid(gid, 0), didn't work either.

I'm on the right track?

UPDATE:

It seems that for some editors that works, like xed. vim and gvim both fails.

Answer 1

With QProcess you can launch a process without blocking the Qt event loop.

In this case I use xterm since I do not know which terminal is established in TERMCMD.

from PyQt5 import QtCore, QtGui, QtWidgets


class EditorWorker(QtCore.QObject):
    finished = QtCore.pyqtSignal()

    def __init__(self, command, parent=None):
        super(EditorWorker, self).__init__(parent)
        self._temp_file = QtCore.QTemporaryFile(self)
        self._process = QtCore.QProcess(self)
        self._process.finished.connect(self.on_finished)
        self._text = ""
        if self._temp_file.open():
            program, *arguments = command
            self._process.start(
                program, arguments + [self._temp_file.fileName()]
            )

    @QtCore.pyqtSlot()
    def on_finished(self):
        if self._temp_file.isOpen():
            self._text = self._temp_file.readAll().data().decode()
            self.finished.emit()

    @property
    def text(self):
        return self._text

    def __del__(self):
        self._process.kill()


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self._button = QtWidgets.QPushButton(
            "Launch VIM", clicked=self.on_clicked
        )
        self._text_edit = QtWidgets.QTextEdit(readOnly=True)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self._button)
        lay.addWidget(self._text_edit)

    @QtCore.pyqtSlot()
    def on_clicked(self):
        worker = EditorWorker("xterm -e vim".split(), self)
        worker.finished.connect(self.on_finished)

    @QtCore.pyqtSlot()
    def on_finished(self):
        worker = self.sender()
        prev_cursor = self._text_edit.textCursor()
        self._text_edit.moveCursor(QtGui.QTextCursor.End)
        self._text_edit.insertPlainText(worker.text)
        self._text_edit.setTextCursor(prev_cursor)
        worker.deleteLater()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

I guess in your case you should change

"xterm -e vim".split()

to

[os.environ['TERMCMD'], "--", "vim"]

Possible commands:

- xterm -e vim
- xfce4-terminal --disable-server -x vim

Update:

Implementing the same logic that you use with pyinotify that is to monitor the file, but in this case using QFileSystemWatcher which is a multiplatform solution:

from PyQt5 import QtCore, QtGui, QtWidgets


class EditorWorker(QtCore.QObject):
    finished = QtCore.pyqtSignal()

    def __init__(self, command, parent=None):
        super(EditorWorker, self).__init__(parent)
        self._temp_file = QtCore.QTemporaryFile(self)
        self._process = QtCore.QProcess(self)
        self._text = ""
        self._watcher = QtCore.QFileSystemWatcher(self)
        self._watcher.fileChanged.connect(self.on_fileChanged)

        if self._temp_file.open():
            self._watcher.addPath(self._temp_file.fileName())

            program, *arguments = command
            self._process.start(
                program, arguments + [self._temp_file.fileName()]
            )

    @QtCore.pyqtSlot()
    def on_fileChanged(self):
        if self._temp_file.isOpen():
            self._text = self._temp_file.readAll().data().decode()
            self.finished.emit()

    @property
    def text(self):
        return self._text

    def __del__(self):
        self._process.kill()


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self._button = QtWidgets.QPushButton(
            "Launch VIM", clicked=self.on_clicked
        )
        self._text_edit = QtWidgets.QTextEdit(readOnly=True)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self._button)
        lay.addWidget(self._text_edit)

    @QtCore.pyqtSlot()
    def on_clicked(self):
        worker = EditorWorker("gnome-terminal -- vim".split(), self)
        worker.finished.connect(self.on_finished)

    @QtCore.pyqtSlot()
    def on_finished(self):
        worker = self.sender()
        prev_cursor = self._text_edit.textCursor()
        self._text_edit.moveCursor(QtGui.QTextCursor.End)
        self._text_edit.insertPlainText(worker.text)
        self._text_edit.setTextCursor(prev_cursor)
        worker.deleteLater()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

Answer 2

The issue I reproduced is that proc is the gnome-terminal process and not the vim process.

Here are the two options that work for me.

1) Find the process of your text editor and not that of your terminal. With the right process ID, the code can wait for the process of your text editor to finish.

With psutil (portable)

Finds the latest editor process in the list of all running processes.

import psutil
def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        editor_cmd=popenArgs[-2]  # vim
        proc = subprocess.Popen(popenArgs)
        proc.wait()

        # Find the latest editor process in the list of all running processes
        editor_processes = []

        for p in psutil.process_iter():
            try:
                process_name = p.name()
                if editor_cmd in process_name:
                    editor_processes.append((process_name, p.pid))
            except:
                pass

        editor_proc = psutil.Process(editor_processes[-1][1])

        rec=editor_proc.wait()
        print('# <runInThread>: rec=', rec)
        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

Without psutil (works on Linux, but not portable to Mac OS or Windows)

Draws from https://stackoverflow.com/a/2704947/241866 and the source code of psutil.

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        editor_cmd=popenArgs[-2]  # vim
        proc = subprocess.Popen(popenArgs)
        proc.wait()

        # Find the latest editor process in the list of all running processes

        pids = [pid for pid in os.listdir('/proc') if pid.isdigit()]

        editor_processes = []
        for pid in pids:
            try:
                process_name = open(os.path.join('/proc', pid, 'cmdline'), 'rb').read().split('\0')[0]
                if editor_cmd in process_name:
                    editor_processes.append((process_name, int(pid)))
            except IOError:
                continue
        editor_proc_pid = editor_processes[-1][1]

        def pid_exists(pid):
            try:
                os.kill(pid, 0)
                return True
            except:
                return 

        while True:
            if pid_exists(editor_proc_pid):
                import time
                time.sleep(1)
            else:
                break

        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

2) As a last resort, you can catch a UI event before updating the text:

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        proc = subprocess.Popen(popenArgs)
        # this immediately finishes OPENING vim.
        rec=proc.wait()
        raw_input("Press Enter")  # replace this with UI event
        print('# <runInThread>: rec=', rec)
        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

Answer 3

I think @eyllanesc's solution is very close to what zim is doing (zim is using GObject.spawn_async() and GObject.child_watch_add(), I've no experiences with GObject, I guess that's the equivalent to QProcess.start()). But we run into some problems regarding how some terminals (like gnome-terminal) handles new terminal session launching.

I tried to monitor the temporary file opened by the editor, and on writing/saving the temp file I could call my callback. The monitoring is done using pyinotify. I've tried gnome-terminal, xterm, urxvt and plain gvim, all seem to work.

Code below:

import threading
from PyQt5 import QtCore, QtGui, QtWidgets
import pyinotify


class EditorWorker(QtCore.QObject):
    file_close_sig = QtCore.pyqtSignal()
    edit_done_sig = QtCore.pyqtSignal()

    def __init__(self, command, parent=None):
        super(EditorWorker, self).__init__(parent)
        self._temp_file = QtCore.QTemporaryFile(self)
        self._process = QtCore.QProcess(self)
        #self._process.finished.connect(self.on_file_close)
        self.file_close_sig.connect(self.on_file_close)
        self._text = ""
        if self._temp_file.open():
            program, *arguments = command
            self._process.start(
                program, arguments + [self._temp_file.fileName()]
            )
            tmpfile=self._temp_file.fileName()
            # start a thread to monitor file saving/closing
            self.monitor_thread = threading.Thread(target=self.monitorFile,
                    args=(tmpfile, self.file_close_sig))
            self.monitor_thread.start()

    @QtCore.pyqtSlot()
    def on_file_close(self):
        if self._temp_file.isOpen():
            print('open')
            self._text = self._temp_file.readAll().data().decode()
            self.edit_done_sig.emit()
        else:
            print('not open')

    @property
    def text(self):
        return self._text

    def __del__(self):
        try:
            self._process.kill()
        except:
            pass

    def monitorFile(self, path, sig):

        class PClose(pyinotify.ProcessEvent):
            def my_init(self):
                self.sig=sig
                self.done=False

            def process_IN_CLOSE(self, event):
                f = event.name and os.path.join(event.path, event.name) or event.path
                self.sig.emit()
                self.done=True

        wm = pyinotify.WatchManager()
        eventHandler=PClose()
        notifier = pyinotify.Notifier(wm, eventHandler)
        wm.add_watch(path, pyinotify.IN_CLOSE_WRITE)

        try:
            while not eventHandler.done:
                notifier.process_events()
                if notifier.check_events():
                    notifier.read_events()
        except KeyboardInterrupt:
            notifier.stop()
            return


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self._button = QtWidgets.QPushButton(
            "Launch VIM", clicked=self.on_clicked
        )
        self._text_edit = QtWidgets.QTextEdit(readOnly=True)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self._button)
        lay.addWidget(self._text_edit)

    @QtCore.pyqtSlot()
    def on_clicked(self):
        worker = EditorWorker(["gnome-terminal", '--', "vim"], self)
        worker.edit_done_sig.connect(self.on_edit_done)

    @QtCore.pyqtSlot()
    def on_edit_done(self):
        worker = self.sender()
        prev_cursor = self._text_edit.textCursor()
        self._text_edit.moveCursor(QtGui.QTextCursor.End)
        self._text_edit.insertPlainText(worker.text)
        self._text_edit.setTextCursor(prev_cursor)
        worker.deleteLater()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

BUT pyinotify only works in Linux. If you could find a cross-platform solution (at least on Mac) please let me know.

UPDATE: this doesn't seem to be robust. pyinotify reports file writing instead of just file closing. I'm depressed.