OPEN_RPA_PUBLIC
/
ORPA-pyOpenRPA
2
0
Fork 1
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

Remove Orchestrator ProcessGSettingsget and ProcessLoggerGet (con not be created singleton... )

Browse Source 
Fix in ProcessDefIntervalCall
dev-linux
Ivan Maslov 3 years ago
parent 1df423c09f
commit 3b2c4e3888
1 changed files with 59 additions and 45 deletions
Show Stats Download Patch File Download Diff File

104
Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py
Unescape View File

@ -942,68 +942,82 @@ def ProcessListGet(inProcessNameWOExeList=None):
pass
return lResult
def ProcessGSettingsGet():
"""
Return GSettings (singleton) from global variable
:return:
"""
global gSettingsDict
return gSettingsDict
def ProcessLoggerGet():
"""
Return Logger from global variable
:return:
"""
global gSettingsDict
lResult = None
if "Logger" in gSettingsDict: lResult = gSettingsDict["Logger"]
return lResult
def ProcessDefIntervalCall(inDef, inIntervalSecFloat, inIntervalAsyncBool=False, inDefArgList=None, inDefArgDict=None, inExecuteInNewThreadBool=True, inLogger=None):
def ProcessDefIntervalCall(inGSettings, inDef, inIntervalSecFloat, inIntervalAsyncBool=False, inDefArgList=None, inDefArgDict=None, inDefArgGSettingsNameStr=None, inDefArgLoggerNameStr=None, inExecuteInNewThreadBool=True, inLogger=None):
"""
Use this procedure if you need to run periodically some def. Set def, args, interval and enjoy :)
:param inGSettings: global settings
:param inDef: def link, which will be called with interval inIntervalSecFloat
:param inIntervalSecFloat: Interval in seconds between call
:param inIntervalAsyncBool: False - wait interval before next call after the previous iteration result; True - wait interval after previous iteration call
:param inDefArgList: List of the args in def. Default None (empty list)
:param inDefArgDict: Dict of the args in def. Default None (empty dict)
:param inDefArgGSettingsNameStr: Name of the GSettings arg name for def (optional)
:param inDefArgLoggerNameStr: Name of the Logger arg name for def (optional). If Use - please check fill of the inLogger arg.
:param inExecuteInNewThreadBool: True - create new thread for the periodic execution; False - execute in current thread. Default: True
:param inLogger: logging def if some case is appear
:return:
"""
#Some edits on start
if inDefArgDict is None: inDefArgDict = {}
if inDefArgList is None: inDefArgList = []
# Check if inDefArgLogger is set and inLogger is exist
if inDefArgLoggerNameStr=="": inDefArgLoggerNameStr=None
if inDefArgGSettingsNameStr=="": inDefArgGSettingsNameStr=None
if inDefArgLoggerNameStr is not None and not inLogger:
raise Exception(f"!ERROR! ProcessDefIntervalCall - You need to send logger in def because your def is require logger. Raise error!")
# Internal def to execute periodically
def __Execute__(inDef, inIntervalSecFloat, inIntervalAsyncBool, inDefArgList, inDefArgDict, inLogger):
while True:
try:
# Call async if needed
if inIntervalAsyncBool==False: # Case wait result then wait
inDef(*inDefArgList, **inDefArgDict)
else: # Case dont wait result - run sleep then new iteration (use many threads)
lThread2 = threading.Thread(target=__Execute__,
kwargs={"inDef": inDef, "inIntervalSecFloat": inIntervalSecFloat,
"inIntervalAsyncBool": inIntervalAsyncBool, "inDefArgList": inDefArgList,
"inDefArgDict": inDefArgDict})
lThread2.start()
except Exception as e:
if inLogger: inLogger.exception(f"ProcessDefIntervalCall: Interval call has been failed. Traceback is below. Code will sleep for the next call")
# Sleep interval
time.sleep(inIntervalSecFloat)
# Check to call in new thread
if inExecuteInNewThreadBool:
lThread = threading.Thread(target=__Execute__,
kwargs={"inDef":inDef, "inIntervalSecFloat":inIntervalSecFloat,
"inIntervalAsyncBool":inIntervalAsyncBool, "inDefArgList":inDefArgList,
"inDefArgDict":inDefArgDict, "inLogger":inLogger})
lThread.start()
# Check thread
if not Core.IsProcessorThread(inGSettings=inGSettings):
if inGSettings["Logger"]: inGSettings["Logger"].warning(f"__Orchestrator__.ProcessDefIntervalCall def was called not from processor queue - activity will be append in the processor queue.")
lProcessorActivityDict = {
"Def": ProcessDefIntervalCall, # def link or def alias (look gSettings["Processor"]["AliasDefDict"])
"ArgList": [], # Args list
"ArgDict": {"inDef": inDef, "inIntervalSecFloat": inIntervalSecFloat,
"inIntervalAsyncBool":inIntervalAsyncBool, "inDefArgList": inDefArgList,
"inDefArgDict": inDefArgDict, "inDefArgGSettingsNameStr":inDefArgGSettingsNameStr,
"inDefArgLoggerNameStr": inDefArgLoggerNameStr, "inExecuteInNewThreadBool": inExecuteInNewThreadBool}, # Args dictionary
"ArgGSettings": "inGSettings", # Name of GSettings attribute: str (ArgDict) or index (for ArgList)
"ArgLogger": "inLogger" # Name of GSettings attribute: str (ArgDict) or index (for ArgList)
}
inGSettings["ProcessorDict"]["ActivityList"].append(lProcessorActivityDict)
else:
__Execute__(inDef=inDef, inIntervalSecFloat=inIntervalSecFloat, inIntervalAsyncBool=inIntervalAsyncBool, inDefArgList=inDefArgList, inDefArgDict=inDefArgDict, inLogger=inLogger)
# Internal def to execute periodically
def __Execute__(inGSettings, inDef, inIntervalSecFloat, inIntervalAsyncBool, inDefArgList, inDefArgDict, inLogger, inDefArgGSettingsNameStr, inDefArgLoggerNameStr):
if inLogger: inLogger.info(f"__Orchestrator__.ProcessDefIntervalCall: Interval execution has been started. Def: {str(inDef)}")
# Prepare gSettings and logger args
if inDefArgGSettingsNameStr is not None:
inDefArgDict[inDefArgGSettingsNameStr] = inGSettings
if inDefArgLoggerNameStr is not None:
inDefArgDict[inDefArgLoggerNameStr] = inLogger
while True:
try:
# Call async if needed
if inIntervalAsyncBool == False: # Case wait result then wait
inDef(*inDefArgList, **inDefArgDict)
else: # Case dont wait result - run sleep then new iteration (use many threads)
lThread2 = threading.Thread(target=inDef,
args=inDefArgList,
kwargs=inDefArgDict)
lThread2.start()
except Exception as e:
if inLogger: inLogger.exception(
f"ProcessDefIntervalCall: Interval call has been failed. Traceback is below. Code will sleep for the next call")
# Sleep interval
time.sleep(inIntervalSecFloat)
# Check to call in new thread
if inExecuteInNewThreadBool:
lThread = threading.Thread(target=__Execute__,
kwargs={"inGSettings":inGSettings, "inDef": inDef, "inIntervalSecFloat": inIntervalSecFloat,
"inIntervalAsyncBool": inIntervalAsyncBool, "inDefArgList": inDefArgList,
"inDefArgDict": inDefArgDict, "inLogger": inLogger})
lThread.start()
else:
__Execute__(inGSettings=inGSettings, inDef=inDef, inIntervalSecFloat=inIntervalSecFloat, inIntervalAsyncBool=inIntervalAsyncBool,
inDefArgList=inDefArgList, inDefArgDict=inDefArgDict, inLogger=inLogger,
inDefArgGSettingsNameStr=inDefArgGSettingsNameStr , inDefArgLoggerNameStr=inDefArgLoggerNameStr)
# Python def - start module function

Write Preview
Loading…
Cancel
Save