Существует идея подкласса класса Popen и расширения его с помощью некоторых простых декораторов методов. Назовем его ExpirablePopen.
from logging import error
from subprocess import Popen
from threading import Event
from threading import Thread
class ExpirablePopen(Popen):
def __init__(self, *args, **kwargs):
self.timeout = kwargs.pop('timeout', 0)
self.timer = None
self.done = Event()
Popen.__init__(self, *args, **kwargs)
def __tkill(self):
timeout = self.timeout
if not self.done.wait(timeout):
error('Terminating process {} by timeout of {} secs.'.format(self.pid, timeout))
self.kill()
def expirable(func):
def wrapper(self, *args, **kwargs):
# zero timeout means call of parent method
if self.timeout == 0:
return func(self, *args, **kwargs)
# if timer is None, need to start it
if self.timer is None:
self.timer = thr = Thread(target=self.__tkill)
thr.daemon = True
thr.start()
result = func(self, *args, **kwargs)
self.done.set()
return result
return wrapper
wait = expirable(Popen.wait)
communicate = expirable(Popen.communicate)
if __name__ == '__main__':
from subprocess import PIPE
print ExpirablePopen('ssh -T git@bitbucket.org', stdout=PIPE, timeout=1).communicate()
Я нашел библиотеку, которая решила мою проблему:
from neupy import algorithms
from neupy.algorithms.rbfn.utils import pdf_between_data
grnn = algorithms.GRNN(std=0.003)
grnn.train(X, y)
# In this part of the code you can do any moifications you want
ratios = pdf_between_data(grnn.input_train, X, grnn.std)
predicted = (np.dot(grnn.target_train.T, ratios) / ratios.sum(axis=0)).T
Это ссылка на библиотеку: http://neupy.com/apidocs/neupy.algorithms.rbfn.grnn .html