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
102 lines (80 loc) · 3.11 KB
/
Copy pathclient.py
File metadata and controls
102 lines (80 loc) · 3.11 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
""" Implement client connection to lognplot protocol.
"""
import socket
import struct
import datetime
import time
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.
"""
timestamp = coerce_timestamp(timestamp)
value = float(value)
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: float, samples):
""" Send equidistant spaced samples.
"""
timestamp = coerce_timestamp(timestamp)
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.
"""
timestamp = coerce_timestamp(timestamp)
self._send_dict(
{"name": name, "t": timestamp, "type": "event", "attributes": attributes}
)
def send_text(self, name, timestamp, text):
timestamp = coerce_timestamp(timestamp)
self._send_dict({"name": name, "t": timestamp, "type": "text", "text": text})
def send_function_enter(self, name, timestamp, function_name):
""" Trace function entry. """
timestamp = coerce_timestamp(timestamp)
self._send_dict({"name": name, "t": timestamp, "type": "profile", "event": "enter", "callee": function_name})
def send_function_exit(self, name, timestamp):
""" Trace function entry. """
timestamp = coerce_timestamp(timestamp)
self._send_dict({"name": name, "t": timestamp, "type": "profile", "event": "exit"})
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)
def coerce_timestamp(timestamp) -> float:
""" Convert a timestamp into a floating point number. """
if isinstance(timestamp, datetime.datetime):
return time.mktime(timestamp.timetuple()) + timestamp.microsecond / 1e6
else:
return float(timestamp)