Programmiererforum

Code: Select all

plugin_test.py

Code: Select all

from subprocess import PIPE, Popen
from threading import Thread
from queue import Queue, Empty
import re
import os
import sys

def read_stdout(stdout, data_q):
for line in stdout.readline():
print(line)
data_q.put(line)
stdout.close()

class Plugin():
name = ''

def __init__(self, **kwargs) -> None:
self.process = None

def run(self) -> bool:
try:
self.process = Popen(
[sys.executable, 'test.py'],
stdin=PIPE,
stdout=PIPE,
text=True)
except Exception as exc:
print("Could not create plugin process {}. {}".format(self.name, exc))
return False

self.data_q = Queue()
self.read_t = Thread(target=read_stdout, args=(self.process.stdout, self.data_q))
self.read_t.daemon = True
self.read_t.start()

try:
startup = self.data_q.get(timeout=10)
except Empty:
print("Plugin took to long to load {}.".format(self.name))
self.stop()
return False
else:
if 'Error' in startup:
print("Could not load plugin {}. {}".format(self.name, startup))
self.stop()
return False
elif '100%' in startup:
print("Plugin \"{}\" started.".format(self.name))

return True

def write(self, data : str) -> None:
self.process.stdin.write(data)
self.process.stdin.flush()

def read(self) -> str:
try:
data = self.data_q.get()
print("Got data ::{}::".format(data))
except Exception as exc:
print("Error reading data from plugin '{}'".format(exc))

def stop(self):
self.process.terminate()
self.process.wait()

if __name__ == '__main__':
plugin = Plugin()
plugin.run()
< /code>
test.py

Code: Select all

print("this is a test")
print("this is a test2")
print("100%")
< /code>
root@osboxes# python plugin_test.py

t
h
i
s

i
s

a

t
e
s
t