contrib/rpc_examples/ocd_rpc_example.py

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