-
Notifications
You must be signed in to change notification settings - Fork 0
/
run.py
115 lines (99 loc) · 2.86 KB
/
run.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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
# Usage:
# python run.py -w <worsheet full path> [-d <fsm def full path>] [-s 'no'] [-t 'no']
#
import argparse
import json
import sys
import cv2
from flow_converter import FlowConverter
from flow_model import FlowModel
from flow_runner import Runner
from flow_storage import FlowStorage, FlowStorageConfig
# Construct the argument parser and parse the arguments
def parseArgs():
ap = argparse.ArgumentParser(description="flow runner fsm")
ap.add_argument("-w", "--ws", required = True,
help = "full path to the worksheet file")
ap.add_argument("-d", "--def", required = False,
help = "full path to the fsm definition file")
ap.add_argument("-s", "--step", required = False,
default="no",
help = "step mode")
ap.add_argument("-t", "--trace", required = False,
default="no",
help = "print output")
args = ap.parse_args()
kwargs = dict((k,v) for k,v in vars(args).items() if k!="message_type")
return kwargs
# Reads the string from the file, parses the JSON data,
# populates a Python dict with the data
def readJson(ffn):
with open(ffn, 'rt') as f:
data = json.load(f)
return data
def writeJson(ffn, data):
with open(ffn, 'w') as fp:
json.dump(data, fp, indent=2)
return
# Read configuration files
def readConfig():
return readJson('./cfg/fsm-cfg.json')
def set_runtime_environment(cfg):
actions_paths = cfg.get('user-actions-paths')
for path in actions_paths:
sys.path.append(path)
def run_by_step(runner: Runner, events):
while(True):
print('q - quit')
prompt = f"waits for an event {events}:"
print(prompt)
event = input()
if event == 'q':
return
if event not in events:
continue
idx = runner.state_idx
runner.run_step(event, idx)
return
def run_all(runner):
runner.run_all()
return
# Main function - the runner's client example
def main(**kwargs):
fsm_conf = readConfig()
set_runtime_environment(fsm_conf)
ffn = kwargs.get("ws")
ws = readJson(ffn)
# Get FRFSM defintion
if kwargs.get("def"):
# directly, from fsm def
fsm_def = readJson(kwargs.get("def"))
else:
# by convert the ws
model = FlowModel(ws)
# fc = FlowConverter(model)
# fsm_def = fc.convert()
# if kwargs['trace'] == 'yes':
# writeJson('../data/fsm-def/fsm-def-test.json', fsm_def)
# # Create Flow Storage
# storage_cfg = fsm_conf.get('storage', '.')
# config = FlowStorageConfig(storage_cfg)
# storage = FlowStorage(config, model.get_as_ws())
# Create the runner
rn = Runner()
rn.build(fsm_conf, model)
rn.reset()
# Recreate context, when a flow meta changed
# rn.storage = storage
# rn.create_frfsm(fsm_conf, fsm_def)
# rn.start()
# rn.init_storage(image)
if kwargs.get("step") == "no":
run_all(rn)
else:
run_by_step(rn, fsm_conf.get('events'))
return
# Entry point
if __name__ == "__main__":
kwargs = parseArgs()
main(**kwargs)