Python 和Webkit,看串口线程,如何避免core dump 运行 javascript

Python and Webkit, watching serial port thread, how to avoid core dump running javascript

我正在写一个使用 python 和 webkit 的游戏,网页是 front-end/GUI。 PC 连接到控制投币器和其他 i/o 的 Arduino。当 Arduino 通过串行发送 'coinin' 时,我在一个串行观察线程中捕获它,然后在网页上 运行 一些 javascript 到 'add' 游戏硬币。

为了简化故障排除,我设置了一个示例,运行是一个测试线程而不是读取串行,但问题是一样的。该线程尝试通过在网页上 运行ning 'addcoin()' 每秒添加一个硬币。如果取消注释 run_javascript() 行,程序核心转储。

我想出了一个键盘黑客解决方法。测试线程不是直接尝试 run_javascript(),而是对 xdotool 进行 os.system 调用以将字母 'conn' 键入程序 window。 window 有一个按键事件侦听器,当它在 keybuffer[] 中获取字母 'conn' 时,它就会 运行 对网页进行所需的 run_javascript() 调用.如果您将这两个文件复制到一个文件夹,并且 运行 程序 python,您将看到硬币文本每秒计数一次(按 BackSpace 键结束程序)。如果您尝试从线程中 运行 javascript,您将看到程序核心转储。

问题是,有没有更好的方法来做到这一点,而不必使用键盘破解 运行 javascript?虽然破解解决了这个问题,但它在游戏中引入了一个弱点。您可以通过在键盘上输入 'conn' 来骗取硬币。我想找到一些其他方式来触发事件,而不必使用键盘事件。

示例网页 index.htm

<html>
<script language="JavaScript"  type="text/javascript">
var mycoins=0;
document.onkeydown = function(evt) {
    evt = evt || window.event;
    cancelKeypress = (evt.ctrlKey && evt.keyCode == 84);
        return false;
};

function addcoin()
{
mycoins+=1;
id('mycoins').innerHTML="You Have "+mycoins.toString()+" coins"
}

function id(myID){return document.getElementById(myID)}
</script>
<html>
<body>
<div id=mycoins>You Have 0 Coins</div>
</body>
</html> 

样本python

#!/usr/bin/python
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gdk
gi.require_version('WebKit2', '4.0')
from gi.repository import WebKit2
import os,time,sys,threading,serial
defaultpath = os.path.dirname(os.path.realpath(__file__))
killthread=False
keybuffer=[]
buffkeys=['c','o','n','h','p','e']
myname=os.path.basename(__file__)
serial_ports=['/dev/ttyUSB0','/dev/ttyUSB1','/dev/ttyACM0','/dev/ttyACM1']
checkserial=True;

class  BrowserView:
    def __init__(self):
       global checkserial
       window = Gtk.Window()
       window.connect("key-press-event", self._key_pressed, window)
       self.view = WebKit2.WebView()              
       self.view.load_uri('file:///'+defaultpath+'/index.htm')
       self.view.connect("notify::title", self.window_title_change)

       window.add(self.view)
       window.fullscreen()
       window.show_all()
       '''
       ######not used for this example#######################################
       serial_port="" 
       for x in serial_ports:
          #print 'trying ',x
          if os.popen('ls '+x+' >/dev/null 2>&1 ; echo $?').read().strip()=='0':
             serial_port=x
             break;
       baud=9600
       if len(serial_port)>1:
          self.ser = serial.Serial(serial_port, baud, timeout=0)
       else:
          self.view.load_uri('file:///'+defaultpath+'/signDOWN.htm?Serial%20Port%20Error|Keno%20will%20auto%20close')
          checkserial=False;
       if checkserial:
          thread = threading.Thread(target=self.read_from_port)
          thread.start()
       ####################################################################### 
       '''
       #####thread test#############
       thread = threading.Thread(target=self.testthread)
       thread.start()        

    def testthread(self):
       while True:
          os.system('xdotool search --name '+myname+' type conn')
          #self.view.run_javascript('addcoin()')                        #causes core dump
          if killthread==True:
             break;
          time.sleep(1)

    def read_from_port(self):
       while True:
          if self.ser.inWaiting()>0:
             response=self.ser.readline()
             print(response)
             if 'coinin' in response:
                os.system('xdotool search --name '+myname+' type conn')
                #self.view.run_javascript('addcoin()')                  #causes core dump   

          if killthread==True:
             break;
          time.sleep(1)

    def checkbuffer(self):
       global keybuffer
       if 'conn' in ''.join(str(x) for x in keybuffer):
          self.view.run_javascript('addcoin()')
          keybuffer=[]

    def window_title_change(self, widget, param):     
       if not self.view.get_title():
          return
       os.chdir(defaultpath)  
       if self.view.get_title().startswith("pythondiag:::"):
          message = self.view.get_title().split(":::",1)[1]
          os.system('zenity --notification --text='+message+' --timeout=2')


    def _key_pressed(self, widget, event, window):
        global keybuffer
        mykey=Gdk.keyval_name(event.keyval)
        isakey=False
        for x in buffkeys:
           if mykey==x:
              isakey=True;
        if isakey:
           keybuffer.append(Gdk.keyval_name(event.keyval))
        else:
           keybuffer=[]
        self.checkbuffer()
        if mykey == 'BackSpace': 
           self.myquit()

    def myquit(self):
       global killthread
       killthread=True
       try:
          self.ser.write('clear\n')
       except:
          pass
       Gtk.main_quit()

if __name__ == "__main__":
    BrowserView()
    Gtk.main()

更新:此答案针对一般情况进行了更新,原始答案如下。

虽然 GIL 在给定时间只允许一个 python 线程处于 运行, 在上下文切换时我们对其他线程状态一无所知 (这就像在单核机器上执行多线程程序一样。) 这就是为什么你应该从它们 "belong" 到的线程调用任何 非 MT 安全方法(包括 GTK 调用,"belong" 到主事件循环)。

如果你想调用这样的函数,你应该安排它在主循环中执行。可能最简单的方法是使用 idle_add. Also note, that idle_add'ed function should return TrueFalse 是否应稍后再次调用。

您的代码应如下所示:

from gi.repository import GLib

...

class ThreadedWork:
  def function(self, arg):
    ''' function to be called in mainloop'''
    if arg:
      return GLib.SOURCE_REMOVE
    return GLib.SOURCE_CONTINUE

  def scheduler(self, function, arg):
    ''' scheduler (purely for readability issues) '''
    GLib.idle_add(function, arg)

  def thread_func(self):
    ''' long long thread function '''
    while True:
       # Do some long work
       # After it is done, schedule execution of mainloop functions.
       self.scheduler(self.function, True)
       time.sleep(1)


原回答: 看起来是由于 run_javascript 不是 MT 安全的(例如,不同于 this method)。

from gi.repository import GLib

...

class  BrowserView:
  def javascript_runner(self, script_name):
    GLib.idle_add(self.view.run_javascript, script_name)

  def testthread(self):
       while True:
          os.system('xdotool search --name '+myname+' type conn')
          # After long work is done, schedule execution of mainloop functions.
          self.javascript_runner('addcoin()') 
          if killthread: # btw, there is no need to check ==True explicitly
             break
          time.sleep(1)

我想 post 完整的测试代码,供任何正在寻找 运行 webkit 方法的人使用,同时 运行 使用线程从串行端口(或任何线程),然后用它做一些有用的事情。在问这个问题之前,我搜索了大约一周的解决方案,但找不到任何专门用于 webkit 的东西。

如果要使用串行部分,取消该部分的注释,并注释测试线程部分。如果您对如何使用它有任何疑问,请提问,我会尽力回答。

run_javascript('your_js_function()') 是 python 指示网页做某事的方式。

def window_title_change(self, widget, param): 函数是您从网页返回 python 的方式。你还必须有 'self.view.connect("notify::title", self.window_title_change)' 行在 BrowserView Class 中,如示例代码所示,因此 python 将检测到变化,并对其采取行动。

例如,在您的网页上,包含此功能:

    function python(x)
    {
        document.title=""
        document.title=x
    }

然后从您的网页调用 python 为您做一些事情,只需像这样调用 python:

    python('pythondiag:::'hello python');

在 python 方面,您可以编写您需要的任何函数,以执行与系统交互所需的任何操作。 Webkit 是一个很棒的解决方案,可以将 HTML 和 javascript 用作 front-end/GUI,然后通过 python.

与您的 PC 交互

感谢 Alexander Dmitriev 对原始问题的出色解决方案, 这是没有核心转储的完整代码...哇哦!我希望这可以帮助其他遇到此问题的人。

示例网页index.htm

<html>
<script language="JavaScript"  type="text/javascript">
var mycoins=0;
document.onkeydown = function(evt) {
    evt = evt || window.event;
    cancelKeypress = (evt.ctrlKey && evt.keyCode == 84);
        return false;
};

function addcoin()
{
mycoins+=1;
id('mycoins').innerHTML="You Have "+mycoins.toString()+" coins"
}

function id(myID){return document.getElementById(myID)}
</script>
<html>
<body>
<div id=mycoins>You Have 0 Coins</div>
</body>
</html> 

样本python

#!/usr/bin/python
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gdk, GLib
gi.require_version('WebKit2', '4.0')
from gi.repository import WebKit2
import os,time,sys,threading,serial
defaultpath = os.path.dirname(os.path.realpath(__file__))
killthread=False
myname=os.path.basename(__file__)
serial_ports=['/dev/ttyUSB0','/dev/ttyUSB1','/dev/ttyACM0','/dev/ttyACM1']
checkserial=True;

class  BrowserView:
    def __init__(self):
       global checkserial
       window = Gtk.Window()
       window.connect("key-press-event", self._key_pressed, window)
       self.view = WebKit2.WebView()              
       self.view.load_uri('file:///'+defaultpath+'/index.htm')
       self.view.connect("notify::title", self.window_title_change)

       window.add(self.view)
       window.fullscreen()
       window.show_all()
       '''
       ######Uncomment this to use the serial port watcher#####################
       serial_port="" 
       for x in serial_ports:
          #print 'trying ',x
          if os.popen('ls '+x+' >/dev/null 2>&1 ; echo $?').read().strip()=='0':
             serial_port=x
             break;
       baud=9600
       if len(serial_port)>1:
          self.ser = serial.Serial(serial_port, baud, timeout=0)
       else:
          self.view.load_uri('file:///'+defaultpath+'/signDOWN.htm?Serial%20Port%20Error|Keno%20will%20auto%20close')
          checkserial=False;
       if checkserial:
          thread = threading.Thread(target=self.read_from_port)
          thread.start()
       ######################################################################### 
       '''
       #####thread test--comment out to use the serial port watcher#############
       thread = threading.Thread(target=self.testthread)
       thread.start()
       #########################################################################

    def javascript_runner(self, script_name):
       GLib.idle_add(self.view.run_javascript, script_name)

    def testthread(self):
       while True:
          self.javascript_runner('addcoin()')
          if killthread: 
             break
          time.sleep(1)

    def read_from_port(self):
       while True:
          if self.ser.inWaiting()>0:
             response=self.ser.readline()
             print(response)
             if 'coinin' in response:
                self.javascript_runner('addcoin()')                    

          if killthread:
             break;
          time.sleep(1)

    def window_title_change(self, widget, param):     
       if not self.view.get_title():
          return
       os.chdir(defaultpath)  
       if self.view.get_title().startswith("pythondiag:::"):
          message = self.view.get_title().split(":::",1)[1]
          os.system('zenity --notification --text='+message+' --timeout=2')

    def _key_pressed(self, widget, event, window):
       mykey=Gdk.keyval_name(event.keyval)
       print mykey
       if mykey == 'BackSpace': 
          self.myquit()

    def myquit(self):
       global killthread
       killthread=True
       try:
          self.ser.write('clear\n')
       except:
          pass
       Gtk.main_quit()

if __name__ == "__main__":
    BrowserView()
    Gtk.main()