You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ORPA-pyOpenRPA/Orchestrator/orchestratorTimer.py

66 lines
2.9 KiB

from threading import Timer
import datetime
import subprocess
import importlib
class RepeatedTimer(object):
def __init__(self, interval, function, *args, **kwargs):
self._timer = None
self.interval = interval
self.function = function
self.args = args
self.kwargs = kwargs
self.is_running = False
self.start()
def _run(self):
self.is_running = False
lResult = self.function(*self.args, **self.kwargs)
if lResult is not None:
if lResult:
self.start()
def start(self):
if not self.is_running:
self._timer = Timer(self.interval, self._run)
self._timer.start()
self.is_running = True
def stop(self):
self._timer.cancel()
self.is_running = False
############################################################
####Техническая функция обработки таймера - потока
############################################################
def activityLoopExecution(inProcessPath,inProcessArgList,inLoopTimeEndDateTime,inPythonPackageName,inPythonFunctionName):
lResultGoLoop=True
lCurrentDateTime=datetime.datetime.now()
print (datetime.datetime.now().isoformat()+":: Loop activity check")
#Запустить процесс, если установлен inProcessPath
if inProcessPath is not None:
if inProcessPath != "":
lItemArgs=[inProcessPath]
lItemArgs.extend(inProcessArgList)
subprocess.Popen(lItemArgs,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
#Импорт Python пакета и запуск функции из него
if inPythonPackageName is not None:
if inPythonPackageName != "":
try:
#Подключить модуль для вызова
lModule=importlib.import_module(inPythonPackageName)
#Найти функцию
lFunction=getattr(lModule,inPythonFunctionName)
except Exception as e:
print (datetime.datetime.now().isoformat()+":: Loop activity error: module/function not founded")
#Выключить таймер, если время наступило
if lCurrentDateTime>=inLoopTimeEndDateTime:
lResultGoLoop=False
#Вернуть результат
return lResultGoLoop
############################################################
####Функция запуска таймера - потока
############################################################
def activityLoopStart(inActivityLoopSeconds,inProcessPath,inProcessArgList,inLoopTimeEndDateTime,inPythonPackageName,inPythonFunctionName):
lTimer = RepeatedTimer(inActivityLoopSeconds, activityLoopExecution, inProcessPath,inProcessArgList,inLoopTimeEndDateTime,inPythonPackageName,inPythonFunctionName) # it auto-starts, no need of rt.start()
lTimer.start()