informatic
/
py9b
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cleanups for py3 migration

legit-fork
informatic 2019-10-18 16:11:14 +02:00
parent c6168dd727
commit 20417bfd6b
6 changed files with 97 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
py9b/link/ble.py
View File

@ -9,23 +9,23 @@ SCAN_TIMEOUT = 3
try:
import queue
import queue
except ImportError:
import Queue as queue
import Queue as queue
class Fifo():
def __init__(self):
self.q = queue.Queue()
def __init__(self):
self.q = queue.Queue()
def write(self, data): # put bytes
for b in data:
self.q.put(b)
def write(self, data): # put bytes
for b in data:
self.q.put(b)
def read(self, size=1, timeout=None): # but read string
res = ''
for i in xrange(size):
res += chr(self.q.get(True, timeout))
return res
def read(self, size=1, timeout=None): # but read string
res = ''
for i in xrange(size):
res += chr(self.q.get(True, timeout))
return res
#_cccd_uuid = '00002902-0000-1000-8000-00805f9b34fb'
@ -35,76 +35,77 @@ _tx_char_uuid = '6e400003-b5a3-f393-e0a9-e50e24dcca9e'
_write_chunk_size = 20 # as in android dumps
class BLELink(BaseLink):
def __init__(self, *args, **kwargs):
super(BLELink, self).__init__(*args, **kwargs)
self._adapter = None
self._dev = None
self._wr_handle = None
self._rx_fifo = Fifo()
def __init__(self, *args, **kwargs):
super(BLELink, self).__init__(*args, **kwargs)
self._adapter = None
self._dev = None
self._wr_handle = None
self._rx_fifo = Fifo()
def __enter__(self):
self._adapter = pygatt.GATTToolBackend()
self._adapter.start()
return self
def __enter__(self):
self._adapter = pygatt.GATTToolBackend()
self._adapter.start()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.close()
def __exit__(self, exc_type, exc_value, traceback):
self.close()
def _make_rx_cb(self): # this is a closure :)
def rx_cb(handle, value):
self._rx_fifo.write(value)
return rx_cb
def _make_rx_cb(self): # this is a closure :)
def rx_cb(handle, value):
self._rx_fifo.write(value)
return rx_cb
def scan(self):
res = []
devices = self._adapter.scan(timeout=SCAN_TIMEOUT)
for dev in devices:
if dev['name'] and dev['name'].startswith((u'MISc', u'NBSc', u'JP2', u'Seg')):
res.append((dev['name'], dev['address']))
return res
def scan(self):
res = []
self._adapter.reset()
devices = self._adapter.scan(timeout=SCAN_TIMEOUT)
for dev in devices:
if dev['name'] and dev['name'].startswith((u'MISc', u'NBSc', u'JP2', u'Seg')):
res.append((dev['name'], dev['address']))
return res
def open(self, port):
try:
self._dev = self._adapter.connect(port, address_type=pygatt.BLEAddressType.random)
self._dev.subscribe(_tx_char_uuid, callback=self._make_rx_cb())
self._wr_handle = self._dev.get_handle(_rx_char_uuid)
except pygatt.exceptions.NotConnectedError:
raise LinkOpenException
def open(self, port):
try:
self._dev = self._adapter.connect(port, address_type=pygatt.BLEAddressType.random)
self._dev.subscribe(_tx_char_uuid, callback=self._make_rx_cb())
self._wr_handle = self._dev.get_handle(_rx_char_uuid)
except pygatt.exceptions.NotConnectedError:
raise LinkOpenException
def close(self):
if self._dev:
self._dev.disconnect()
self._dev = None
if self._adapter:
self._adapter.stop()
def close(self):
if self._dev:
self._dev.disconnect()
self._dev = None
if self._adapter:
self._adapter.stop()
def read(self, size):
try:
data = self._rx_fifo.read(size, timeout=self.timeout)
except queue.Empty:
raise LinkTimeoutException
if self.dump:
print '<', hexlify(data).upper()
return data
def read(self, size):
try:
data = self._rx_fifo.read(size, timeout=self.timeout)
except queue.Empty:
raise LinkTimeoutException
if self.dump:
print('<', hexlify(data).upper())
return data
def write(self, data):
if self.dump:
print '>', hexlify(data).upper()
size = len(data)
ofs = 0
while size:
chunk_sz = min(size, _write_chunk_size)
self._dev.char_write_handle(self._wr_handle, bytearray(data[ofs:ofs+chunk_sz]))
ofs += chunk_sz
size -= chunk_sz
def write(self, data):
if self.dump:
print('>', hexlify(data).upper())
size = len(data)
ofs = 0
while size:
chunk_sz = min(size, _write_chunk_size)
self._dev.char_write_handle(self._wr_handle, bytearray(data[ofs:ofs+chunk_sz]))
ofs += chunk_sz
size -= chunk_sz
__all__ = ['BLELink']

4
py9b/link/serial.py
View File

@ -48,13 +48,13 @@ class SerialLink(BaseLink):
if len(data)<size:
raise LinkTimeoutException
if self.dump:
print "<", hexlify(data).upper()
print("<", hexlify(data).upper())
return data
def write(self, data):
if self.dump:
print ">", hexlify(data).upper()
print(">", hexlify(data).upper())
self.com.write(data)

6
py9b/link/tcp.py
View File

@ -44,7 +44,7 @@ class TCPLink(BaseLink):
p = port.partition(':')
host = p[0]
port = int(p[2], 10)
print host, port
print(host, port)
try:
self.sock.connect((host, port))
except socket.timeout:
@ -61,13 +61,13 @@ class TCPLink(BaseLink):
def read(self, size):
data = recvall(self.sock, size)
if data and self.dump:
print "<", hexlify(data).upper()
print("<", hexlify(data).upper())
return data
def write(self, data):
if self.dump:
print ">", hexlify(data).upper()
print(">", hexlify(data).upper())
size = len(data)
ofs = 0
while size:

2
py9b/transport/base.py
View File

@ -3,7 +3,7 @@
def checksum(data):
s = 0
for c in data:
s += ord(c)
s += c
return (s & 0xFFFF) ^ 0xFFFF

19
py9b/transport/packet.py
View File

@ -2,15 +2,16 @@ from binascii import hexlify
from .base import BaseTransport as BT
class BasePacket(object):
def __init__(self, src=0, dst=0, cmd=0, arg=0, data=""):
self.src = src
self.dst = dst
self.cmd = cmd
self.arg = arg
self.data = data
def __init__(self, src=0, dst=0, cmd=0, arg=0, data=""):
self.src = src
self.dst = dst
self.cmd = cmd
self.arg = arg
self.data = data
print(self.data)
def __str__(self):
return "%s->%s: %02X @%02X %s" % (BT.GetDeviceName(self.src), BT.GetDeviceName(self.dst), self.cmd, self.arg, hexlify(self.data).upper())
def __str__(self):
return "%s->%s: %02X @%02X %s" % (BT.GetDeviceName(self.src), BT.GetDeviceName(self.dst), self.cmd, self.arg, hexlify(self.data).upper())
__all__ = ["BasePacket"]

28
py9b/transport/xiaomi.py
View File

@ -23,7 +23,7 @@ class XiaomiTransport(BT):
BT.MOTOR : {BT.HOST : MOTOR, BT.ESC : MOTOR, BT.BMS : MOTOR } }
# TBC
_BleAddr2SaDa = { MASTER2ESC : (BT.HOST, BT.ESC),
_BleAddr2SaDa = { MASTER2ESC : (BT.HOST, BT.ESC),
ESC2MASTER : (BT.ESC, BT.HOST),
MASTER2BMS : (BT.HOST, BT.BMS),
BMS2MASTER : (BT.BMS, BT.HOST),
@ -31,7 +31,7 @@ class XiaomiTransport(BT):
BLE2MASTER : (BT.BLE, BT.HOST),
MOTOR : (BT.MOTOR, BT.HOST) }
_BmsAddr2SaDa = { MASTER2ESC : (BT.BMS, BT.ESC),
_BmsAddr2SaDa = { MASTER2ESC : (BT.BMS, BT.ESC),
ESC2MASTER : (BT.ESC, BT.BMS),
MASTER2BMS : (BT.ESC, BT.BMS),
BMS2MASTER : (BT.BMS, BT.ESC),
@ -48,7 +48,7 @@ class XiaomiTransport(BT):
def _make_addr(self, src, dst):
return XiaomiTransport._SaDa2Addr[src][dst]
def _split_addr(self, addr):
if self.device==BT.BMS:
return XiaomiTransport._BmsAddr2SaDa[addr]
@ -60,13 +60,13 @@ class XiaomiTransport(BT):
while True:
while True:
c = self.link.read(1)
if c=="\x55":
if c==b"\x55":
break
while True:
c = self.link.read(1)
if c=="\xAA":
return True
if c!="\x55":
if c==b"\xAA":
return True
if c!=b"\x55":
break # start waiting 55 again, else - this is 55, so wait for AA
@ -74,21 +74,21 @@ class XiaomiTransport(BT):
self._wait_pre()
pkt = self.link.read(1)
l = ord(pkt)+3
for i in xrange(l):
pkt += self.link.read(1)
for i in range(l):
pkt.extend(self.link.read(1))
ck_calc = checksum(pkt[0:-2])
ck_pkt = unpack("<H", pkt[-2:])[0]
if ck_pkt!=ck_calc:
print "Checksum mismatch !"
print("Checksum mismatch !")
return None
sa, da = self._split_addr(ord(pkt[1]))
return BasePacket(sa, da, ord(pkt[2]), ord(pkt[3]), pkt[4:-2]) # sa, da, cmd, arg, data
sa, da = self._split_addr(pkt[1])
return BasePacket(sa, da, pkt[2], pkt[3], pkt[4:-2]) # sa, da, cmd, arg, data
def send(self, packet):
dev = self._make_addr(packet.src, packet.dst)
pkt = pack("<BBBB", len(packet.data)+2, dev, packet.cmd, packet.arg)+packet.data
pkt = "\x55\xAA" + pkt + pack("<H", checksum(pkt))
pkt = b"\x55\xAA" + pkt + pack("<H", checksum(pkt))
self.link.write(pkt)