注意:我找到了一种无需 QApplication
即可创建类似问题的方法;但是,这可能是相同的问题或不同的问题。 Reported on pystray repo。
注意:部分答案是问题是由竞争条件引起的(从 time.sleep
“修复”问题可以明显看出,
并缓存 multiprocessing
可加快导入速度);然而,剩下的问题(为什么 QApplication
可用时图标不显示)仍未解决。
要可靠地重现该问题,请将第 (3)
行移至第 (4)
行之前。
鉴于此代码:
#import multiprocessing # (1)
def g():
def f():
print("Function `f` is run")
from PIL import Image
from pystray import Icon
icon=Icon("Plover")
icon.icon=Image.new("RGB", (10, 10), (255, 255, 255))
icon.visible=True
icon.run()
import multiprocessing
p=multiprocessing.Process(target=f)
p.start()
p.join()
if __name__=="__main__":
from PyQt5.QtWidgets import QApplication
from threading import Thread
import time
t=Thread(target=g)
t.start() # (4)
app=QApplication([]) # (3)
t.join()
这段代码应该创建一个线程(运行函数 g
),并创建一个子进程(运行函数 f
),打印 Function `f` is run
,并创建一个系统托盘图标(包含一个白色方块)在其中。
然而,代码只打印 Function `f` is run
,而不创建图标。 (图标不可见,没有错误信息打印到控制台)
为什么会发生这种情况?
python --version
-> Python 3.9.2
。pip list
-> pystray 0.17.2
和 PyQt5 5.15.3
。PYSTRAY_BACKEND=appindicator
强制并且行为保持不变。 (我在 GNU/Linux 上使用 X)如果我取消注释第 1 行(或者在构造 multiprocessing
之前首次导入 QApplication
),则错误不会发生。 (2)
(注意:由上面的竞争条件解释解决)
如果我在程序的第一行添加 import random
,错误也不会发生。
我检查了这两种情况下的环境变量是否相同,方法是将其添加到函数 time.sleep(1)
中的打印语句之后:
QApplication
我可以在我的机器上可靠地重现该问题。
多进程启动是必要的,因为函数f
不在主线程上运行,pystray必须在主线程上运行。否则(如果启动子进程的代码被import os, hashlib
print(hashlib.sha1(str(os.environ).encode('u8')).hexdigest())
代替),pystray会抛出这个错误
g
请注意,pystray确实在以 f()
开头时(如上面的第一个代码块中一样)可以工作,只要 File "/path/to/file.py", line 17, in f
icon.run()
File "/path/to/python3.9/site-packages/pystray/_base.py", line 192, in run
self._run()
File "/path/to/python3.9/site-packages/pystray/_util/gtk.py", line 68, in _run
signal.signal(signal.SIGINT, signal.SIG_DFL)
File "/path/to/python3.9/signal.py", line 47, in signal
handler = _signal.signal(_enum_to_int(signalnum), _enum_to_int(handler))
ValueError: signal only works in main thread of the main interpreter
对象是稍后创建的或 multiprocessing
更快地导入(如上面 QApplication
中所述)
作为一种解决方法,可以使用 multiprocessing
来生成一个独立的 Python 进程(尽管这可能是特定于平台的)。此代码确实显示了图标:
(2)
所以我更想知道原始代码中 subprocess
的问题是什么,而不是找到有效的方法。
这是一个成功运行的跟踪日志:(通过添加获得
def g():
import tempfile, sys, subprocess
with tempfile.NamedTemporaryFile() as f:
f.write(r"""
print("Function `f` is run")
import os, hashlib
print(hashlib.sha1(str(os.environ).encode('u8')).hexdigest())
from PIL import Image
from pystray import Icon
icon=Icon("Plover")
icon.icon=Image.new("RGB", (10, 10), (255, 255, 255))
icon.visible=True
icon.run()
""".encode('u8'))
f.flush()
subprocess.run([sys.executable, f.name])
围绕要执行的代码):
multiprocessing
这是失败运行的跟踪日志:
import trace
trace.Trace(trace=True).run("""
...
""")
很明显,由于某些原因,[...]
--- modulename: __init__, funcname: __getattr__
__init__.py(32): return getattr(self._introspection_module, name)
_ossighelper.py(113): (GLib.IOCondition.IN | GLib.IOCondition.HUP |
_ossighelper.py(114): GLib.IOCondition.NVAL | GLib.IOCondition.ERR),
--- modulename: __init__, funcname: __getattr__
__init__.py(32): return getattr(self._introspection_module, name)
_ossighelper.py(113): (GLib.IOCondition.IN | GLib.IOCondition.HUP |
_ossighelper.py(115): signal_notify)
_ossighelper.py(110): source_id = GLib.io_add_watch(
--- modulename: GLib, funcname: io_add_watch
GLib.py(686): channel, priority, condition, func, user_data = _io_add_watch_get_args(*args, **kwargs)
--- modulename: GLib, funcname: _io_add_watch_get_args
GLib.py(639): if not isinstance(priority_, int) or isinstance(priority_, GLib.IOCondition):
GLib.py(657): if len(cb_and_user_data) < 1 or not callable(cb_and_user_data[0]):
GLib.py(659): callback = cb_and_user_data[0]
GLib.py(660): user_data = cb_and_user_data[1:]
GLib.py(663): if isinstance(channel, int):
GLib.py(666): elif isinstance(channel, socket.socket) and sys.platform == 'win32':
GLib.py(669): elif hasattr(channel, 'fileno'):
GLib.py(674): assert isinstance(channel, GLib.IOChannel)
GLib.py(675): func_fdtransform = callback
GLib.py(676): real_channel = channel
GLib.py(678): return real_channel, priority_, condition, func_fdtransform, user_data
GLib.py(687): return GLib.io_add_watch(channel, priority, condition, func, *user_data)
--- modulename: module, funcname: __getattr__
module.py(121): info = repository.find_by_name(self._namespace, name)
module.py(122): if not info:
module.py(126): if isinstance(info, EnumInfo):
module.py(162): elif isinstance(info, RegisteredTypeInfo):
module.py(213): elif isinstance(info, FunctionInfo):
module.py(214): wrapper = info
module.py(223): self.__dict__[name] = wrapper
module.py(224): return wrapper
--- modulename: __init__, funcname: __getattr__
__init__.py(32): return getattr(self._introspection_module, name)
--- modulename: __init__, funcname: __getattr__
__init__.py(32): return getattr(self._introspection_module, name)
_ossighelper.py(116): try:
_ossighelper.py(117): yield
GLib.py(497): super(MainLoop, self).run()
--- modulename: gtk, funcname: callback
gtk.py(44): try:
gtk.py(45): f(*args, **kwargs)
--- modulename: _appindicator, funcname: _update_icon
_appindicator.py(60): self._remove_fs_icon()
--- modulename: gtk, funcname: _remove_fs_icon
gtk.py(143): try:
gtk.py(144): if self._icon_path:
gtk.py(149): self._icon_valid = False
_appindicator.py(61): self._update_fs_icon()
--- modulename: gtk, funcname: _update_fs_icon
gtk.py(159): self._icon_path = tempfile.mktemp()
--- modulename: tempfile, funcname: mktemp
tempfile.py(393): if dir is None:
tempfile.py(394): dir = gettempdir()
--- modulename: tempfile, funcname: gettempdir
tempfile.py(283): if tempdir is None:
tempfile.py(290): return tempdir
tempfile.py(396): names = _get_candidate_names()
--- modulename: tempfile, funcname: _get_candidate_names
函数没有被调用。
pystray中的相关代码是这样的([...]
GLib.py(639): if not isinstance(priority_, int) or isinstance(priority_, GLib.IOCondition):
GLib.py(657): if len(cb_and_user_data) < 1 or not callable(cb_and_user_data[0]):
GLib.py(659): callback = cb_and_user_data[0]
GLib.py(660): user_data = cb_and_user_data[1:]
GLib.py(663): if isinstance(channel, int):
GLib.py(666): elif isinstance(channel, socket.socket) and sys.platform == 'win32':
GLib.py(669): elif hasattr(channel, 'fileno'):
GLib.py(674): assert isinstance(channel, GLib.IOChannel)
GLib.py(675): func_fdtransform = callback
GLib.py(676): real_channel = channel
GLib.py(678): return real_channel, priority_, condition, func_fdtransform, user_data
GLib.py(687): return GLib.io_add_watch(channel, priority, condition, func, *user_data)
--- modulename: module, funcname: __getattr__
module.py(121): info = repository.find_by_name(self._namespace, name)
module.py(122): if not info:
module.py(126): if isinstance(info, EnumInfo):
module.py(162): elif isinstance(info, RegisteredTypeInfo):
module.py(213): elif isinstance(info, FunctionInfo):
module.py(214): wrapper = info
module.py(223): self.__dict__[name] = wrapper
module.py(224): return wrapper
--- modulename: __init__, funcname: __getattr__
__init__.py(32): return getattr(self._introspection_module, name)
--- modulename: __init__, funcname: __getattr__
__init__.py(32): return getattr(self._introspection_module, name)
_ossighelper.py(116): try:
_ossighelper.py(117): yield
GLib.py(497): super(MainLoop, self).run()
[stuck until manual interrupt]
)
callback
我正在为一个程序编写一个插件,该插件在一个单独的(非主)线程上执行我的代码。我希望插件能够控制托盘图标。
因此,在主机程序创建 _util/gtk.py
之前,我无法在主线程上运行任何代码或导入 def callback(*args, **kwargs):
"""A callback that executes ``f`` and then returns ``False``.
"""
try:
f(*args, **kwargs)
finally:
return False
# Execute the callback as an idle function
GObject.idle_add(callback, *args, **kwargs)
模块。
因为代码是在单独的线程上执行的,所以我没有想出任何简单的方法来正确使用multiprocessing
,所以我查看了外部包QApplication
,但后来遇到了这种奇怪的行为。< /p>