From 5ab4669dedf67efd44b9d9465fa8d43fc1aa8ae2 Mon Sep 17 00:00:00 2001
From: Piotr Dobrowolski <admin@tastycode.pl>
Date: Sun, 22 Apr 2018 07:21:00 +0200
Subject: [PATCH] Initial commit

---
 .gitignore |   1 +
 rdatool.py | 290 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 291 insertions(+)
 create mode 100644 .gitignore
 create mode 100644 rdatool.py

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..539da74
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+*.py[co]
diff --git a/rdatool.py b/rdatool.py
new file mode 100644
index 0000000..64f67aa
--- /dev/null
+++ b/rdatool.py
@@ -0,0 +1,290 @@
+import struct
+import logging
+import argparse
+import time
+import binascii
+from contextlib import contextmanager
+
+import serial
+
+PACKET_TRACE = 128
+PACKET_EVENT = 255
+
+CMD_RD_DWORD = 2
+CMD_RD_REG = 4
+CMD_WR_DWORD = 130
+CMD_WR_REG = 132
+
+REG_CHIP_ID = 0x1a24000
+REG_RESET_CAUSE = 0x1a000a0
+
+CTRL_CFG_REG = 3
+
+class TimeoutException(serial.SerialTimeoutException):
+    pass
+
+
+class RDAClient(serial.Serial):
+    """RDA8851 and friends ("Coolsand") programming interface library"""
+
+    def __init__(self, path, baudrate=921600, *args, **kwargs):
+        super(RDAClient, self).__init__(path, baudrate, *args, **kwargs)
+        self.trace_logger = logging.getLogger('rda.trace')
+
+    def run(self):
+        while True:
+            try:
+                self.handle_packet(*self.read_packet())
+            except KeyboardInterrupt:
+                return
+            except serial.SerialException:
+                raise
+            except:
+                logging.exception('Oops!')
+                time.sleep(0.01)
+
+    def wait_for_packet(self, pid, handle=False):
+        """Waits for single packet from device, handles every other packet when
+        needed"""
+        while True:
+            packet_id, data = self.read_packet()
+            if packet_id == pid:
+                return data
+
+            try:
+                if handle:
+                    self.handle_packet(packet_id, data)
+            except:
+                logging.exception('Oops!')
+
+    def wait_for_event(self, event, handle=False):
+        """Waits for specific event packet"""
+        while True:
+            data = self.wait_for_packet(PACKET_EVENT, handle)
+            if struct.unpack('<xI', data)[0] == event:
+                return True
+
+    def handle_packet(self, packet_type, data):
+        """Generic debugging packet handler"""
+        if packet_type == PACKET_TRACE:
+            traceline = data[2:data.find('\x00', 2)]
+            self.trace_logger.info(traceline)
+        elif packet_type == PACKET_EVENT:
+            logging.debug('Event: %08x', *struct.unpack('<xI', data))
+        elif packet_type == 131:
+            logging.debug('131: %d', *struct.unpack('<I', data))
+        elif packet_type == 162:
+            logging.info('162: %r %r', packet_type, data)
+        else:
+            logging.debug('Unhandled packet %d (%r)', packet_type, data)
+
+    def read_packet(self):
+        """Reads single packet from device, returns tuple with packet ID/type
+        and its data"""
+        preamble = self.read(1)
+        if preamble != '\xad':
+            raise Exception('Invalid preamble (got %r)' % preamble)
+
+        packet_len, = struct.unpack('>H', self.read_unescape(2))
+
+        if packet_len == 0 or packet_len == 1:
+            raise Exception('Empty packet?')
+
+        data = self.read_unescape(packet_len)
+
+        expected_checksum = ord(self.read_unescape(1))
+        checksum = self.calculate_checksum(data)
+        if checksum != expected_checksum:
+            raise Exception('Invalid checksum (%d expected, got %d)' % (expected_checksum, checksum))
+
+        logging.debug(' <- %s', binascii.hexlify(data))
+
+        return (ord(data[0]), data[1:])
+
+    def write_packet(self, packet_id, data):
+        """Sends single packet to device"""
+        data = chr(packet_id) + data
+        data_len = len(data)
+
+        logging.debug(' -> %s', binascii.hexlify(data))
+
+        payload = '\xad' + struct.pack('>H%dsB' % data_len, data_len, data, self.calculate_checksum(data))
+        self.write_escape(payload)
+
+    timeouts = []
+
+    @contextmanager
+    def do_timeout(self, t):
+        """Context manager that adds timeouts to blocking operations"""
+        self.timeouts.append(time.time() + t)
+        yield
+        self.timeouts.pop()
+
+    def check_timeout(self):
+        """Returns seconds left to earliest timeout"""
+        if not self.timeouts:
+            return None
+
+        t = min(self.timeouts) - time.time()
+
+        if t < 0:
+            raise TimeoutException()
+
+        return t
+
+    def _sleep(self, t):
+        """Sleep for t seconds or raise TimeoutException if not enough time left"""
+        if self.check_timeout() < t:
+            raise TimeoutException()
+
+        time.sleep(t)
+
+    def read(self, size):
+        """Reads bytes with timeout exception support"""
+        self.timeout = self.check_timeout()
+        buf = super(RDAClient, self).read(size)
+
+        if len(buf) != size:
+            raise TimeoutException()
+
+        return buf
+
+    def read_unescape(self, l):
+        """Reads and unescapes data"""
+        s = bytearray(self.read(l))
+        while True:
+            read_esc_count = s.count('\\') + l - len(s)
+            if read_esc_count == 0:
+                break
+            s.append(self.read(read_esc_count))
+
+        s = s.replace('\\\xee', '\x11')
+        s = s.replace('\\\xec', '\x13')
+        s = s.replace('\\\xa3', '\\')
+        return s
+
+    def write_escape(self, data):
+        """Escapes and writes data to device"""
+        data = data.replace('\\',   '\\\xa3') \
+                   .replace('\x11', '\\\xee') \
+                   .replace('\x13', '\\\xec')
+        self.write(data)
+
+    def calculate_checksum(self, data):
+        """Calculates XOR-based checksum"""
+        checksum = 0
+        for b in data:
+            checksum = checksum ^ ord(b)
+
+        return checksum
+
+    _cnt = 1
+    def read_value(self, cmd, addr, raw=False):
+        """Reads single value (either "reg" or dword from memory address)"""
+        self._cnt = (self._cnt % 255) + 1
+        self.write_packet(255, struct.pack('<BIB', cmd, addr, self._cnt))
+
+        while True:
+            data = self.wait_for_packet(PACKET_EVENT)
+            if ord(data[0]) == self._cnt:
+                if raw:
+                    return data[1:]
+                elif cmd == CMD_RD_DWORD:
+                    return struct.unpack('<I', data[1:])[0]
+                elif cmd == CMD_RD_REG:
+                    return struct.unpack('<B', data[1:])[0]
+
+    def read_reg(self, addr, raw=False):
+        """Helper function to read single reg"""
+        return self.read_value(CMD_RD_REG, addr, raw)
+
+    def read_dword(self, addr, raw=False):
+        """Helper function to read single memory dword"""
+        return self.read_value(CMD_RD_DWORD, addr, raw)
+
+    def read_block(self, addr, read_len):
+        """Reads memory block"""
+        data = ""
+
+        for offset in range(0, read_len, 4):
+            data += self.read_dword(addr + offset, raw=True)
+
+        return data
+
+    def write_value(self, cmd, addr, value):
+        """Writes single value (either "reg" or single memory dword)"""
+        if cmd == CMD_WR_REG:
+            payload = struct.pack('<BIB', cmd, addr, value)
+        elif cmd == CMD_WR_DWORD:
+            payload = struct.pack('<BII', cmd, addr, value)
+
+        self.write_packet(255, payload)
+
+    def write_reg(self, addr, value):
+        """Helper function to write reg value"""
+        return self.write_value(CMD_WR_REG, addr, value)
+
+    def write_dword(self, addr, value):
+        """Helper function to write single dword to memory"""
+        return self.write_value(CMD_WR_DWORD, addr, value)
+
+    # High level interface
+    def chipid(self):
+        return self.read_dword(REG_CHIP_ID)
+
+    def host_mode_enable(self):
+        self.write_dword(REG_RESET_CAUSE, 0x6a0000)
+        self.write_reg(0, 1)
+
+        self._sleep(1)
+
+        for i in range(3):
+            self.write_reg(CTRL_CFG_REG, 128)
+
+        self._sleep(1)
+        self.reset_input_buffer()
+
+        self.write_reg(1, 2)
+        self.write_reg(5, 253)
+
+        self.wait_for_event(0xff000001)
+
+        self.write_dword(REG_RESET_CAUSE, 0x600000)
+
+
+def cli_trace(client, args):
+    client.run()
+
+def cli_at(client, args):
+    at_cmd = 'AT+POWER=2,200\r'
+    client.write_packet(162, '\x04' + at_cmd + '\xc0')
+
+    client.run()
+
+def cli_chipid(client, args):
+    logging.info('Chip ID: %08x', client.chipid())
+
+parser = argparse.ArgumentParser()
+parser.add_argument('--verbose', '-v', action='count', default=0)
+parser.add_argument('--port', '-p', default='/dev/serial/by-id/usb-Silicon_Labs_CP2102_USB_to_UART_Bridge_Controller_0001-if00-port0')
+subparsers = parser.add_subparsers()
+
+trace_parser = subparsers.add_parser('trace', help='show trace')
+trace_parser.set_defaults(func=cli_trace)
+
+trace_parser = subparsers.add_parser('at', help='execute AT command')
+trace_parser.set_defaults(func=cli_at)
+
+trace_parser = subparsers.add_parser('chipid', help='show chip ID')
+trace_parser.set_defaults(func=cli_chipid)
+
+if __name__ == "__main__":
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO - 10*args.verbose)
+    client = RDAClient(args.port)
+    args.func(client, args)
+
+    #c = RDAClient('/dev/serial/by-id/usb-Silicon_Labs_CP2102_USB_to_UART_Bridge_Controller_0001-if00-port0') #'/dev/serial/by-id/usb-1a86_USB2.0-Serial-if00-port0')
+    #c = RDAClient('/dev/serial/by-id/usb-1a86_USB2.0-Serial-if00-port0')
+    #c.run()