forked from windelbouwman/lognplot
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathclient.py
More file actions
73 lines (57 loc) · 1.96 KB
/
Copy pathclient.py
File metadata and controls
73 lines (57 loc) · 1.96 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
""" Implement client connection to lognplot protocol.
"""
import socket
import struct
import cbor
class LognplotTcpClient:
""" Use this client to transmit sample to the lognplot tool.
"""
def __init__(self, hostname="localhost", port=12345):
self._hostname = hostname
self._port = port
def connect(self):
""" Connect to the server.
"""
self._sock = socket.create_connection((self._hostname, self._port))
def send_sample(self, name: str, timestamp, value: float):
""" Send a single timestamp / value pair to the trace with the given name.
"""
self._send_dict(
{"name": name, "t": timestamp, "type": "sample", "value": value}
)
def send_sample_batch(self, name: str, samples):
""" Send a batch of samples.
samples is a list of tuples of what you would pass to send_sample.
"""
self._send_dict(
{"type": "batch", "name": name, "batch": samples,}
)
def send_samples(self, name: str, timestamp, dt, samples):
""" Send equidistant spaced samples.
"""
self._send_dict(
{
"name": name,
"t": timestamp,
"type": "samples",
"dt": dt,
"values": samples,
}
)
def send_event(self, name, timestamp, attributes):
""" Emit an event.
Attributes can be given as a dictionary of key/value strings.
"""
self._send_dict(
{"name": name, "t": timestamp, "type": "event", "attributes": attributes}
)
def _send_dict(self, data):
data2 = cbor.dumps(data)
self._send_message(data2)
def _send_message(self, msg_data):
""" Transmit a whole message prefixed with a length.
"""
data = bytearray()
data.extend(struct.pack(">I", len(msg_data)))
data.extend(msg_data)
self._sock.sendall(data)