-
Notifications
You must be signed in to change notification settings - Fork 0
/
Build.py
executable file
·206 lines (160 loc) · 6.36 KB
/
Build.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
195
196
197
198
199
200
201
202
203
204
205
206
#!/usr/bin/env python3
import BuildConfig
import time
import websocket
from websocket import create_connection
import json
import threading
import argparse
import os
import re
import shutil
# websocket to dev rust server
ws = None
# whether we detected a successful compilation via rcon
reload_successful = False
min_hook_time = 1000
max_hook_time = 0
def main():
parser = argparse.ArgumentParser(description='Manage build operations for the Tebex Oxide plugin (Rust/7 Days).')
parser.add_argument('--TestRemoteReload', action='store_true', help='Connect to the dev Rust server and check if the plugin compiles/reloads.')
parser.add_argument('--DeployTest', action='store_true', help='Runs a deployment script.')
parser.add_argument('--OpenDevConsole', action='store_true', help='Connect to the dev Rust server and open an interactive console.')
parser.add_argument('--WatchHookTimes', action='store_true', help='Continually monitor the average hook time.')
args = parser.parse_args()
# Merge is the main build action
merge_source_files()
if args.DeployTest:
print("Deploying to test server...")
os.system("./DeployTest.sh")
if args.TestRemoteReload or args.OpenDevConsole or args.WatchHookTimes:
threading.Thread(target=connect_websocket_and_read,
args=(f"ws://{BuildConfig.dev_server_ip}:{BuildConfig.dev_server_rcon_port}/{BuildConfig.dev_server_rcon_password}",), daemon=True).start()
time.sleep(1)
if args.TestRemoteReload:
test_remote_reload()
if args.WatchHookTimes:
watch_hook_times()
if args.OpenDevConsole:
open_dev_console()
def merge_source_files():
print('Merging source files...')
output = {}
for file_name in os.listdir(BuildConfig.source_dir):
if file_name not in BuildConfig.source_files:
print("skipping " + file_name)
continue
file_path = f'{BuildConfig.source_dir}/{file_name}'
with open(file_path, 'r') as file:
inside_namespace = False
inside_block = 0
for line in file:
stripped_line = line.strip()
if stripped_line.startswith('namespace'):
inside_namespace = True
elif inside_namespace:
if '{' in stripped_line:
inside_block += 1
if '}' in stripped_line:
inside_block -= 1
if inside_block == 0:
inside_namespace = False
continue
if inside_block > 0:
# Don't print opening braces for the namespace block
if inside_block == 1 and (stripped_line == "{"):
continue
if file_name in output.keys():
output[file_name] += line
else:
output[file_name] = line
with open(BuildConfig.output_file, 'w') as file:
file.write(BuildConfig.plugin_header + "\n")
file.write("namespace Oxide.Plugins\n")
file.write("{\n")
# Write the plugin first as our classes are defined inside of it.
file.write(output["Tebex.cs"][:-2]) # remove the close curly brace to leave class definition open
# Write the rest of our files
for sourceFile in output.keys():
if sourceFile != "Tebex.cs":
file.write(output[sourceFile])
# Closing braces
file.write("\t}\n")
def test_remote_reload():
print('Checking if the plugin compiles/reloads...')
shutil.copyfile(BuildConfig.output_file, BuildConfig.dev_server_dir + "/oxide/plugins/Tebex.cs")
send_rcon_command("oxide.reload Tebex")
time.sleep(2)
if reload_successful:
print("Successfully reloaded plugin on remote server.")
else:
print("Failed to reload plugin.")
def open_dev_console():
print('Opening development RCON console on remote. Enter `exit` to close.')
while True:
command = input(">>> ")
if command == "exit":
break
else:
send_rcon_command(command)
def connect_websocket_and_read(wsUrl):
global ws
print(f"Connecting via {wsUrl}")
ws = create_connection(wsUrl)
ws.settimeout(5)
while True:
try:
response = ws.recv()
if response:
parsed_response = json.loads(response)
on_rcon_response(parsed_response)
except websocket.WebSocketTimeoutException:
continue
def on_rcon_response(parsed_response):
global reload_successful
canPrint = True
message = parsed_response["Message"]
# Ignore print listing since it can be polled
if "Listing" in message or message == '':
canPrint = False
if canPrint:
print("> ", parsed_response)
if "TebexDonate was compiled successfully" in message:
reload_successful = True
# Check for plugin listing
if "Listing" in message:
# Extract hook time with a regex
try:
match = re.search(r'\((\d+\.\d+)s\)', message)
if match:
global min_hook_time
global max_hook_time
lastHookTime = float(match.group(1))
if lastHookTime < min_hook_time:
min_hook_time = lastHookTime
if lastHookTime > max_hook_time:
max_hook_time = lastHookTime
print("hook times: min: {} cur: {} max: {}".format(min_hook_time, lastHookTime, max_hook_time))
else:
print("No match")
except Exception as e:
print(e)
def watch_hook_times():
print("Beginning watch of hook times...")
hook_watch_thread = threading.Thread(target=hook_time_command_sender)
hook_watch_thread.daemon = True #stop when the main app stops
hook_watch_thread.start()
def hook_time_command_sender():
while True:
send_rcon_command("plugins")
time.sleep(1)
def send_rcon_command(command):
global ws
message = json.dumps({
"Identifier": -1,
"Message": command,
"Name": "WebRcon"
})
ws.send(message)
if __name__ == '__main__':
main()