Added Ninebot protocol support

Added unified register read tool
2018-12-07 01:38:16 +01:00 · 2018-12-07 01:38:16 +01:00 · 794d5bb24d
parent 4079e4f619
commit 794d5bb24d
12 changed files with 360 additions and 109 deletions
--- a/fwupd.py
+++ b/fwupd.py
@ -6,11 +6,9 @@ import argparse
 from progressbar import ProgressBar

 from py9b.link.base import LinkOpenException, LinkTimeoutException
-from py9b.link.tcp import TCPLink
-from py9b.link.ble import BLELink
-from py9b.link.serial import SerialLink
 from py9b.transport.base import BaseTransport as BT
 from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
 from py9b.command.regio import ReadRegs, WriteRegs
 from py9b.command.update import *

@ -59,7 +57,8 @@ def UpdateFirmware(link, tran, dev, fwfile):
 		chunk_sz = min(fw_size, fw_page_size)
 		data = fwfile.read(chunk_sz)
 		chk = checksum(chk, data)
-		tran.execute(WriteUpdate(dev, page, data))
+		#tran.execute(WriteUpdate(dev, page, data))
+		tran.execute(WriteUpdate(dev, page, data+b'\x00'*(fw_page_size-chunk_sz))) # TODO: Ninebot wants this padding. Will it work on M365 too?
 		page += 1
 		fw_size -= chunk_sz
 	pb.finish()
@ -86,13 +85,12 @@ parser.add_argument('device', help='target device', type=str.lower, choices=devi

 parser.add_argument('file', type=argparse.FileType('rb'), help='firmware file')

-interfaces = {'ble' : BLELink, 'serial' : SerialLink, 'tcp' : TCPLink}
 parser.add_argument('-i', '--interface', help='communication interface, default: %(default)s', type=str.lower,
-	choices=interfaces,  default='ble')
+	choices=('ble', 'serial', 'tcp'),  default='ble')

 parser.add_argument('-a', '--address', help='communication address (ble: BDADDR, serial: port, tcp: host:port), default: first available')

-protocols = {'xiaomi' : XiaomiTransport} # TODO: add Ninebot ES
+protocols = {'xiaomi' : XiaomiTransport, 'ninebot' : NinebotTransport }
 parser.add_argument('-p', '--protocol', help='communication protocol, default: %(default)s', type=str.lower,
 	choices=protocols, default='xiaomi')

@ -102,12 +100,25 @@ if len(argv)==1:
 args = parser.parse_args()

 dev = devices.get(args.device)
-link = interfaces.get(args.interface)()

+if args.interface=='ble':
+	try:
+		from py9b.link.ble import BLELink
+	except:
+		exit('BLE is not supported on your system !')
+	link = BLELink()
+elif args.interface=='tcp':
+	from py9b.link.tcp import TCPLink
+	link = TCPLink()
+elif args.interface=='serial':
+	from py9b.link.serial import SerialLink
+	link = SerialLink()
+else:
+	exit('!!! BUG !!! Unknown interface selected: '+args.interface)
+		
 with link:
 	tran = protocols.get(args.protocol)(link)

-
 	if args.address:
 		addr = args.address
 	else:
@ -124,4 +135,4 @@ with link:
 		UpdateFirmware(link, tran, dev, args.file)
 	except Exception as e:
 		print('Error:', e)
-
+		raise
--- a/powerdown.py
+++ b/powerdown.py
@ -0,0 +1,31 @@
+#!python2-32
+from py9b.link.base import LinkOpenException, LinkTimeoutException
+from py9b.link.tcp import TCPLink
+from py9b.link.ble import BLELink
+from py9b.link.serial import SerialLink
+from py9b.transport.base import BaseTransport as BT
+from py9b.transport.packet import BasePacket as PKT
+from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
+from py9b.command.regio import ReadRegs, WriteRegs
+
+#link = SerialLink(dump=True)
+#link = TCPLink()
+link = BLELink()
+
+with link:
+	print "Scanning..."
+	ports = link.scan()
+	print ports
+
+	#tran = XiaomiTransport(link)
+	tran = NinebotTransport(link)
+
+	#link.open(("192.168.1.45", 6000))
+	link.open(ports[0][1])
+	print "Connected"
+
+	print('Power off...')
+	tran.execute(WriteRegs(BT.ESC, 0x79, '<H', 0x0001)) 
+
+	link.close()
--- a/py9b/command/regio.py
+++ b/py9b/command/regio.py
@ -28,10 +28,18 @@ class WriteRegs(BaseCommand):
 		self.reg = reg

 	def handle_response(self, response):
-		if response.arg!=self.reg or len(response.data)!=1:
+		if response.cmd==0x02: # xiaomi style
+			if response.arg!=self.reg or len(response.data)!=1:
+				raise InvalidResponse("WriteRegs {0:X}:{1:X}".format(self.dev, self.reg))
+			if unpack("<B", response.data)[0]!=1:
+				raise WriteProtectError("WriteRegs {0:X}:{1:X}".format(self.dev, self.reg))
+		elif response.cmd==0x05: # ninebot style
+			if len(response.data)!=0:
+				raise InvalidResponse("WriteRegs {0:X}:{1:X}".format(self.dev, self.reg))
+			if response.arg!=0:
+				raise WriteProtectError("WriteRegs {0:X}:{1:X}".format(self.dev, self.reg))
+		else:
 			raise InvalidResponse("WriteRegs {0:X}:{1:X}".format(self.dev, self.reg))
-		if unpack("<B", response.data)[0]!=1:
-			raise WriteProtectError("WriteRegs {0:X}:{1:X}".format(self.dev, self.reg))
 		return True


--- a/py9b/link/ble.py
+++ b/py9b/link/ble.py
@ -19,15 +19,15 @@ class Fifo():
 			self.q.put(b)

 	def read(self, size=1, timeout=None): # but read string
-		res = ""
+		res = ''
 		for i in xrange(size):
 			res += chr(self.q.get(True, timeout))
 		return res


-#_cccd_uuid = "00002902-0000-1000-8000-00805f9b34fb"
-_rx_char_uuid = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"
-_tx_char_uuid = "6e400003-b5a3-f393-e0a9-e50e24dcca9e"
+#_cccd_uuid = '00002902-0000-1000-8000-00805f9b34fb'
+_rx_char_uuid = '6e400002-b5a3-f393-e0a9-e50e24dcca9e'
+_tx_char_uuid = '6e400003-b5a3-f393-e0a9-e50e24dcca9e'

 _write_chunk_size = 20 # as in android dumps

@ -60,8 +60,8 @@ class BLELink(BaseLink):
 		res = []
 		devices = self._adapter.scan(timeout=SCAN_TIMEOUT)
 		for dev in devices:
-			if dev["name"].startswith(u"MISc"):
-				res.append((dev["name"], dev["address"]))
+			if dev['name'].startswith((u'MISc', u'NBSc')):
+				res.append((dev['name'], dev['address']))
 		return res


@ -88,13 +88,13 @@ class BLELink(BaseLink):
 		except queue.Empty:
 			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()
 		size = len(data)
 		ofs = 0
 		while size:
@ -104,4 +104,4 @@ class BLELink(BaseLink):
 			size -= chunk_sz


-__all__ = ["BLELink"]
+__all__ = ['BLELink']
--- a/py9b/link/tcp.py
+++ b/py9b/link/tcp.py
@ -4,73 +4,77 @@ import socket
 from binascii import hexlify
 from .base import BaseLink, LinkTimeoutException, LinkOpenException

-HOST, PORT = "127.0.0.1", 6000
-
-_write_chunk_size = 20 # as in android dumps
-
-def recvall(sock, size):
-	data = ""
-	while len(data)<size:
-		try:
-			pkt = sock.recv(size-len(data))
-		except socket.timeout:
-			raise LinkTimeoutException()
-		data+=pkt
-	return data
-
-
-class TCPLink(BaseLink):
-	def __init__(self, *args, **kwargs):
-		super(TCPLink, self).__init__(*args, **kwargs)
-		self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-		self.sock.settimeout(self.timeout)
-		self.connected = False
-
-
-	def __enter__(self):
-		return self
-
-
-	def __exit__(self, exc_type, exc_value, traceback):
-		self.close()
-
-
-	def scan(self):
-		res = [("Android UART Bridge", (HOST, PORT))]
-		return res
-
-
-	def open(self, port):
-		try:
-			self.sock.connect(port)
-		except socket.timeout:
-			raise LinkOpenException
-		self.connected = True
-
-
-	def close(self):
-		if self.connected:
-			self.sock.close()
-			self.connected = False
-
-
-	def read(self, size):
-		data = recvall(self.sock, size)
-		if data and 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.sock.sendall(data[ofs:ofs+chunk_sz])
-			ofs += chunk_sz
-			size -= chunk_sz
-
-
-__all__ = ["TCPLink"]
+HOST, PORT = "127.0.0.1", 6000
+
+_write_chunk_size = 20 # 20 as in android dumps
+
+def recvall(sock, size):
+	data = ""
+	while len(data)<size:
+		try:
+			pkt = sock.recv(size-len(data))
+		except socket.timeout:
+			raise LinkTimeoutException()
+		data+=pkt
+	return data
+
+
+class TCPLink(BaseLink):
+	def __init__(self, *args, **kwargs):
+		super(TCPLink, self).__init__(*args, **kwargs)
+		self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+		self.sock.settimeout(self.timeout)
+		self.connected = False
+
+
+	def __enter__(self):
+		return self
+
+
+	def __exit__(self, exc_type, exc_value, traceback):
+		self.close()
+
+
+	def scan(self):
+		res = [("Android UART Bridge", HOST+':'+str(PORT))]
+		return res
+
+
+	def open(self, port):
+		p = port.partition(':')
+		host = p[0]
+		port = int(p[2], 10)
+		print host, port
+		try:
+			self.sock.connect((host, port))
+		except socket.timeout:
+			raise LinkOpenException
+		self.connected = True
+
+
+	def close(self):
+		if self.connected:
+			self.sock.close()
+			self.connected = False
+
+
+	def read(self, size):
+		data = recvall(self.sock, size)
+		if data and 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.sock.sendall(data[ofs:ofs+chunk_sz])
+			ofs += chunk_sz
+			size -= chunk_sz
+
+
+__all__ = ["TCPLink"]
--- a/py9b/transport/base.py
+++ b/py9b/transport/base.py
@ -13,9 +13,10 @@ class BaseTransport(object):
 	ESC = 0x20
 	BLE = 0x21
 	BMS = 0x22
+	EXTBMS = 0x23
 	HOST = 0x3E
 	
-	DeviceNames = { MOTOR : "MOTOR", ESC : "ESC", BLE : "BLE", BMS : "BMS", HOST : "HOST" }
+	DeviceNames = { MOTOR : "MOTOR", ESC : "ESC", BLE : "BLE", BMS : "BMS", EXTBMS : "EXTBMS", HOST : "HOST" }

 	def __init__(self, link):
 		self.link = link
--- a/py9b/transport/ninebot.py
+++ b/py9b/transport/ninebot.py
@ -0,0 +1,47 @@
+"""Ninebot packet transport"""
+from struct import pack, unpack
+from .base import checksum, BaseTransport as BT
+from .packet import BasePacket
+
+
+class NinebotTransport(BT):
+	def __init__(self, link, device=BT.HOST):
+		super(NinebotTransport, self).__init__(link)
+		self.device = device
+
+
+	def _wait_pre(self):
+		while True:
+			while True:
+				c = self.link.read(1)
+				if c=="\x5A":
+					break
+			while True:
+				c = self.link.read(1)
+				if c=="\xA5":
+					return True 
+				if c!="\x5A":
+					break # start waiting 5A again, else - this is 5A, so wait for A5
+
+
+	def recv(self):
+		self._wait_pre()
+		pkt = self.link.read(1)
+		l = ord(pkt)+6
+		for i in xrange(l):
+			pkt += 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 !"
+			return None
+		return BasePacket(ord(pkt[1]), ord(pkt[2]), ord(pkt[3]), ord(pkt[4]), pkt[5:-2]) # sa, da, cmd, arg, data
+
+	
+	def send(self, packet):
+		pkt = pack("<BBBBB", len(packet.data), packet.src, packet.dst, packet.cmd, packet.arg)+packet.data
+		pkt = "\x5A\xA5" + pkt + pack("<H", checksum(pkt))
+		self.link.write(pkt)
+
+
+__all__ = ["NinebotTransport"]
--- a/read_bms.py
+++ b/read_bms.py
@ -6,6 +6,7 @@ from py9b.link.serial import SerialLink
 from py9b.transport.base import BaseTransport as BT
 from py9b.transport.packet import BasePacket as PKT
 from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
 from py9b.command.regio import ReadRegs

 READ_CHUNK_SIZE = 0x10
@ -19,7 +20,8 @@ with link:
 	ports = link.scan()
 	print ports

-	tran = XiaomiTransport(link)
+	#tran = XiaomiTransport(link)
+	tran = NinebotTransport(link)

 	#link.open(("192.168.1.45", 6000))
 	link.open(ports[0][1])
--- a/read_esc.py
+++ b/read_esc.py
@ -6,27 +6,29 @@ from py9b.link.serial import SerialLink
 from py9b.transport.base import BaseTransport as BT
 from py9b.transport.packet import BasePacket as PKT
 from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
 from py9b.command.regio import ReadRegs

 READ_CHUNK_SIZE = 0x10

-#link = SerialLink(dump=True)
+link = SerialLink()
 #link = TCPLink()
-link = BLELink()
+#link = BLELink()

 with link:
 	print "Scanning..."
 	ports = link.scan()
 	print ports

-	tran = XiaomiTransport(link)
+	#tran = XiaomiTransport(link)
+	tran = NinebotTransport(link)

 	#link.open(("192.168.1.45", 6000))
 	link.open(ports[0][1])
 	print "Connected"

 	hfo = open("EscRegs.bin", "wb")
-	for i in xrange(0x0, 0x100, READ_CHUNK_SIZE):
+	for i in xrange(0x0, 0x200, READ_CHUNK_SIZE):
 		print ".",
 		for retry in xrange(5):
 			try:
--- a/readregs.py
+++ b/readregs.py
@ -0,0 +1,108 @@
+#!python2-32
+from __future__ import print_function
+from sys import argv, exit
+import os
+import argparse
+from progressbar import ProgressBar
+
+from py9b.link.base import LinkOpenException, LinkTimeoutException
+from py9b.link.tcp import TCPLink
+from py9b.link.ble import BLELink
+from py9b.link.serial import SerialLink
+from py9b.transport.base import BaseTransport as BT
+from py9b.transport.packet import BasePacket as PKT
+from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
+from py9b.command.regio import ReadRegs
+
+READ_CHUNK_SIZE = 0x10
+
+def ReadAllRegs(link, tran, dev, hfo):
+	size = 0x200 if dev==BT.ESC else 0x100
+	pb = ProgressBar(maxval=size).start()
+	for i in xrange(0x0, size, READ_CHUNK_SIZE):
+		pb.update(i)
+		for retry in xrange(5):
+			try:
+				data = tran.execute(ReadRegs(dev, i>>1, '16s'))[0]
+			except LinkTimeoutException:
+				continue
+			break
+		else:
+			print('No response !')
+			return False
+		hfo.write(data)
+	pb.finish()
+	print('OK')
+	return True
+
+
+##########################################################################################
+
+parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,
+	description='Xiaomi/Ninebot register reader',
+	epilog='Example 1:  %(prog)s esc esc_regs.bin  - read ESC regs to esc_regs.bin using default communication parameters' 
+	'\nExample 2:  %(prog)s -i tcp -a 192.168.1.10:6000 bms bms_regs.bin  - flash BMS regs over TCP-BLE bridge at 192.168.1.10:6000'
+	'\nExample 3:  %(prog)s -i serial -a COM2 esc esc_regs.bin  - read ESC regs via COM2'
+	'\nExample 4:  %(prog)s -i ble -a 12:34:56:78:9A:BC esc esc_regs.bin  - read ESC regs via BLE, use specified BLE address')
+	
+devices = {'esc' : BT.ESC, 'bms' : BT.BMS, 'extbms' : BT.EXTBMS }
+parser.add_argument('device', help='device to read from', type=str.lower, choices=devices)
+
+parser.add_argument('file', type=argparse.FileType('wb'), help='output file')
+
+parser.add_argument('-i', '--interface', help='communication interface, default: %(default)s', type=str.lower,
+	choices=('ble', 'serial', 'tcp'),  default='ble')
+
+parser.add_argument('-a', '--address', help='communication address (ble: BDADDR, serial: port, tcp: host:port), default: first available')
+
+protocols = {'xiaomi' : XiaomiTransport, 'ninebot' : NinebotTransport }
+parser.add_argument('-p', '--protocol', help='communication protocol, default: %(default)s', type=str.lower,
+	choices=protocols, default='xiaomi')
+
+if len(argv)==1:
+	parser.print_usage()
+	exit()
+args = parser.parse_args()
+
+if args.device=='extbms' and args.protocol!='ninebot':
+	exit('Only Ninebot supports External BMS !')
+
+dev = devices.get(args.device)
+
+if args.interface=='ble':
+	try:
+		from py9b.link.ble import BLELink
+	except:
+		exit('BLE is not supported on your system !')
+	link = BLELink()
+elif args.interface=='tcp':
+	from py9b.link.tcp import TCPLink
+	link = TCPLink()
+elif args.interface=='serial':
+	from py9b.link.serial import SerialLink
+	link = SerialLink()
+else:
+	exit('!!! BUG !!! Unknown interface selected: '+args.interface)
+		
+with link:
+	tran = protocols.get(args.protocol)(link)
+
+	if args.address:
+		addr = args.address
+	else:
+		print('Scanning...')
+		ports = link.scan()
+		if not ports:
+			exit('No interfaces found !')
+		print('Connecting to', ports[0][0])
+		addr = ports[0][1]
+
+	link.open(addr)
+	print('Connected')
+	try:
+		ReadAllRegs(link, tran, dev, args.file)
+		args.file.close()
+	except Exception as e:
+		print('Error:', e)
+		raise
--- a/reboot.py
+++ b/reboot.py
@ -0,0 +1,31 @@
+#!python2-32
+from py9b.link.base import LinkOpenException, LinkTimeoutException
+from py9b.link.tcp import TCPLink
+from py9b.link.ble import BLELink
+from py9b.link.serial import SerialLink
+from py9b.transport.base import BaseTransport as BT
+from py9b.transport.packet import BasePacket as PKT
+from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
+from py9b.command.regio import ReadRegs, WriteRegs
+
+#link = SerialLink(dump=True)
+#link = TCPLink()
+link = BLELink()
+
+with link:
+	print "Scanning..."
+	ports = link.scan()
+	print ports
+
+	#tran = XiaomiTransport(link)
+	tran = NinebotTransport(link)
+
+	#link.open(("192.168.1.45", 6000))
+	link.open(ports[0][1])
+	print "Connected"
+
+	print('Reboot...')
+	tran.execute(WriteRegs(BT.ESC, 0x78, '<H', 0x0001)) 
+
+	link.close()
--- a/wr_esc_sn.py
+++ b/wr_esc_sn.py
@ -8,12 +8,14 @@ from py9b.link.serial import SerialLink
 from py9b.transport.base import BaseTransport as BT
 from py9b.transport.packet import BasePacket as PKT
 from py9b.transport.xiaomi import XiaomiTransport
+from py9b.transport.ninebot import NinebotTransport
 from py9b.command.regio import ReadRegs, WriteRegs
 from py9b.command.mfg import WriteSN
+from time import sleep


-new_sn = "16133/00101234"
-
+#new_sn = "16133/00101234"
+#new_sn = "N2GTR1826C1234"

 def CalcSnAuth(oldsn, newsn, uid3):
 	s = 0
@ -37,7 +39,8 @@ with link:
 	ports = link.scan()
 	print(ports)

-	tran = XiaomiTransport(link)
+	#tran = XiaomiTransport(link)
+	tran = NinebotTransport(link)

 	#link.open(("192.168.1.45", 6000))
 	link.open(ports[0][1])
@ -57,7 +60,7 @@ with link:

 	
 	#lock
-	tran.execute(WriteRegs(BT.ESC, 0x70, "<B", 0x01))
+	#tran.execute(WriteRegs(BT.ESC, 0x70, "<H", 0x01))

 	old_sn = tran.execute(ReadRegs(BT.ESC, 0x10, "14s"))[0]
 	print("Old S/N:", old_sn)
@ -66,15 +69,18 @@ with link:
 	print("UID3: %08X" % (uid3))

 	auth = CalcSnAuth(old_sn, new_sn, uid3)
+	#auth = 0
 	print("Auth: %08X"  % (auth))

-	for i in range(3):
-		try:
-			tran.execute(WriteSN(BT.ESC, new_sn, auth))
-			print("OK")
-			break
-		except LinkTimeoutException:
-			print("Timeout !")
+	try:
+		tran.execute(WriteSN(BT.ESC, new_sn, auth))
+		print("OK")
+	except LinkTimeoutException:
+		print("Timeout !")
+
+	# save config and restart
+	tran.execute(WriteRegs(BT.ESC, 0x78, "<H", 0x01))
+	sleep(3)

 	old_sn = tran.execute(ReadRegs(BT.ESC, 0x10, "14s"))[0]
 	print("Current S/N:", old_sn)