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
|
import json
from pathlib import Path
OUTPUT_LOG = False
class EscapeEncoder(json.JSONEncoder):
"""Custom encoder for numpy data types"""
def default(self, obj):
if isinstance(
obj,
(bytes,),
):
return str(type(obj))
return json.JSONEncoder.default(self, obj)
def initialize_logger(config):
global OUTPUT_LOG # noqa: PLW0603
OUTPUT_LOG = config.get("log_network", False)
if OUTPUT_LOG and Path(OUTPUT_LOG).exists():
Path(OUTPUT_LOG).unlink()
class StateExchangeType:
STATE_INITIAL = "----------- INITIAL STATE -----------\n"
STATE_CLIENT_TO_SERVER = "----------- STATE: Client => Server -----------\n"
STATE_SERVER_TO_CLIENT = "----------- STATE: Server => Client -----------\n"
ACTION_CLIENT_TO_SERVER = "----------- EVENT: Client => Server -----------\n"
ACTION_SERVER_TO_CLIENT = "----------- EVENT: Server => Client -----------\n"
def state_exchange(exchange, data):
if OUTPUT_LOG:
with Path(OUTPUT_LOG).open(mode="a") as f:
f.write(exchange)
f.write(json.dumps(data, indent=2, cls=EscapeEncoder))
f.write("\n")
f.write("-" * 60)
f.write("\n")
def initial_state(data):
state_exchange(StateExchangeType.STATE_INITIAL, data)
def state_c2s(data):
state_exchange(StateExchangeType.STATE_CLIENT_TO_SERVER, data)
def state_s2c(data):
state_exchange(StateExchangeType.STATE_SERVER_TO_CLIENT, data)
def action_s2c(data):
state_exchange(StateExchangeType.ACTION_SERVER_TO_CLIENT, data)
def action_c2s(data):
state_exchange(StateExchangeType.ACTION_CLIENT_TO_SERVER, data)
def error(message):
print(f"Error: {message}", flush=True)
if OUTPUT_LOG:
with Path(OUTPUT_LOG).open(mode="a") as f:
f.write("-" * 60)
f.write("\nERROR: ")
f.write(message)
f.write("\n")
f.write("-" * 60)
f.write("\n")
|
