aboutsummaryrefslogtreecommitdiffstats
path: root/src/gui/net.py
blob: 715e7a086aee600c6f5175e205b97fad627afeeb (plain)
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
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):
        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])