forked from keepkey/python-keepkey
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtransport_udp.py
More file actions
65 lines (49 loc) · 1.79 KB
/
Copy pathtransport_udp.py
File metadata and controls
65 lines (49 loc) · 1.79 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
from __future__ import print_function
'''SocketTransport implements TCP socket interface for Transport.'''
import socket
from select import select
from .transport import Transport
class FakeRead(object):
# Let's pretend we have a file-like interface
def __init__(self, func):
self.func = func
def read(self, size):
return self.func(size)
class UDPTransport(Transport):
def __init__(self, device, *args, **kwargs):
self.buffer = b''
device = device.split(':')
if len(device) < 2:
device = ('0.0.0.0', int(device[0]))
else:
device = (device[0], int(device[1]))
self.socket = None
super(UDPTransport, self).__init__(device, *args, **kwargs)
def _open(self):
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.socket.connect(self.device)
def _close(self):
self.socket.close()
self.socket = None
self.buffer = ''
def ready_to_read(self):
rlist, _, _ = select([self.socket], [], [], 0)
return len(rlist) > 0
def _write(self, msg, protobuf_msg):
for chunk in [msg[i:i+63] for i in range(0, len(msg), 63)]:
chunk = chunk + b'\0' * (63 - len(chunk))
self.socket.send(b'?' + chunk)
def _read(self):
try:
(msg_type, datalen) = self._read_headers(FakeRead(self._raw_read))
return (msg_type, self._raw_read(datalen))
except socket.error:
print("Failed to read from device")
return None
def _raw_read(self, length):
while len(self.buffer) < length:
data = self.socket.recv(64)
self.buffer += data[1:]
ret = self.buffer[:length]
self.buffer = self.buffer[length:]
return ret