-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathducks.py
47 lines (38 loc) · 1.49 KB
/
ducks.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from SpiffWorkflow.bpmn.workflow import BpmnWorkflow
from SpiffWorkflow.camunda.parser.CamundaParser import CamundaParser
from SpiffWorkflow.camunda.specs.UserTask import EnumFormField, UserTask
def show_form(task):
form = task.task_spec.form
if task.data is None:
task.data = {}
for field in form.fields:
prompt = field.label
if isinstance(field, EnumFormField):
prompt += " (Options: " + ', '.join([str(option.id) for option in field.options]) + ")"
if field.type == "boolean":
prompt += " (Options: true, false)"
prompt += " : "
answer = input(prompt)
if field.type == "long":
answer = int(answer)
if field.type == "boolean":
answer = answer.lower().strip()
answer = (answer == 'true' or answer == 'yes')
task.update_data_var(field.id, answer)
parser = CamundaParser()
parser.add_bpmn_file('ducks.bpmn')
spec = parser.get_spec('duck_process')
workflow = BpmnWorkflow(spec)
workflow.do_engine_steps()
ready_tasks = workflow.get_ready_user_tasks()
while len(ready_tasks) > 0:
for task in ready_tasks:
if isinstance(task.task_spec, UserTask):
show_form(task)
print(task.data)
else:
print("Complete Task ", task.task_spec.name)
workflow.complete_task_from_id(task.id)
workflow.do_engine_steps()
ready_tasks = workflow.get_ready_user_tasks()
# print(workflow.last_task.data)