-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathmain.py
executable file
·194 lines (168 loc) · 7.04 KB
/
main.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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Main entrypoint script for AH.
Author: 8baller
Authorised use only
"""
import os
import sys
import time
from argparse import ArgumentParser
AH_LOGO = " _ _ \n / \\ _ _| |_ ___ _ __ ___ _ __ ___ ___ _ _ ___ \n / _ \\| | | | __/ _ \\| '_ \\ / _ \\| '_ ` _ \\ / _ \\| | | / __|\n / ___ \\ |_| | || (_) | | | | (_) | | | | | | (_) | |_| \\__ \\\n/_/ \\_\\__,_|\\__\\___/|_| |_|\\___/|_| |_| |_|\\___/ \\__,_|___/\n \n _ _ _ _ \n| | | | ___ __ _(_) ___(_) __ _ _ __ \n| |_| |/ _ \\/ _` | |/ __| |/ _` | '_ \\ \n| _ | __/ (_| | | (__| | (_| | | | |\n|_| |_|\\___|\\__, |_|\\___|_|\\__,_|_| |_|\n |___/ \n"
NUMBER_DB_CREATIONS = 1
WHITESPACE = "\n------------------------\n\n"
def parse_args():
"""Parse arguments."""
parser = ArgumentParser(description="Cli tool for the Autonomouse Hegician.")
parser.add_argument("-d", "--dev_mode", action="store_true")
parser.add_argument(
"-o",
"--options",
help="Comma seperated list of actions to take.",
default="",
)
return parser.parse_args()
def docker_cleanup(func):
"""Decorator that manages docker teardown before and after."""
def wrap(*args, **kwargs):
print(f"{WHITESPACE}Removing exising docker containers...")
code = os.system("docker-compose down")
if code != 0:
raise RuntimeError("Failed to destroy existing containers!")
try:
result = func(*args, **kwargs)
except RuntimeError:
raise
finally:
print(f"{WHITESPACE}Removing exising docker containers...")
code = os.system("docker-compose down")
if code != 0:
print(f"Error on `docker-compose down`. Code={code}")
return result
return wrap
@docker_cleanup
def run_tests():
"""Run all tests."""
print(f"{WHITESPACE}Starting backend services....")
code = os.system("docker-compose up -d postgresdb ganachecli api")
if code != 0:
raise RuntimeError("Failed to start test environment containers!")
print(f"{WHITESPACE}Installing virtual environment with dependencies....")
code = os.system("cd agents; pipenv install --skip-lock")
if code != 0:
raise RuntimeError("Failed to install dependencies!")
print(f"{WHITESPACE}Creating database schema....")
cmd = "cd agents; pipenv run python autonomous_hegician/skills/option_management/db_communication.py"
for _ in range(NUMBER_DB_CREATIONS):
code = os.system(cmd)
if code == 0:
continue
if code != 0:
raise RuntimeError("Failed to create database!")
print(f"{WHITESPACE}Updating the ledger from the environment vars....")
code = os.system("cd agents; pipenv run update_ledger")
if code != 0:
raise RuntimeError("Failed to update ledger of Autonomous Hegician!")
print(f"{WHITESPACE}Attempting to deploy contracts to the chain....")
code = os.system("cd agents; pipenv run deploy_contracts")
if code != 0:
raise RuntimeError("Deploying contracts has failed!")
print(f"{WHITESPACE}Updating contracts from testnet in Autonomous Hegician config...")
code = os.system("cd agents; pipenv run update_contracts_testnet")
if code != 0:
raise RuntimeError(
"Failed to update newly deployed contracts to Autonomous Hegician!"
)
print(f"{WHITESPACE}Prepping autonomous hegician ....")
code = os.system("cd agents; pipenv run prep_ah")
if code != 0:
raise RuntimeError("Failed to run autonomous hegician prep!")
print(f"{WHITESPACE}Running agent functionality tests ....")
code = os.system("cd agents; pipenv run test_ah")
if code != 0:
raise RuntimeError("Failed to run integration tests successfully!")
print(f"{WHITESPACE}Running api functionality tests ....")
code = os.system("cd agents; pipenv run test_ah_via_api")
if code != 0:
raise RuntimeError("Failed to run api integration tests successfully!")
@docker_cleanup
def deploy_contracts_to_testnet():
"""Deploy contracts to testnet."""
print(f"{WHITESPACE}Starting backend services....")
code = os.system("docker-compose up -d ganachecli")
if code != 0:
raise RuntimeError("Failed to started local chain")
print(f"{WHITESPACE}Installing virtual environment with dependencies....")
code = os.system("cd agents; pipenv install --skip-lock")
if code != 0:
raise RuntimeError("Failed to install dependencies!")
code = os.system("cd agents; pipenv run deploy_contracts")
if code != 0:
raise RuntimeError("Deploying contracts has failed!")
def launch_containers():
"""Launch docker containers."""
code = os.system("docker-compose up -d --build")
if code != 0:
raise RuntimeError("Launching containers has failed!")
print(
"Containers running in background.\nVisit: `http://0.0.0.0:3001`.\nTo shut down: `docker-compose down`."
)
def setup_live():
code = os.system("cd agents; pipenv run update_contracts_live")
if code != 0:
raise RuntimeError("Deploying contracts has failed!")
launch_containers()
def update_ah_config(config="testnet"):
"""Update the AH config."""
if config != "testnet":
i = os.system("cd agents; pipenv install --skip-lock")
i2 = os.system("cd agents; pipenv run update_contracts")
i3 = os.system("cd agents; pipenv run update_ledger")
if sum([i, i2, i3]) != 0:
raise RuntimeError("Updateing the AH config has failed!")
def main():
"""Run the main method."""
choices = {
k + 1: v
for k, v in enumerate(
[
["Run local tests.", run_tests],
["Launch live containerised Autonomous Hegician.", setup_live],
]
)
}
args = parse_args()
if args.dev_mode:
print("Dev mode currently inconsequential!")
if args.options != "":
for k in [k for k in args.options.split(",") if k != ""]:
try:
name, func = choices[int(k)]
print(f"Executing {k}... {name}")
func()
except KeyError:
print("Invalid options selected!")
return
print("Please choose from the following actions;")
[print(f"\n{i[0]}. {i[1][0]}") for i in choices.items()]
try:
i = int(input())
name, func = choices[i]
print(f"\nRunning:\n\n{name}\n")
except KeyboardInterrupt:
print("\nBye! See you soon!")
return
except KeyError:
print("Invalid option!")
func()
def check_python_version():
"""Check python version satisfies requirements."""
if not (sys.version_info[0] >= 3 and sys.version_info[1] >= 6):
print("Python 3.6 or higher required!")
sys.exit(1)
if __name__ == "__main__":
print(AH_LOGO)
check_python_version()
main()
sys.exit(0)