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
|
import select
import socket
from urllib.parse import urlparse
import re
import numpy as np
from numpy_ringbuffer import RingBuffer
import dearpygui.dearpygui as dpg
class udpsource:
"""
Creates an UDP listening socket
"""
def __init__(self, url, dtype, timeout=0.05):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.url = urlparse(url)
self.dtype = dtype
self.timeout = timeout
def __del__(self):
self.sock.close()
def bind(self):
self.sock.setblocking(False)
self.sock.bind((self.url.hostname, self.url.port))
# self.sock.listen()
def read(self, nblocks):
# TODO: run in a separate thread (it will be painful to implement)
ready, _w, _x = select.select([self.sock], [], [], self.timeout)
if not ready:
return None
# read from socket
blocksize = 1024 * 4
string = ready[0].recv(nblocks * blocksize).decode("ascii")
# decode string, remove empty values
chunks = filter(None, re.split(r"\[(.+?)\]", string))
def chunk_to_samples(chunk):
samples = chunk.split(",")
if samples:
try:
return list(map(self.dtype, samples))
except ValueError:
return []
# convert each chunk into a list of samples
chunk_values = map(chunk_to_samples, chunks)
# flatten list of lists into a single list
values = sum(chunk_values, [])
return values
class network_plot(udpsource):
"""
Wraps a udpsource while at the same time intefacing with DearPyGUI as a plot element.
"""
def __init__(self, url, dtype, nsamples, **kwargs):
udpsource.__init__(self, url, dtype)
self.nsamples = nsamples
self._init_buffers()
self._init_dpg_plot(**kwargs)
# listen for connections
self.bind()
def _init_buffers(self):
# create buffers for x and y values
self.xvalues = RingBuffer(capacity=self.nsamples, dtype=np.dtype(self.dtype))
self.yvalues = RingBuffer(capacity=self.nsamples, dtype=np.dtype(self.dtype))
self.xvalues.extend(np.arange(0, self.nsamples))
self.yvalues.extend(np.zeros(self.nsamples))
def _init_dpg_plot(self, **kwargs):
if "tag" in kwargs:
self.tag = kwargs["tag"]
self.series_tag = f"{self.tag}_series"
self.xaxis_tag = f"{self.tag}_xaxis"
self.yaxis_tag = f"{self.tag}_yaxis"
self.window_tag = f"window_{self.tag}"
self.plot = dpg.plot(**kwargs)
# Map `with' expressions to the underlying plot
def __enter__(self):
return self.plot.__enter__()
def __exit__(self, t, val, tb):
self.plot.__exit__(t, val, tb)
@property
def xdata(self):
# unwrap ringbuffer
return np.array(self.xvalues)
@property
def ydata(self):
# unwrap ringbuffer
return np.array(self.yvalues)
def refresh_series(self, tag):
new_values = self.read(10)
if new_values:
self.yvalues.extendleft(new_values)
dpg.set_value(tag, [self.xdata, self.ydata])
class network_constellation_plot(network_plot):
"""
Special case of a plot, where complex numbers are drawn into a scatter plot
"""
def __init__(self, url, nsamples, **kwargs):
network_plot.__init__(self, url, np.complex64, nsamples, **kwargs)
def _init_buffers(self):
self.xvalues = RingBuffer(capacity=self.nsamples, dtype=np.float32)
self.yvalues = RingBuffer(capacity=self.nsamples, dtype=np.float32)
self.xvalues.extend(np.zeros(self.nsamples))
self.yvalues.extend(np.zeros(self.nsamples))
def refresh_series(self, tag):
new_values = self.read(1)
if new_values:
self.xvalues.extendleft(np.real(new_values))
self.yvalues.extendleft(np.imag(new_values))
dpg.set_value(tag, [self.xdata, self.ydata])
|
