contrib/rpc_examples/ocd_rpc_example.py

   1 #!/usr/bin/env python3
   2 """
   3 OpenOCD RPC example, covered by GNU GPLv3 or later
   4 Copyright (C) 2014 Andreas Ortmann (ortmann@finf.uni-hannover.de)
   5
   6
   7 Example output:
   8 ./ocd_rpc_example.py
   9 echo says hi!
  10
  11 target state: halted
  12 target halted due to debug-request, current mode: Thread
  13 xPSR: 0x01000000 pc: 0x00000188 msp: 0x10000fd8
  14
  15 variable @ 0x10000000: 0x01c9c380
  16
  17 variable @ 0x10000000: 0xdeadc0de
  18
  19 memory (before): ['0xdeadc0de', '0x00000011', '0xaaaaaaaa', '0x00000023',
  20 '0x00000042', '0x0000ffff']
  21
  22 memory (after): ['0x00000001', '0x00000000', '0xaaaaaaaa', '0x00000023',
  23 '0x00000042', '0x0000ffff']
  24 """
  25
  26 import socket
  27 import itertools
  28
  29 def strToHex(data):
  30     return map(strToHex, data) if isinstance(data, list) else int(data, 16)
  31
  32 def hexify(data):
  33     return "<None>" if data is None else ("0x%08x" % data)
  34
  35 def compareData(a, b):
  36     for i, j, num in zip(a, b, itertools.count(0)):
  37         if i != j:
  38             print("difference at %d: %s != %s" % (num, hexify(i), hexify(j)))
  39
  40
  41 class OpenOcd:
  42     COMMAND_TOKEN = '\x1a'
  43     def __init__(self, verbose=False):
  44         self.verbose = verbose
  45         self.tclRpcIp       = "127.0.0.1"
  46         self.tclRpcPort     = 6666
  47         self.bufferSize     = 4096
  48
  49         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  50
  51     def __enter__(self):
  52         self.connect()
  53         return self
  54
  55     def __exit__(self, type, value, traceback):
  56         self.disconnect()
  57
  58     def connect(self):
  59         self.sock.connect((self.tclRpcIp, self.tclRpcPort))
  60
  61     def disconnect(self):
  62         try:
  63             self.send("exit")
  64         finally:
  65             self.sock.close()
  66
  67     def send(self, cmd):
  68         """Send a command string to TCL RPC. Return the result that was read."""
  69         data = (cmd + OpenOcd.COMMAND_TOKEN).encode("utf-8")
  70         if self.verbose:
  71             print("<- ", data)
  72
  73         self.sock.send(data)
  74         return self._recv()
  75
  76     def _recv(self):
  77         """Read from the stream until the token (\x1a) was received."""
  78         data = bytes()
  79         while True:
  80             chunk = self.sock.recv(self.bufferSize)
  81             data += chunk
  82             if bytes(OpenOcd.COMMAND_TOKEN, encoding="utf-8") in chunk:
  83                 break
  84
  85         if self.verbose:
  86             print("-> ", data)
  87
  88         data = data.decode("utf-8").strip()
  89         data = data[:-1] # strip trailing \x1a
  90
  91         return data
  92
  93     def readVariable(self, address):
  94         raw = self.send("mdw 0x%x" % address).split(": ")
  95         return None if (len(raw) < 2) else strToHex(raw[1])
  96
  97     def readMemory(self, wordLen, address, n):
  98         self.send("array unset output") # better to clear the array before
  99         self.send("mem2array output %d 0x%x %d" % (wordLen, address, n))
 100
 101         output = [*map(int, self.send("return $output").split(" "))]
 102         d = dict([tuple(output[i:i + 2]) for i in range(0, len(output), 2)])
 103
 104         return [d[k] for k in sorted(d.keys())]
 105
 106     def writeVariable(self, address, value):
 107         assert value is not None
 108         self.send("mww 0x%x 0x%x" % (address, value))
 109
 110     def writeMemory(self, wordLen, address, n, data):
 111         array = " ".join(["%d 0x%x" % (a, b) for a, b in enumerate(data)])
 112
 113         self.send("array unset 1986ве1т") # better to clear the array before
 114         self.send("array set 1986ве1т { %s }" % array)
 115         self.send("array2mem 1986ве1т 0x%x %s %d" % (wordLen, address, n))
 116
 117 if __name__ == "__main__":
 118
 119     def show(*args):
 120         print(*args, end="\n\n")
 121
 122     with OpenOcd() as ocd:
 123         ocd.send("reset")
 124
 125         show(ocd.send("capture { echo \"echo says hi!\" }")[:-1])
 126         show(ocd.send("capture \"halt\"")[:-1])
 127
 128         # Read the first few words at the RAM region (put starting adress of RAM
 129         # region into 'addr')
 130         addr = 0x10000000
 131
 132         value = ocd.readVariable(addr)
 133         show("variable @ %s: %s" % (hexify(addr), hexify(value)))
 134
 135         ocd.writeVariable(addr, 0xdeadc0de)
 136         show("variable @ %s: %s" % (hexify(addr), hexify(ocd.readVariable(addr))))
 137
 138         data = [1, 0, 0xaaaaaaaa, 0x23, 0x42, 0xffff]
 139         wordlen = 32
 140         n = len(data)
 141
 142         read = ocd.readMemory(wordlen, addr, n)
 143         show("memory (before):", list(map(hexify, read)))
 144
 145         ocd.writeMemory(wordlen, addr, n, data)
 146
 147         read = ocd.readMemory(wordlen, addr, n)
 148         show("memory  (after):", list(map(hexify, read)))
 149
 150         compareData(read, data)
 151
 152         ocd.send("resume")