PyGObject

Question

我正在开发一个简单的 python GUI，用于使用 PyGObject/Gtk + Glade 来试验 gphoto2。我改编了 Gnome 教程，它使用生成器创建伪线程，允许将更新传递到主 Gtk 线程。子进程运行 gphoto2 并将命令传递给它以从 DSLR 相机拍摄一系列照片。我已经设法让程序在两次拍摄之间更新 GUI。但是，我计划进行更长的拍摄，实现取消生成器中间任务的选项会很有用。我将如何以最简单的方式实现它？

我在本[教程]中使用了示例：https://wiki.gnome.org/Projects/PyGObject/Threading

下面是正在编写的代码的一个简单示例。实际应用要大得多。

import gi
import re
import time
import subprocess

gi.require_version("Gtk", "3.0")
from gi.repository import Gio, GLib, Gtk, GObject


def app_main():

    builder = Gtk.Builder()
    builder.add_from_file("example.glade")

    window = builder.get_object("GUI")
    window.connect("destroy", Gtk.main_quit)

    capture_button = builder.get_object("btn_capture")

    def capture():
        cmd = ['gphoto2', '--auto-detect']

        process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
        output = process.stdout.read().decode('utf-8')
        process.wait()

        usb_devices = re.findall('usb:001,' + '[0-9][0-9][0-9]', output)
        working_directory = "/home/richard"

        for i in range(1, 5):
            new_port = '--port=' + usb_devices[0]
            cmd = ['gphoto2', new_port, '--capture-image', '--keep']
            process = subprocess.Popen(cmd, cwd=working_directory)
            process.wait()
            update_progress(f"Photo {i} of 5 taken!\n")

            yield True

        capture_button.set_sensitive(True)

    def update_progress(text):
        output = builder.get_object("txtview_console")
        textviewbuffer = output.get_buffer()
        start_iter = textviewbuffer.get_start_iter()
        textviewbuffer.insert(start_iter, text)
        return False

    def on_capture_clicked(button):
        capture_button.set_sensitive(False)
        time.sleep(1)
        run_generator(capture)

    def on_cancel_clicked(button):
        print("You pressed cancel!")
        # I would like this to cancel the generator some how

    def run_generator(function):
        gen = function()
        GLib.idle_add(lambda: next(gen, False), priority=GLib.PRIORITY_LOW)

    handlers = {
        "on_capture_clicked": on_capture_clicked,
        "on_cancel_clicked": on_cancel_clicked
    }

    builder.connect_signals(handlers)

    window.show_all()


if __name__ == "__main__":

    app_main()
    Gtk.main()

Answer 1

以下是有关如何停止线程的一些示例： https://www.google.at/amp/s/www.geeksforgeeks.org/python-different-ways-to-kill-a-thread/amp/

Answer 2

为了回答我自己的问题，我用如下示例解决了这个问题：

import os
import signal
import time

import gi

gi.require_version("Gtk", "3.0")
from gi.repository import GLib, Gtk, GObject
from multiprocessing import Pipe, Process

proc = 0


class GUI(Gtk.Window):

    def run_process(self, target, **kwargs):

        iterator = kwargs.get('iterator')
        option = kwargs.get('option')
        pid = kwargs.get('pid')

        parent_conn, child_conn = Pipe(duplex=False)

        if target == capture:
            Process(target=target, args=[child_conn, iterator], daemon=True).start()

        if target == manager:
            Process(target=target, args=[child_conn, option, pid], daemon=True).start()

        child_conn.close()

        def read_data(source, condition):
            assert parent_conn.poll()
            try:
                i = parent_conn.recv()
            except EOFError as E:
                print(E)
                return False

            if isinstance(i, list) and not isinstance(i[0], str):
                self.update_progress(f"Photo {i[0]} of {i[1]} taken!\n")

            elif isinstance(i, str):
                self.update_progress(i)

            elif isinstance(i, list) and i[0] == "PID":
                global proc
                proc = i[1]

            return True

        GLib.io_add_watch(parent_conn.fileno(), GObject.IO_IN, read_data)

    def update_progress(self, text):
        output = builder.get_object("txtview_console")
        textviewbuffer = output.get_buffer()
        start_iter = textviewbuffer.get_start_iter()
        textviewbuffer.insert(start_iter, text)
        return False

    def on_capture_clicked(self, button):
        btn_capture = builder.get_object("start_capture_button")
        btn_capture.set_sensitive(False)
        self.run_process(capture, iterator=[1, 11])

    def on_cancel_clicked(self, button):
        global proc

        btn_capture = builder.get_object("start_capture_button")
        btn_capture.set_sensitive(True)

        if proc == 0:
            self.update_progress("No capture running!\n")
        else:
            self.run_process(manager, option=1, pid=proc)
            proc = 0


def capture(c_conn, iterator):
    proc = os.getpid()
    c_conn.send(['PID', proc])

    for i in range(iterator[0], iterator[1]):
        # blocking method or subprocess
        time.sleep(2)
        c_conn.send([i, iterator[1]-1])

    c_conn.send(['PID', 0])


def manager(c_conn, option, pid):
    if pid == 0:
        c_conn.send("No capture running!\n")
        return False
    else:
        if option == 1:
            os.kill(pid, signal.SIGKILL)
            c_conn.send("Capture cancelled!\n")


if __name__ == "__main__":
    builder = Gtk.Builder()
    builder.add_from_file("example.glade")
    builder.connect_signals(GUI())

    win = builder.get_object("GUI")
    win.connect("delete-event", Gtk.main_quit)
    win.show_all()

    Gtk.main()

使用 glade 文件提供 GUI 如下：

<?xml version="1.0" encoding="UTF-8"?>
<!-- Generated with glade 3.22.1 -->
<interface>
  <requires lib="gtk+" version="3.20"/>
  <object class="GtkWindow" id="GUI">
    <property name="can_focus">False</property>
    <child>
      <placeholder/>
    </child>
    <child>
      <object class="GtkGrid">
        <property name="visible">True</property>
        <property name="can_focus">False</property>
        <property name="row_homogeneous">True</property>
        <property name="column_homogeneous">True</property>
        <child>
          <object class="GtkButton" id="start_capture_button">
            <property name="label" translatable="yes">Capture</property>
            <property name="visible">True</property>
            <property name="can_focus">True</property>
            <property name="receives_default">True</property>
            <signal name="clicked" handler="on_capture_clicked" swapped="no"/>
          </object>
          <packing>
            <property name="left_attach">0</property>
            <property name="top_attach">0</property>
          </packing>
        </child>
        <child>
          <object class="GtkButton" id="cancel_capture_button">
            <property name="label" translatable="yes">Cancel</property>
            <property name="visible">True</property>
            <property name="can_focus">True</property>
            <property name="receives_default">True</property>
            <signal name="clicked" handler="on_cancel_clicked" swapped="no"/>
          </object>
          <packing>
            <property name="left_attach">1</property>
            <property name="top_attach">0</property>
          </packing>
        </child>
        <child>
          <object class="GtkScrolledWindow">
            <property name="visible">True</property>
            <property name="can_focus">True</property>
            <property name="shadow_type">in</property>
            <child>
              <object class="GtkTextView" id="txtview_console">
                <property name="visible">True</property>
                <property name="can_focus">True</property>
              </object>
            </child>
          </object>
          <packing>
            <property name="left_attach">0</property>
            <property name="top_attach">1</property>
            <property name="width">2</property>
            <property name="height">2</property>
          </packing>
        </child>
      </object>
    </child>
  </object>
</interface>

上面的示例创建了一个 window，其中包含一个捕获按钮和一个取消按钮。如果捕获过程运行使捕获按钮的灵敏度变为 False 并且阻止过程运行使用管道中的进程提供反馈，并打印到文本视图中。在捕获过程中的任何时候，如果按下取消按钮，它将尝试为捕获进程获取全局 pid 号并将其终止。

PyGObject - 如何在执行期间中断 GLib 生成器函数？

PyGObject - How do I interrupt a GLib generator function during execution?

python

gtk

user-interface

generator