-
Notifications
You must be signed in to change notification settings - Fork 4
/
replay.py
executable file
·85 lines (64 loc) · 2.39 KB
/
replay.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
#!/usr/bin/python3
"""Host a server that sends data specified by some JSON file.
"""
import json
import time
import socket
import base64
DEFAULT_PORT = 25565 # default minecraft port
DEFAULT_SPACING = 100000 # default ns interval for compressed replay (.1ms)
def ns_time():
return time.time() * 10**9
def host_debug_server(host, port, data, do_echo):
"""Send a non-zero number of packets to the connecting client.
"""
if type(data) == list:
data_offset = data[0]['timestamp']
else:
data_offset = 0 # because generator
server_offset = ns_time()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind((host, port))
s.listen(1)
conn, addr = s.accept()
with conn:
print('Connected to', addr)
for packet in data:
ts = packet['timestamp']
direction = packet['bountTo']
payload = packet['payload']
if direction != "clientbound":
continue
# busy wait is the only accurate wait
while ns_time() - server_offset < ts - data_offset:
pass
conn.sendall(base64.b64decode(payload))
if do_echo:
print("sent: ", payload)
def _compress(index, data, spacing=DEFAULT_SPACING):
data['timestamp'] = index * spacing
return data
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("data", help="path to json data file")
parser.add_argument("-v", "--verbose",
action="store_true",
help="toggle echoing server output")
parser.add_argument("-p", "--port",
type=int, default=DEFAULT_PORT,
help="port to listen from")
parser.add_argument("--fast",
action='store_true',
help="use fast replay interval (1packet/500ns)")
args = parser.parse_args()
_json_path = args.data
_port = args.port
_echo = args.verbose
# load data and compress if needed
with open(_json_path) as json_data:
_data = json.load(json_data)['packets']
if args.fast:
_data = (_compress(i, pkt, args.fast) for i, pkt in enumerate(_data))
# listen on all available interfaces
host_debug_server('', _port, _data, _echo)