Extended positional variables to allow operation information to pass through to signal registry and dependency check at the action level for the new process registry (Working). Introduced track for operation
This commit is contained in:
@@ -3,9 +3,9 @@ from core.runtime.runtimecontract import buildRuntimeAttr, buildPreRuntimeAttr
|
|||||||
from core.runtime.stopsignals import addStopSignal
|
from core.runtime.stopsignals import addStopSignal
|
||||||
|
|
||||||
|
|
||||||
def buildAction (action, outputs):
|
def buildAction (action, outputs, process):
|
||||||
|
|
||||||
outputs, resolved, match, filter = buildPreRuntimeAttr(action, outputs)
|
outputs, resolved, match, filter = buildPreRuntimeAttr(action, outputs, process)
|
||||||
|
|
||||||
if hasattr(action, "allowaction"):
|
if hasattr(action, "allowaction"):
|
||||||
if not action.allowaction:
|
if not action.allowaction:
|
||||||
@@ -19,6 +19,7 @@ def buildAction (action, outputs):
|
|||||||
if response is None:
|
if response is None:
|
||||||
return outputs, None
|
return outputs, None
|
||||||
|
|
||||||
|
#outputs["processes"][process.name][action.name]["response"] = response
|
||||||
outputs, response = buildRuntimeAttr(action, outputs, resolved, response)
|
outputs, response = buildRuntimeAttr(action, outputs, resolved, response)
|
||||||
|
|
||||||
return outputs, response
|
return outputs, response
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from core.execution.monitor import buildMonitor
|
|||||||
from core.runtime import manageFailure
|
from core.runtime import manageFailure
|
||||||
from core.execution.recursive import buildRecursive
|
from core.execution.recursive import buildRecursive
|
||||||
|
|
||||||
def executeContract(module, outputs):
|
def executeContract(module, outputs, process):
|
||||||
#print("DEBUGGING")
|
#print("DEBUGGING")
|
||||||
#print(module.name)
|
#print(module.name)
|
||||||
if module.kind == "operation":
|
if module.kind == "operation":
|
||||||
@@ -15,7 +15,7 @@ def executeContract(module, outputs):
|
|||||||
|
|
||||||
elif module.kind == "action":
|
elif module.kind == "action":
|
||||||
|
|
||||||
outputs, response = buildAction(module, outputs)
|
outputs, response = buildAction(module, outputs, process)
|
||||||
|
|
||||||
return outputs, response
|
return outputs, response
|
||||||
|
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ def buildMonitor(monitor, outputs):
|
|||||||
module = filter["execute"]
|
module = filter["execute"]
|
||||||
|
|
||||||
if module.kind == "action":
|
if module.kind == "action":
|
||||||
outputs, response = buildAction(module, outputs)
|
outputs, response = buildAction(module, outputs, None)
|
||||||
if module.kind == "operation":
|
if module.kind == "operation":
|
||||||
outputs, responses = buildOperation(module, outputs)
|
outputs, responses = buildOperation(module, outputs)
|
||||||
response = responses
|
response = responses
|
||||||
|
|||||||
@@ -1,46 +1,69 @@
|
|||||||
import logging
|
import logging
|
||||||
|
import pprint
|
||||||
from core.runtime.managefailure import manageFailure
|
from core.runtime.managefailure import manageFailure
|
||||||
from core.execution.action.action import buildAction
|
|
||||||
from core.runtime.stopsignals import checkDependency
|
from core.runtime.stopsignals import checkDependency
|
||||||
from core.execution.processregistry import registerProcess
|
from core.execution.processregistry import registerProcess
|
||||||
#from core.runtime.skipstep import skipStep
|
#from tqdm import tqdm
|
||||||
#from core.runtime.recursive import recursiveSet
|
from rich.progress import track
|
||||||
|
|
||||||
outputs = {}
|
outputs = {}
|
||||||
|
|
||||||
def buildOperation(definition, outputs):
|
def buildOperation(definition, outputs):
|
||||||
from core.execution.executioncontract import executeContract
|
from core.execution.executioncontract import executeContract
|
||||||
|
|
||||||
actions = definition.actions
|
actions = definition.actions
|
||||||
outputs = registerProcess(definition, outputs)
|
outputs = registerProcess(definition, outputs)
|
||||||
|
|
||||||
responses = []
|
responses = []
|
||||||
outputs["stopsignals"] = []
|
|
||||||
for action in actions:
|
|
||||||
#if action.kind == "action":
|
|
||||||
# outputs["processes"][definition.name][action.name] = {}
|
|
||||||
stop = checkDependency(action, outputs)
|
|
||||||
|
|
||||||
#print(f"STOP SIGNAL DEBUGGING during {action.name}")
|
#for action in tqdm(
|
||||||
#print(outputs["stopsignals"])
|
# actions,
|
||||||
|
# desc=definition.name,
|
||||||
|
# unit="action"
|
||||||
|
#):
|
||||||
|
|
||||||
|
|
||||||
|
if getattr(definition, "track", False):
|
||||||
|
actions = track(
|
||||||
|
actions,
|
||||||
|
#description=definition.name
|
||||||
|
description=f"[green]{definition.name}[/green]"
|
||||||
|
)
|
||||||
|
|
||||||
|
for action in actions:
|
||||||
|
|
||||||
|
#Check if dependencies are in stop signal registry
|
||||||
|
stop = checkDependency(action, outputs, definition)
|
||||||
|
|
||||||
if stop:
|
if stop:
|
||||||
outputs["processes"][definition.name]["stopsignals"].append(action.alias)
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
#Register action before its execution
|
||||||
if action.kind == "action":
|
if action.kind == "action":
|
||||||
outputs["processes"][definition.name][action.name] = {}
|
outputs["processes"][definition.name][action.name] = {}
|
||||||
|
|
||||||
outputs, response = executeContract(action, outputs)
|
#Pass the definition through to the action so it can be passed into buildPreRuntimeAttr
|
||||||
|
#Definition should be passed through in order to give access to addStopSignal so it can write directly to the definition during buildPreRuntimeAttr
|
||||||
#if hasattr(action, recursive)
|
#Stop signal can optionally be engaged by doing addStopSignal inside the action definition
|
||||||
|
outputs, response = executeContract(action, outputs, definition)
|
||||||
|
|
||||||
responses.append(response)
|
responses.append(response)
|
||||||
|
|
||||||
|
#Register responses in process registry
|
||||||
if action.kind == "action":
|
if action.kind == "action":
|
||||||
outputs["processes"][definition.name][action.name]["response"] = response
|
outputs["processes"][definition.name][action.name]["response"] = response
|
||||||
|
|
||||||
outputs["processes"][definition.name]["completed"].append(action.alias)
|
|
||||||
#print(outputs["processes"])
|
|
||||||
|
|
||||||
|
#Print completion status after runtime processes
|
||||||
|
print(f"{action.name} completed")
|
||||||
|
|
||||||
|
##Append completed process to process registry
|
||||||
|
outputs["processes"][definition.name]["completed"].append(action.alias)
|
||||||
|
|
||||||
|
#For debugging purposes only
|
||||||
|
process_registry_logs = False
|
||||||
|
if process_registry_logs:
|
||||||
|
pprint.pprint(outputs["processes"])
|
||||||
|
|
||||||
|
#Temporarily disabled manageFailure functions
|
||||||
#except Exception as error:
|
#except Exception as error:
|
||||||
#outputs = manageFailure(error, action, outputs)
|
#outputs = manageFailure(error, action, outputs)
|
||||||
|
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ from core.runtime.applyconditions import applyConditions
|
|||||||
from core.runtime.responsepolicy import responsePolicy
|
from core.runtime.responsepolicy import responsePolicy
|
||||||
from core.runtime.stopsignals import addStopSignal
|
from core.runtime.stopsignals import addStopSignal
|
||||||
|
|
||||||
def buildPreRuntimeAttr(action, outputs):
|
def buildPreRuntimeAttr(action, outputs, process):
|
||||||
outputs = loadSave(action, outputs)
|
outputs = loadSave(action, outputs)
|
||||||
|
|
||||||
resolved = resolveItem(action, outputs)
|
resolved = resolveItem(action, outputs)
|
||||||
@@ -17,7 +17,7 @@ def buildPreRuntimeAttr(action, outputs):
|
|||||||
match, filter = applyConditions(action, resolved, outputs)
|
match, filter = applyConditions(action, resolved, outputs)
|
||||||
|
|
||||||
if match:
|
if match:
|
||||||
outputs = addStopSignal(action, outputs)
|
outputs = addStopSignal(action, outputs, process)
|
||||||
|
|
||||||
return outputs, resolved, match, filter
|
return outputs, resolved, match, filter
|
||||||
|
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
def checkDependency(action, outputs):
|
#def checkDependency(action, outputs):
|
||||||
#def checkDependency(action, operation, outputs):
|
def checkDependency(action, outputs, operation):
|
||||||
if hasattr(action, "depends"):
|
if hasattr(action, "depends"):
|
||||||
stop = any(
|
stop = any(
|
||||||
#dependency in outputs["processes"][operation.name]["stopsignals"]
|
dependency in outputs["processes"][operation.name]["stopsignals"]
|
||||||
dependency in outputs["stopsignals"]
|
#dependency in outputs["stopsignals"]
|
||||||
for dependency in action.depends
|
for dependency in action.depends
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -12,9 +12,9 @@ def checkDependency(action, outputs):
|
|||||||
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def addStopSignal(action, outputs):
|
#def addStopSignal(action, outputs):
|
||||||
#def addStopSignal(action, operation, outputs):
|
def addStopSignal(action, outputs, operation):
|
||||||
if hasattr(action, "depends"):
|
if hasattr(action, "depends"):
|
||||||
#outputs["processes"][operation.name]["stopsignals"].append(action.alias)
|
outputs["processes"][operation.name]["stopsignals"].append(action.alias)
|
||||||
outputs["stopsignals"].append(action.alias)
|
#outputs["stopsignals"].append(action.alias)
|
||||||
return outputs
|
return outputs
|
||||||
@@ -48,7 +48,7 @@ def buildListener(listener, self):
|
|||||||
|
|
||||||
#Process and forward to RMASS
|
#Process and forward to RMASS
|
||||||
#thread, lifecycles = buildThread(listener.execute, lifecycles)
|
#thread, lifecycles = buildThread(listener.execute, lifecycles)
|
||||||
outputs, responses = executeContract(listener.execute, outputs)
|
outputs, responses = executeContract(listener.execute, outputs, None)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ def manageLifecycle(module, lifecycles): #lifecycles, state):
|
|||||||
while lifecycles[module.name]["mode"] == "continuous":
|
while lifecycles[module.name]["mode"] == "continuous":
|
||||||
print(lifecycles)
|
print(lifecycles)
|
||||||
|
|
||||||
outputs = executeContract(module)
|
outputs = executeContract(module, None, None)
|
||||||
|
|
||||||
#print(lifecycles)
|
#print(lifecycles)
|
||||||
|
|
||||||
@@ -19,7 +19,7 @@ def manageLifecycle(module, lifecycles): #lifecycles, state):
|
|||||||
|
|
||||||
elif lifecycles[module.name]["mode"] == "oneshot":
|
elif lifecycles[module.name]["mode"] == "oneshot":
|
||||||
print(lifecycles)
|
print(lifecycles)
|
||||||
outputs = executeContract(module)
|
outputs = executeContract(module, None, None)
|
||||||
|
|
||||||
#print(lifecycles)
|
#print(lifecycles)
|
||||||
return
|
return
|
||||||
|
|||||||
Reference in New Issue
Block a user