我正在尝试自动化 win 应用程序和 java 应用程序之间的同步关系。 我的标准是:
(请注意,我们正在应用程序的控制台中执行命令,即两个应用程序都需要是活动进程。)
我很难尝试实现互斥锁或二进制信号量。
这是我试图完成的代码片段:
import datetime
import time
import multiprocessing
class Win_App():
def __init__(self):
self.input_command = [] # list of string
self.output_log = [] #list of string
self.next_jav_command=''
def execute_win_application(self):
exe_path = 'path_ to_exe'
init_command= 'command to execute '
win_process = subprocess.Popen('cmd.exe', stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True)
win_process.stdin.write(exe_path)
win_process.stdin.write(init_command)
print('Win Application Started !')
while True:
#TODO: mutex/semaphore acquire and start tasks
win_process.stdin.write(self.input_command[-1]) # execute the last command in the list
for lines in process.stdout():
write_into_file("win_app_output.log",line)
self.output_log.append(lines)
self.next_jav_command = parse_lines_and_get_command()
if (nex_win_command == '-1'):
self.next_jav_command == 'exit'
break:
# TODO: hold the process here and move to Java application
# TODO: release mutex/ semaphore
class Java_App():
def __init__(self):
self.input_command = [] # list of string
self.output_log = [] #list of string
self.win_next_command = ''
def execute_java_application(self):
jav_path = 'path_ to_java_app'
init_command= 'command to execute '
jav_process = subprocess.Popen('cmd.exe', stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True)
print('Java Application Started!')
while True:
#TODO: mutex/semaphore acquire and start tasks
jav_process.stdin.write(self.input_command[-1]) # execute the last command in the list
for lines in process.stdout():
write_into_file("jav_app_output.log",line)
self.output_log.append(lines)
nex_win_command = parse_lines_and_get_command()
if (nex_win_command == 'exit'):
break:
# TODO: hold the process here and move to Java application
# TODO: release mutex/ semaphore
jav_process.kill()
##### MAIN PROGRAM #####
if __name__ == '__main__':
# Initialize process given certain parameters
win = Win_App()
jav = Java_App()
p1 = multiprocessing.Process(target=win.execute_win_application)
p1.start()
print(" win App started!")
p2 = multiprocessing.Process(target=jav.execute_java_application)
p2.start()
while True:
# TODO : implement context switch between process in execution
# like a binary semaphore or mutex
我能够处理微型函数/方法,但我无法实现上面代码中的#TODO 任务。
预先感谢您的所有帮助。
我尝试将输入逐行发送到通道中的
subprocess.Popen
,但没有任何运气。肯定的选择是使用 pexpect
,正如 AKX 指出的那样。
这是一个帮助您入门的玩具示例。首先要做的是创建一个虚拟环境并在该虚拟环境中安装
pexpect
。
作为这个玩具示例的一部分,我编写了 2 个应用程序:
代码:
#!/usr/bin/env python3
"""
app1.py
Print text in reversed
"""
while (input_text := input("> ")) != "q":
print(input_text[::-1])
#!/usr/bin/env python3
"""
app2.py
Reverse each words
"""
while (text := input("> ")) != "q":
words = [word[::-1] for word in text.split()]
print(" ".join(words))
#!/usr/bin/env python3
"""
main.py
Spawn 2 different apps, send input to them and receiving output.
"""
import pexpect
PROMPT = "> "
def main():
"""App Entry"""
app1 = pexpect.spawn("python3 app1.py", echo=False)
app2 = pexpect.spawn("python3 app2.py", echo=False)
text_input = [
"The more you learn",
"I love my dog",
]
for text in text_input:
app1.expect(PROMPT)
app1.sendline(text)
output = app1.readline()
output = output.rstrip().decode() # Convert bytes to str
print(f"{text} -> {output} ", end="")
app2.expect(PROMPT)
app2.sendline(output)
output2 = app2.readline()
output2 = output2.rstrip().decode() # Convert bytes to str
print(f"-> {output2}")
# Quit apps
app1.expect(PROMPT)
app1.sendline("q")
app2.expect(PROMPT)
app2.sendline("q")
if __name__ == "__main__":
main()
一些注意事项:
echo=False
来防止输入文本的回显