-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathcppyplot_server.py
More file actions
205 lines (171 loc) · 5.87 KB
/
Copy pathcppyplot_server.py
File metadata and controls
205 lines (171 loc) · 5.87 KB
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
### Custom Imports ###
lib_sym = {}
## Import numpy and register the object in the symbol table
import numpy as np
lib_sym['np'] = np
## Import matplotlib and register plot object in the symbol table
import matplotlib.pyplot as plt
import matplotlib.cm as cm
lib_sym['plt'] = plt
lib_sym['cm'] = cm
# from matplotlib.animation import FuncAnimation
# lib_sym['FuncAnimation'] = FuncAnimation
## Import seaborn and register seaborn object in the symbol table
# import seaborn as sns
# lib_sym['sns'] = sns
## Import pandas and register pandas object in the symbol table
# import pandas as pd
# lib_sym['pd'] = pd
# Import bokeh and register bokeh objects in the symbol table
# from bokeh.layouts import column, row
# lib_sym['column'] = column
# lib_sym['row'] = row
# from bokeh.plotting import ColumnDataSource, figure, output_file, show
# lib_sym['ColumnDataSource'] = ColumnDataSource
# lib_sym['figure'] = figure
# lib_sym['output_file'] = output_file
# lib_sym['show'] = show
# Import plotly
# import plotly.io as pio
# # pio.renderers.default = "browser" # comment this when using Dash for rendering
# import plotly.graph_objects as go
# from plotly.subplots import make_subplots
# lib_sym['go'] = go
# lib_sym['make_subplots'] = make_subplots
# lib_sym['pio'] = pio
# Import dash
# import dash
# import dash_core_components as dcc
# import dash_html_components as html
# lib_sym['dash'] = dash
# lib_sym['dcc'] = dcc
# lib_sym['html'] = html
#### required imports ####
import zmq
import sys
from threading import Thread
from struct import unpack
from queue import Queue
from asteval import Interpreter, make_symbol_table
#### Globals #####
recv_msgs = Queue()
parsed_msgs = Queue()
kill_thread = False
aeval = Interpreter()
aeval.symtable = make_symbol_table(use_numpy=True, **lib_sym, no_print=False)
# indices for data access
SYM_IDX = 1
TYPE_IDX = 2
LEN_IDX = 3
SHAPE_IDX = 4
#### utility functions ####
def subscriber(addr):
global recv_msgs
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect(addr)
socket.setsockopt(zmq.LINGER, 0)
socket.setsockopt_string(zmq.SUBSCRIBE, "")
print("[INFO] started subscriber thread ")
print(f"[INFO] listening to {addr}")
while (not kill_thread):
if (socket.poll(50, zmq.POLLIN)):
zmq_message = socket.recv()
recv_msgs.put(zmq_message)
def parse_shape(shape_str:str)->list:
shape = []
axis_size = ""
for char in shape_str:
if (char.isdigit()):
axis_size = axis_size + char
elif(char==',' or char==')'):
if (axis_size != ""):
shape.append(int(axis_size))
axis_size = ""
return shape
def handle_payload(data, data_type, data_len, data_shape, _unpack=unpack):
if ((data_type == 'c') or (data_type == 'b') or (data_type == 'B')):
data_converted = _unpack("="+(data_type*data_len), data)
return (b''.join(data_converted)).decode("utf-8")
else:
if data_shape[0] > 0:
return np.ndarray(data_shape, dtype="="+data_type, buffer=data)
else:
return (_unpack("="+data_type, data))[0]
def update_data(header, data, plot_data:dict)->dict:
data_info = header.decode("utf-8").split('|')
# 0: data, 1: var_name, 2: var_type, 3: n_elems, 4: array_shape
data_type = data_info[TYPE_IDX]
data_len = int(data_info[LEN_IDX])
data_shape = parse_shape(data_info[SHAPE_IDX])
plot_data[data_info[SYM_IDX]] = handle_payload(data, data_type, data_len, data_shape)
return plot_data
def update_cmd(zmq_message)->str:
return zmq_message.decode("utf-8")
def parse_msgs():
global parsed_msgs, recv_msgs
plot_cmd = None
plot_data = {}
while (not kill_thread):
zmq_message = None
if (not recv_msgs.empty()):
zmq_message = recv_msgs.get()
recv_msgs.task_done()
else:
continue
if (zmq_message[0:4] == b"data"):
header = zmq_message
data = recv_msgs.get()
recv_msgs.task_done()
plot_data = update_data(header, data, plot_data)
elif (zmq_message[0:8] == b"finalize"):
parsed_msgs.put(("plot", plot_cmd, plot_data, ))
plot_cmd = None
plot_data = {}
elif (zmq_message == b"exit"):
parsed_msgs.put(("exit", 0,))
else:
plot_cmd = update_cmd(zmq_message)
def plot_handler(plot_cmd:str, plot_data:dict)->None:
global aeval
print("[INFO] plotting ...")
aeval.symtable = {**aeval.symtable, **plot_data}
aeval.eval(plot_cmd)
if (aeval.error_msg != None):
aeval.error_msg = None
plt.figure(figsize=(6,5))
plt.title("Exception from ASTEVAL, check stdout", fontsize=14)
plt.show()
print("[INFO] done")
def exit_handler(exit_code):
global kill_thread
kill_thread = True
print("[INFO] requested to shutdown, goodbye!")
def run_main():
global parsed_msgs
cmd_handler = {}
cmd_handler["plot"] = plot_handler
cmd_handler["exit"] = exit_handler
print(f"[INFO] plotting server initialized")
try:
while(not kill_thread):
if (not parsed_msgs.empty()):
msg = parsed_msgs.get_nowait()
parsed_msgs.task_done()
func = cmd_handler[msg[0]]
func(*(msg[1:]))
except KeyboardInterrupt:
print("[Warning] received keyboardInterrupt, killing plotting server")
exit_handler(0)
#### main ####
if __name__ == '__main__':
addr = "tcp://127.0.0.1:5555"
if (len(sys.argv) > 1):
addr = sys.argv[1]
sub_thread = Thread(target=subscriber, args=[addr])
parser_thread = Thread(target=parse_msgs)
sub_thread.start()
parser_thread.start()
run_main()
sub_thread.join()
parser_thread.join()