有一个解决方案是使用包装器(如 Vinay 提供的链接中所述),该包装器在新的控制台窗口中使用 Windows start 命令启动。
包装代码:
#wrapper.py
import subprocess, time, signal, sys, os
def signal_handler(signal, frame):
time.sleep(1)
print 'Ctrl+C received in wrapper.py'
signal.signal(signal.SIGINT, signal_handler)
print "wrapper.py started"
subprocess.Popen("python demo.py")
time.sleep(3) #Replace with your IPC code here, which waits on a fire CTRL-C request
os.kill(signal.CTRL_C_EVENT, 0)
捕捉 CTRL-C 的程序代码:
#demo.py
import signal, sys, time
def signal_handler(signal, frame):
print 'Ctrl+C received in demo.py'
time.sleep(1)
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
print 'demo.py started'
#signal.pause() # does not work under Windows
while(True):
time.sleep(1)
启动包装器,例如:
PythonPrompt> import subprocess
PythonPrompt> subprocess.Popen("start python wrapper.py", shell=True)
您需要添加一些 IPC 代码,以便控制包装器触发 os.kill(signal.CTRL_C_EVENT, 0) 命令。为此,我在我的应用程序中使用了套接字。
说明:
预告
-
send_signal(CTRL_C_EVENT) 不起作用,因为 CTRL_C_EVENT 仅适用于 os.kill。 [REF1]
-
os.kill(CTRL_C_EVENT)将信号发送给当前cmd窗口中运行的所有进程[REF2]
-
Popen(..., creationflags=CREATE_NEW_PROCESS_GROUP) 不起作用,因为CTRL_C_EVENT 被进程组忽略。 [REF2]
这是python文档[REF3]中的一个错误
实施的解决方案
- 使用 Windows shell 命令start,让您的程序在不同的 cmd 窗口中运行。
- 在您的控制应用程序和应该获得 CTRL-C 信号的应用程序之间添加一个 CTRL-C 请求包装器。包装器将在与应获得 CTRL-C 信号的应用程序相同的 cmd 窗口中运行。
- 包装器将自行关闭,并通过向 cmd 窗口中的所有进程发送 CTRL_C_EVENT 来关闭应获取 CTRL-C 信号的程序。
- 控制程序应该能够请求包装器发送 CTRL-C 信号。这可以通过 IPC 手段来实现,例如插座。
有用的帖子是:
我必须删除链接前面的 http,因为我是新用户,不允许发布两个以上的链接。
更新:基于 IPC 的 CTRL-C Wrapper
在这里你可以找到一个自写的 python 模块,它提供了一个 CTRL-C 包装,包括一个基于套接字的 IPC。
语法与子进程模块非常相似。
用法:
>>> import winctrlc
>>> p1 = winctrlc.Popen("python demo.py")
>>> p2 = winctrlc.Popen("python demo.py")
>>> p3 = winctrlc.Popen("python demo.py")
>>> p2.send_ctrl_c()
>>> p1.send_ctrl_c()
>>> p3.send_ctrl_c()
代码
import socket
import subprocess
import time
import random
import signal, os, sys
class Popen:
_port = random.randint(10000, 50000)
_connection = ''
def _start_ctrl_c_wrapper(self, cmd):
cmd_str = "start \"\" python winctrlc.py "+"\""+cmd+"\""+" "+str(self._port)
subprocess.Popen(cmd_str, shell=True)
def _create_connection(self):
self._connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._connection.connect(('localhost', self._port))
def send_ctrl_c(self):
self._connection.send(Wrapper.TERMINATION_REQ)
self._connection.close()
def __init__(self, cmd):
self._start_ctrl_c_wrapper(cmd)
self._create_connection()
class Wrapper:
TERMINATION_REQ = "Terminate with CTRL-C"
def _create_connection(self, port):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('localhost', port))
s.listen(1)
conn, addr = s.accept()
return conn
def _wait_on_ctrl_c_request(self, conn):
while True:
data = conn.recv(1024)
if data == self.TERMINATION_REQ:
ctrl_c_received = True
break
else:
ctrl_c_received = False
return ctrl_c_received
def _cleanup_and_fire_ctrl_c(self, conn):
conn.close()
os.kill(signal.CTRL_C_EVENT, 0)
def _signal_handler(self, signal, frame):
time.sleep(1)
sys.exit(0)
def __init__(self, cmd, port):
signal.signal(signal.SIGINT, self._signal_handler)
subprocess.Popen(cmd)
conn = self._create_connection(port)
ctrl_c_req_received = self._wait_on_ctrl_c_request(conn)
if ctrl_c_req_received:
self._cleanup_and_fire_ctrl_c(conn)
else:
sys.exit(0)
if __name__ == "__main__":
command_string = sys.argv[1]
port_no = int(sys.argv[2])
Wrapper(command_string, port_no)