e71aee73260d0aeda780608d83570d8fb0f51d01
[renesas-ra-flasher] / src / RAFlasher.py
1 import os
2 import math
3 import time
4 import argparse
5 import tempfile
6 from tqdm import tqdm
7 from RAConnect import *
8 from RAPacker import *
9
10 SECTOR_SIZE = 2048
11
12 def int_to_hex_list(num):
13     hex_string = hex(num)[2:].upper()  # convert to hex string
14     hex_string = hex_string.zfill(8) # pad for 8 char's long
15     hex_list = [f'0x{hex_string[c:c+2]}' for c in range(0, 8, 2)]
16     return hex_list
17
18 def hex_type(string):
19     try:
20         value = int(string, 16)
21         return value
22     except ValueError:
23         raise argparse.ArgumentTypeError(f"'{string}' is not a valid hexadecimal value.")
24
25 def set_size_boundaries(start_addr, size):
26     if start_addr % SECTOR_SIZE:
27         raise ValueError(f"start addr not aligned on sector size {SECTOR_SIZE}")
28
29     if size < SECTOR_SIZE:
30         print("Warning: you are trying to write something that is less than one sector size: padding with zeroes")
31
32     blocks = (size + SECTOR_SIZE - 1) // SECTOR_SIZE
33     end_addr = blocks * SECTOR_SIZE + start_addr - 1
34
35     if (end_addr <= start_addr):
36         raise ValueError(f"End address smaller or equal than start_address")
37
38     if (end_addr > 0x3FFFF):
39         raise ValueError(f"Binary file is bigger than available ROM space")
40
41     return (start_addr, end_addr)
42
43 def inquire_connection(dev):
44     packed = pack_pkt(INQ_CMD, "")
45     dev.send_data(packed)
46     info = dev.recv_data(7)
47     if info == bytearray(b'\x00') or info == bytearray(b''):
48         return False
49     msg = unpack_pkt(info)
50     #print("Connection already established")
51     return True
52
53 def get_area_info(dev):
54     for i in [0,1,2]:
55         packed = pack_pkt(ARE_CMD, [str(i)])
56         dev.send_data(packed)
57         info = dev.recv_data(23)
58         msg = unpack_pkt(info)
59         fmt = '>BIIII'
60         KOA, SAD, EAD, EAU, WAU = struct.unpack(fmt, bytes(int(x, 16) for x in msg))
61         print(f'Area {KOA}: {hex(SAD)}:{hex(EAD)} (erase {hex(EAU)} - write {hex(WAU)})')
62
63 def get_dev_info(dev):
64     packed = pack_pkt(SIG_CMD, "")
65     dev.send_data(packed)
66     info = dev.recv_data(18)
67     fmt = '>IIIBBHH'
68     _HEADER, SCI, RMB, NOA, TYP, BFV, _FOOTER = struct.unpack(fmt, info)
69     print('====================')
70     if TYP == 0x02:
71         print('Chip: RA MCU + RA2/RA4 Series')
72     elif TYP == 0x03:
73         print('Chip: RA MCU + RA6 Series')
74     else:
75         rint('Unknown MCU type')
76     print(f'Serial interface speed: {SCI} Hz')
77     print(f'Recommend max UART baud rate: {RMB} bps')
78     print(f'User area in Code flash [{NOA & 0x1}|{NOA & 0x02 >> 1}]')
79     print(f'User area in Data flash [{NOA & 0x03 >> 2}]')
80     print(f'Config area [{NOA & 0x04 >> 3}]')
81     print(f'Boot firmware: version {BFV >> 8}.{BFV & 0xFF}')
82
83 def erase_chip(dev, start_addr, size):
84     
85     if size == None:
86         size = 0x3FFFF - start_addr # erase all
87     
88     (start_addr, end_addr) = set_size_boundaries(start_addr, size)
89     print(f'Erasing {hex(start_addr)}:{hex(end_addr)}')
90
91     # setup initial communication
92     SAD = int_to_hex_list(start_addr)
93     EAD = int_to_hex_list(end_addr)
94     packed = pack_pkt(ERA_CMD, SAD + EAD)
95     dev.send_data(packed)
96     
97     ret = dev.recv_data(7, timeout=1000) # erase takes usually a bit longer
98     unpack_pkt(ret) 
99     print("Erase complete")
100
101 def read_img(dev, img, start_addr, size):
102     
103     if size == None:
104         size = 0x3FFFF - start_addr # read maximum possible
105     
106     (start_addr, end_addr) = set_size_boundaries(start_addr, size)
107
108     # setup initial communication
109     SAD = int_to_hex_list(start_addr)
110     EAD = int_to_hex_list(end_addr)
111     packed = pack_pkt(REA_CMD, SAD + EAD)
112     dev.send_data(packed)
113
114     # calculate how many packets are have to be received
115     nr_packet = (end_addr - start_addr) // 1024 # TODO: set other than just 1024
116
117     with open(img, 'wb') as f:
118         for i in tqdm(range(0, nr_packet+1), desc="Reading progress"):
119             ret = dev.recv_data(1024 + 6)
120             chunk = unpack_pkt(ret)
121             chunky = bytes(int(x, 16) for x in chunk)
122             f.write(chunky)
123             packed = pack_pkt(REA_CMD, ['0x00'], ack=True)
124             dev.send_data(packed)
125
126
127 def write_img(dev, img, start_addr, size, verify=False):
128
129     if os.path.exists(img):
130         file_size = os.path.getsize(img)
131     else:
132         raise Exception(f'file {img} does not exist')
133
134     if size == None:
135         size = file_size
136
137     if size > file_size:
138         raise ValueError("Write size > file size")
139
140     (start_addr, end_addr) = set_size_boundaries(start_addr, size)
141     
142     chunk_size = 1024 # max is 1024 according to protocol
143
144     # setup initial communication
145     SAD = int_to_hex_list(start_addr)
146     EAD = int_to_hex_list(end_addr)
147     packed = pack_pkt(WRI_CMD, SAD + EAD)
148     dev.send_data(packed)
149     ret = dev.recv_data(7)
150     unpack_pkt(ret) 
151
152     totalread = 0
153     with open(img, 'rb') as f:
154         with tqdm(total=size, desc="Writing progress") as pbar:
155             chunk = f.read(chunk_size)
156             pbar.update(len(chunk))
157             while chunk and totalread < size:
158                 if len(chunk) != chunk_size:
159                     padding_length = chunk_size - len(chunk)
160                     chunk += b'\0' * padding_length
161                 packed = pack_pkt(WRI_CMD, chunk, ack=True)
162                 dev.send_data(packed)
163                 reply_len = 7
164                 reply = dev.recv_data(reply_len)
165                 msg = unpack_pkt(reply)
166                 chunk = f.read(chunk_size)
167                 totalread += chunk_size
168                 pbar.update(len(chunk))
169
170     
171     if verify:
172         with tempfile.NamedTemporaryFile(prefix='.hidden_', delete=False) as tmp_file, open(img, 'rb') as cmp_file:
173             read_img(dev, tmp_file.name, start_addr, size)
174             c1 = tmp_file.read(file_size) # due to byte alignment read file is longer
175             c2 = cmp_file.read()
176             if c1 == c2:
177                 print("Verify complete")
178             else: 
179                 print("Verify failed")
180
181 def main():
182     parser = argparse.ArgumentParser(description="RA Flasher Tool")
183
184     subparsers = parser.add_subparsers(dest="command", title="Commands")
185
186     # Subparser for the write command
187     write_parser = subparsers.add_parser("write", help="Write data to flash")
188     write_parser.add_argument("--start_address", type=hex_type, default='0x0000', help="Start address")
189     write_parser.add_argument("--size", type=hex_type, default=None, help="Size in bytes")
190     write_parser.add_argument("--verify", action="store_true", help="Verify after writing")
191     write_parser.add_argument("file_name", type=str, help="File name")
192
193     # Subparser for the read command
194     read_parser = subparsers.add_parser("read", help="Read data from flash")
195     read_parser.add_argument("--start_address", type=hex_type, default='0x0000', help="Start address")
196     read_parser.add_argument("--size", type=hex_type, default=None, help="Size in bytes")
197     read_parser.add_argument("file_name", type=str, help="File name")
198
199     erase_parser = subparsers.add_parser("erase", help="Erase sectors")
200     erase_parser.add_argument("--start_address", default='0x0000', type=hex_type, help="Start address")
201     erase_parser.add_argument("--size", type=hex_type, help="Size")
202
203     # Subparser for the info command
204     subparsers.add_parser("info", help="Show flasher information")
205
206     args = parser.parse_args()
207
208     if args.command == "write":
209         dev = RAConnect(vendor_id=0x045B, product_id=0x0261)
210         status_con = inquire_connection(dev)
211         if not status_con:
212             dev.confirm_connection()
213         write_img(dev, args.file_name, args.start_address, args.size, args.verify)
214     elif args.command == "read":
215         dev = RAConnect(vendor_id=0x045B, product_id=0x0261)
216         status_con = inquire_connection(dev)
217         if not status_con:
218             dev.confirm_connection()
219         read_img(dev, args.file_name, args.start_address, args.size)
220     elif args.command == "erase":
221         dev = RAConnect(vendor_id=0x045B, product_id=0x0261)
222         status_con = inquire_connection(dev)
223         if not status_con:
224             dev.confirm_connection()
225         erase_chip(dev, args.start_address, args.size)
226     elif args.command == "info":
227         dev = RAConnect(vendor_id=0x045B, product_id=0x0261)
228         status_con = inquire_connection(dev)
229         if not status_con:
230             dev.confirm_connection()
231         get_dev_info(dev)
232         get_area_info(dev)
233     else:
234         parser.print_help()
235
236 if __name__ == "__main__":
237     main()
238