Files

1076 lines
42 KiB
Python
Raw Permalink Normal View History

2026-08-03 19:57:38 +03:00
#!/usr/bin/env python3
"""
MT6261D BROM Direct SPI NOR Flasher - 4MB Flash Version
Optimized for 4MB (32Mbit) SPI NOR flash
"""
import serial
import time
import struct
import hashlib
import argparse
import glob
import os
import sys
import json
from typing import Dict, List, Optional, Tuple, Union
# ==================== BROM PROTOCOL CONSTANTS ====================
# BootROM Commands
BROM_CMD_SYNC = 0xA0
BROM_CMD_START = 0x10
BROM_CMD_SEND_DATA = 0x11
BROM_CMD_END = 0x12
BROM_CMD_GET_CHIP_ID = 0x13
BROM_CMD_GET_HW_CODE = 0x14
BROM_CMD_JUMP_DA = 0x15
BROM_CMD_BAUD_CHANGE = 0x16
BROM_CMD_CRC32 = 0x17
BROM_CMD_READ_REG = 0x18
BROM_CMD_WRITE_REG = 0x19
BROM_CMD_ERASE_FLASH = 0x1A
BROM_CMD_READ_FLASH = 0x1B
BROM_CMD_WRITE_FLASH = 0x1C
BROM_CMD_GET_FLASH_INFO = 0x1D
BROM_CMD_SPI_CMD = 0x1E
BROM_CMD_NAND_CTRL = 0x1F
# Response codes
BROM_SUCCESS = 0x00
BROM_ERROR = 0x01
BROM_TIMEOUT = 0x02
BROM_CRC_ERROR = 0x03
BROM_INVALID_CMD = 0x04
BROM_FLASH_ERROR = 0x05
# ==================== SPI NOR FLASH COMMANDS ====================
# Standard SPI NOR commands
SPI_CMD_READ_ID = (0x9F, 0, 0, 3) # Read JEDEC ID
SPI_CMD_READ_STATUS = (0x05, 0, 0, 1) # Read Status Register
SPI_CMD_WRITE_STATUS = (0x01, 0, 0, 1) # Write Status Register
SPI_CMD_READ_DATA = (0x03, 3, 0, 0) # Read Data
SPI_CMD_FAST_READ = (0x0B, 3, 1, 0) # Fast Read
SPI_CMD_PAGE_PROGRAM = (0x02, 3, 0, 0) # Page Program
SPI_CMD_SECTOR_ERASE = (0x20, 3, 0, 0) # Sector Erase (4KB)
SPI_CMD_BLOCK_ERASE = (0xD8, 3, 0, 0) # Block Erase (64KB)
SPI_CMD_BLOCK_ERASE_32K = (0x52, 3, 0, 0) # Block Erase (32KB)
SPI_CMD_CHIP_ERASE = (0xC7, 0, 0, 0) # Chip Erase
SPI_CMD_WRITE_ENABLE = (0x06, 0, 0, 0) # Write Enable
SPI_CMD_WRITE_DISABLE = (0x04, 0, 0, 0) # Write Disable
SPI_CMD_BULK_ERASE = (0x60, 0, 0, 0) # Bulk Erase
# Status register bits
SR_WIP = 0x01 # Write In Progress
SR_WEL = 0x02 # Write Enable Latch
SR_BP0 = 0x04 # Block Protect 0
SR_BP1 = 0x08 # Block Protect 1
SR_BP2 = 0x10 # Block Protect 2
SR_BP3 = 0x20 # Block Protect 3
SR_QE = 0x40 # Quad Enable
SR_SRWD = 0x80 # Status Register Write Disable
# ==================== MT6261D MEMORY MAP ====================
# Bootloader addresses (these remain the same)
BOOTLOADER_BD_ADDRESS = 0x00000000 # Boot Region 1
BOOTLOADER_EB_ADDRESS = 0x00004000 # Extended Boot Region
BOOTLOADER_REGION3_ADDRESS = 0x00008000 # Third boot region
BOOTLOADER_END_ADDRESS = 0x0000C000 # End of boot regions
# Bootloader partition sizes
BOOT1_SIZE = BOOTLOADER_EB_ADDRESS - BOOTLOADER_BD_ADDRESS # 0x4000 (16KB)
BOOT2_SIZE = BOOTLOADER_REGION3_ADDRESS - BOOTLOADER_EB_ADDRESS # 0x4000 (16KB)
BOOT3_SIZE = BOOTLOADER_END_ADDRESS - BOOTLOADER_REGION3_ADDRESS # 0x4000 (16KB)
GAP_SIZE = 0x4000 # Gap between boot regions and main firmware
# 4MB Flash Configuration (changed from 16MB)
FLASH_SIZE_4MB = 0x400000 # 4MB
FLASH_SIZE_4MB_BITS = 32 # 32Mbit
FLASH_PAGE_SIZE = 0x100 # 256 bytes
FLASH_SECTOR_SIZE = 0x1000 # 4KB
FLASH_BLOCK_SIZE = 0x10000 # 64KB
FLASH_BLOCK_32K_SIZE = 0x8000 # 32KB blocks (some flash chips)
MAX_TRANSFER_SIZE = 0x1000 # 4KB max transfer (reduced for 4MB)
# 4MB Flash layout (typical MT6261D with 4MB flash)
# 0x000000 - 0x00FFFF : Bootloader area (64KB)
# 0x010000 - 0x01FFFF : System parameters (64KB)
# 0x020000 - 0x3FFFFF : Main firmware (3.875MB)
FLASH_LAYOUT_4MB = {
'bootloader': {'start': 0x000000, 'size': 0x10000, 'name': 'Bootloader'},
'sysparam': {'start': 0x010000, 'size': 0x10000, 'name': 'System Params'},
'main_firmware': {'start': 0x020000, 'size': 0x3E0000, 'name': 'Main Firmware'}
}
# Common 4MB SPI NOR Flash chips for MT6261D
SUPPORTED_FLASH_CHIPS = {
'EF4016': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'GD25Q32': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'W25Q32': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'MX25L3205': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'SST25VF032': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'EN25Q32': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'M25P32': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'IS25LQ032': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
'PM25LQ032': {'size': FLASH_SIZE_4MB, 'page': 0x100, 'sector': 0x1000, 'block': 0x10000},
}
# Exception Classes
class BROMFlashingError(Exception):
"""Custom exception for BROM flashing errors"""
pass
class FlashError(BROMFlashingError):
"""Flash memory specific error"""
pass
class SPIError(BROMFlashingError):
"""SPI communication error"""
pass
class VerificationError(BROMFlashingError):
"""Data verification error"""
pass
class UnsupportedFlashError(FlashError):
"""Unsupported flash chip error"""
pass
# ==================== BROM FLASHER CLASS ====================
class MT6261BROMFlasher:
def __init__(self, port: str, baudrate: int = 115200, timeout: int = 2):
"""Initialize the BROM flasher connection with retry support"""
self.port = port
self.baudrate = baudrate
self.timeout = timeout
self.serial = None
self.connected = False
self.chip_id = None
self.hw_code = None
self.flash_info = None
self.flash_size = FLASH_SIZE_4MB # Set to 4MB by default
self.flash_chip_name = None
self.block_size = FLASH_BLOCK_SIZE
self.page_size = FLASH_PAGE_SIZE
self.sector_size = FLASH_SECTOR_SIZE
self.current_baud = baudrate
self.is_4mb_flash = False
self._connect_port()
def _connect_port(self):
print(f"Waiting for serial port {self.port} or BootROM device...", flush=True)
start = time.time()
while time.time() - start < 600:
ports = glob.glob('/dev/ttyUSB*') + glob.glob('/dev/ttyACM*')
for p in ports:
try:
self.serial = serial.Serial(
port=p,
baudrate=self.baudrate,
timeout=self.timeout,
rtscts=True,
dsrdtr=True,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE
)
self.port = p
return
except (serial.SerialException, OSError):
pass
time.sleep(0.05)
raise serial.SerialException(f"Could not open serial port {self.port}")
def close(self):
"""Close the serial connection"""
if self.serial.is_open:
self.serial.close()
# ==================== BASIC COMMUNICATION ====================
def send_raw(self, data: bytes):
"""Send raw bytes to device"""
self.serial.write(data)
self.serial.flush()
def read_raw(self, length: int) -> bytes:
"""Read raw bytes from device"""
return self.serial.read(length)
def send_brom_command(self, command: int, data: bytes = b''):
"""Send a BootROM command"""
length = len(data)
packet = struct.pack('<BHB', command, length, 0x00) + data
self.serial.write(packet)
self.serial.flush()
def receive_brom_response(self) -> Tuple[int, int, bytes]:
"""Receive and parse BootROM response"""
header = self.serial.read(4)
if len(header) < 4:
raise BROMFlashingError("Failed to receive BROM response header")
status, length_lo, length_hi = struct.unpack('<BHB', header)
response_data = b''
if length_lo > 0:
response_data = self.serial.read(length_lo)
return status, length_lo, response_data
def check_brom_error(self, status: int):
"""Check for errors in BROM response"""
if status == BROM_TIMEOUT:
raise BROMFlashingError("BROM timeout")
if status == BROM_CRC_ERROR:
raise BROMFlashingError("CRC error in BROM communication")
if status == BROM_FLASH_ERROR:
raise FlashError("Flash operation failed")
if status == BROM_INVALID_CMD:
raise BROMFlashingError("Invalid BROM command")
if status != BROM_SUCCESS:
raise BROMFlashingError(f"BROM error: status 0x{status:02X}")
def calculate_crc32(self, data: bytes) -> int:
"""Calculate CRC32 for data verification"""
import binascii
return binascii.crc32(data) & 0xFFFFFFFF
# ==================== FLASH DETECTION ====================
def detect_flash_chip(self) -> Tuple[str, int]:
"""Detect flash chip and verify 4MB size"""
print("\nDetecting flash chip...")
# Read JEDEC ID
jedec_id = self.read_flash_id()
if len(jedec_id) >= 3:
manufacturer = jedec_id[0]
device_id = jedec_id[1] << 8 | jedec_id[2]
print(f"JEDEC ID: {jedec_id.hex()}")
print(f"Manufacturer: 0x{manufacturer:02X}")
print(f"Device ID: 0x{device_id:04X}")
# Try to identify the chip
for name, info in SUPPORTED_FLASH_CHIPS.items():
# This is simplified - actual ID matching would be more specific
if manufacturer in [0x01, 0x1C, 0x20, 0xC2, 0xEF, 0x9D]: # Known manufacturers
if info['size'] == FLASH_SIZE_4MB:
self.flash_chip_name = name
self.flash_size = info['size']
self.page_size = info['page']
self.sector_size = info['sector']
self.block_size = info['block']
self.is_4mb_flash = True
print(f"Detected: {name} (4MB)")
return name, self.flash_size
# If not in supported list, use default 4MB settings
print(f"Unknown flash chip (MFR:0x{manufacturer:02X}, ID:0x{device_id:04X})")
print("Assuming 4MB flash based on device type")
self.flash_size = FLASH_SIZE_4MB
self.is_4mb_flash = True
return "Unknown", self.flash_size
else:
# No JEDEC ID response, assume 4MB
print("No JEDEC ID response - assuming 4MB flash")
self.flash_size = FLASH_SIZE_4MB
self.is_4mb_flash = True
return "Unknown", self.flash_size
def validate_address_range(self, address: int, size: int):
"""Validate that address and size are within 4MB flash bounds"""
if address < 0 or address >= self.flash_size:
raise FlashError(f"Address 0x{address:08X} outside 4MB flash range")
if size < 0:
raise FlashError("Invalid size")
if address + size > self.flash_size:
raise FlashError(f"Address range 0x{address:08X} + 0x{size:X} exceeds 4MB flash")
def check_boot_region_bounds(self, address: int, size: int):
"""Validate boot region is within safe bounds (top of flash)"""
# Boot regions should be within the first 64KB
if address + size > 0x10000:
raise FlashError(f"Boot region exceeds 64KB boundary: 0x{address:08X} + 0x{size:X}")
print(f"Boot region valid: 0x{address:08X} to 0x{address + size:08X}")
# ==================== BOOTROM MODE ENTRY ====================
def wait_for_brom_mode(self, timeout: int = 600) -> bool:
"""Wait for device to enter MT6261 BootROM mode"""
start_time = time.time()
print("\n" + "="*60, flush=True)
print("Waiting for MT6261 BootROM Mode", flush=True)
print("="*60, flush=True)
print("1. Turn OFF phone", flush=True)
print("2. Hold BOOT key and connect USB cable now...", flush=True)
dot_count = 0
while time.time() - start_time < timeout:
try:
self.send_raw(b'\xA0')
time.sleep(0.005)
if self.serial.in_waiting > 0:
b = self.serial.read(1)
if b and b[0] == 0x5F:
self.send_raw(b'\x0A\x50\x05')
ack3 = self.serial.read(3)
if ack3 == b'\xF5\xAF\xFA':
print("\nConnected to MT6261 BootROM!", flush=True)
self.connected = True
# Read Chip ID via 0xA2 command
cmd = b'\xA2' + struct.pack('>II', 0x80000008, 1)
self.send_raw(cmd)
echo = self.serial.read(len(cmd))
resp = self.serial.read(2)
if len(resp) == 2:
self.chip_id = struct.unpack('>H', resp)[0]
print(f"Chip ID: 0x{self.chip_id:04X}", flush=True)
# Disable watchdogs and enable USB download
self.write_reg16(0xa0030000, 0x2200) # disable system watchdog
self.write_reg16(0xa0700a28, 0x8000) # enable USB download mode
self.write_reg16(0xa0700a24, 2) # disable battery watchdog
self.write_reg32(0xa0510000, 2) # memory map mode 2
print("BootROM initialized & watchdogs disabled", flush=True)
return True
except (serial.SerialException, OSError):
pass
dot_count += 1
if dot_count % 20 == 0:
sys.stdout.write(".")
sys.stdout.flush()
time.sleep(0.01)
return False
def write_reg16(self, addr: int, val: int):
cmd = b'\xD2' + struct.pack('>II', addr, 1)
self.send_raw(cmd)
echo = self.serial.read(len(cmd))
status = self.serial.read(2)
data = struct.pack('>H', val)
self.send_raw(data)
echo2 = self.serial.read(len(data))
status2 = self.serial.read(2)
def write_reg32(self, addr: int, val: int):
cmd = b'\xD4' + struct.pack('>II', addr, 1)
self.send_raw(cmd)
echo = self.serial.read(len(cmd))
status = self.serial.read(2)
data = struct.pack('>I', val)
self.send_raw(data)
echo2 = self.serial.read(len(data))
status2 = self.serial.read(2)
def initialize_flash(self):
"""Initialize and detect SPI NOR flash"""
print("\nInitializing SPI NOR Flash...")
# Get flash info from BROM
try:
self.send_brom_command(BROM_CMD_GET_FLASH_INFO)
status, length, data = self.receive_brom_response()
self.check_brom_error(status)
if len(data) >= 12:
# Parse flash info
brom_flash_size = struct.unpack('<I', data[0:4])[0]
# Check if BROM reports correct size
print(f"BROM reports flash size: 0x{brom_flash_size:X} ({brom_flash_size // (1024*1024)} MB)")
if brom_flash_size == FLASH_SIZE_4MB:
self.flash_size = brom_flash_size
self.is_4mb_flash = True
else:
print(f"WARNING: BROM reports {brom_flash_size // (1024*1024)}MB but expecting 4MB")
print("Will use detected size from BROM")
self.flash_size = brom_flash_size
except:
print("Could not get flash info from BROM")
# Detect flash chip
flash_name, detected_size = self.detect_flash_chip()
if detected_size == FLASH_SIZE_4MB:
print(f"\n[OK] 4MB Flash confirmed: {flash_name}")
print(f" Size: {self.flash_size} bytes ({self.flash_size // (1024*1024)} MB)")
print(f" Page: 0x{self.page_size:X} ({self.page_size} bytes)")
print(f" Sector: 0x{self.sector_size:X} ({self.sector_size} bytes)")
print(f" Block: 0x{self.block_size:X} ({self.block_size // 1024} KB)")
else:
raise UnsupportedFlashError(f"Unsupported flash size: {detected_size}")
# Print flash layout for 4MB
print(f"\n4MB Flash Layout:")
print(f" Bootloader: 0x000000 - 0x00FFFF ({FLASH_LAYOUT_4MB['bootloader']['size'] // 1024} KB)")
print(f" System Parms: 0x010000 - 0x01FFFF ({FLASH_LAYOUT_4MB['sysparam']['size'] // 1024} KB)")
print(f" Main FW: 0x020000 - 0x3FFFFF ({FLASH_LAYOUT_4MB['main_firmware']['size'] // (1024*1024)} MB)")
self.flash_info = {
'size': self.flash_size,
'page_size': self.page_size,
'sector_size': self.sector_size,
'block_size': self.block_size,
'chip_name': flash_name,
'is_4mb': self.is_4mb_flash
}
def read_reg32(self, addr: int) -> int:
cmd = b'\xD1' + struct.pack('>II', addr, 1)
self.send_raw(cmd)
echo = self.serial.read(len(cmd))
resp = self.serial.read(8)
if len(resp) >= 8:
return struct.unpack('<I', resp[2:6])[0]
return 0
def sfi_mac_cmd_write(self, cmd_byte: int, addr: Optional[int] = None, data: bytes = b''):
"""Send SPI flash command via SFI MAC mode (matching DZ09 sfi.c exact sequence)"""
SFI_BASE = 0xA0140000
SFI_MAC_CTL = SFI_BASE + 0x0000
SFI_MAC_OUTL = SFI_BASE + 0x0010
SFI_MAC_INL = SFI_BASE + 0x0014
SFI_MISC_CTL3 = SFI_BASE + 0x0044
SFI_GPRAM = SFI_BASE + 0x0800
payload = bytes([cmd_byte])
if addr is not None:
payload += addr.to_bytes(3, 'big')
payload += data
total_len = len(payload)
# 1. Write payload to GPRAM
for i in range(0, total_len, 4):
chunk = payload[i:i+4]
val = int.from_bytes(chunk.ljust(4, b'\x00'), 'little')
self.write_reg32(SFI_GPRAM + i, val)
# 2. Mask AHB ch2
self.write_reg32(SFI_MISC_CTL3, self.read_reg32(SFI_MISC_CTL3) | (1 << 9))
# 3. Enable MAC mode first (MAC_SEL bit 28 | MAC_EN bit 3)
mac_val = (1 << 28) | (1 << 3)
self.write_reg32(SFI_MAC_CTL, mac_val)
# 4. Set OUTL and INL while MAC is enabled
self.write_reg32(SFI_MAC_OUTL, total_len)
self.write_reg32(SFI_MAC_INL, 0)
# 5. Trigger operation (TRIG bit 2)
self.write_reg32(SFI_MAC_CTL, mac_val | (1 << 2))
# 6. Wait for completion (WIP_READY bit 1 set, WIP bit 0 clear)
for _ in range(200):
v = self.read_reg32(SFI_MAC_CTL)
if (v & 0x02) and not (v & 0x01):
break
time.sleep(0.001)
# 7. Clean up MAC mode
self.write_reg32(SFI_MAC_CTL, 0)
self.write_reg32(SFI_MISC_CTL3, self.read_reg32(SFI_MISC_CTL3) & ~(1 << 9))
def sfi_mac_cmd_read(self, cmd_byte: int, read_len: int) -> bytes:
"""Send SPI flash read command via SFI MAC mode (matching DZ09 sfi.c exact sequence)"""
SFI_BASE = 0xA0140000
SFI_MAC_CTL = SFI_BASE + 0x0000
SFI_MAC_OUTL = SFI_BASE + 0x0010
SFI_MAC_INL = SFI_BASE + 0x0014
SFI_MISC_CTL3 = SFI_BASE + 0x0044
SFI_GPRAM = SFI_BASE + 0x0800
# 1. Write command to GPRAM
self.write_reg32(SFI_GPRAM, cmd_byte)
# 2. Mask AHB ch2
self.write_reg32(SFI_MISC_CTL3, self.read_reg32(SFI_MISC_CTL3) | (1 << 9))
# 3. Enable MAC mode first
mac_val = (1 << 28) | (1 << 3)
self.write_reg32(SFI_MAC_CTL, mac_val)
# 4. Set OUTL and INL while MAC is enabled
self.write_reg32(SFI_MAC_OUTL, 1)
self.write_reg32(SFI_MAC_INL, read_len)
# 5. Trigger operation
self.write_reg32(SFI_MAC_CTL, mac_val | (1 << 2))
# 6. Wait for completion
for _ in range(200):
v = self.read_reg32(SFI_MAC_CTL)
if (v & 0x02) and not (v & 0x01):
break
time.sleep(0.001)
# 7. Read response from GPRAM
total_bytes = 1 + read_len
res_bytes = b''
for i in range(0, total_bytes + 3, 4):
val = self.read_reg32(SFI_GPRAM + i)
res_bytes += struct.pack('<I', val)
# 8. Clean up MAC mode
self.write_reg32(SFI_MAC_CTL, 0)
self.write_reg32(SFI_MISC_CTL3, self.read_reg32(SFI_MISC_CTL3) & ~(1 << 9))
return res_bytes[1:1+read_len]
def read_flash_id(self) -> bytes:
"""Read JEDEC ID from SPI NOR flash"""
return self.sfi_mac_cmd_read(0x9F, 3)
# ==================== SPI NOR FLASH OPERATIONS ====================
def read_status_register(self) -> int:
"""Read SPI NOR flash status register"""
data = self.sfi_mac_cmd_read(0x05, 1)
return data[0] if len(data) >= 1 else 0
def write_status_register(self, value: int):
"""Write SPI NOR flash status register"""
self.sfi_mac_cmd_write(0x50) # Write enable for status register
self.sfi_mac_cmd_write(0x01, data=bytes([value & ~(SR_BP0 | SR_BP1 | SR_BP2 | SR_BP3)]))
self.wait_until_ready()
def wait_until_ready(self, timeout: int = 10):
"""Wait until flash is ready (WIP bit cleared)"""
start_time = time.time()
while time.time() - start_time < timeout:
sr = self.read_status_register()
if not (sr & SR_WIP):
return
time.sleep(0.005)
raise FlashError("Timeout waiting for flash to be ready")
def check_write_protection(self):
"""Check and disable write protection"""
sr = self.read_status_register()
print(f"SPI Flash Status Register: 0x{sr:02X}")
if sr & (SR_BP0 | SR_BP1 | SR_BP2 | SR_BP3):
print("Write protection enabled - disabling...")
self.write_status_register(0x00)
sr = self.read_status_register()
print(f"Status Register after unlocking: 0x{sr:02X}")
if sr & (SR_BP0 | SR_BP1 | SR_BP2 | SR_BP3):
raise FlashError("Could not disable write protection")
print("Write protection disabled")
def enable_write(self):
"""Enable write operation (WEL bit)"""
self.sfi_mac_cmd_write(0x06)
def disable_write(self):
"""Disable write operation"""
self.sfi_mac_cmd_write(0x04)
def erase_sector(self, address: int):
"""Erase a sector (4KB) at address"""
self.validate_address_range(address, self.sector_size)
print(f"Erasing sector at 0x{address:08X}...")
self.wait_until_ready()
self.enable_write()
self.sfi_mac_cmd_write(0x20, addr=address)
self.wait_until_ready()
print("Sector erased")
def erase_block(self, address: int):
"""Erase a block (64KB) at address"""
self.validate_address_range(address, self.block_size)
print(f"Erasing block at 0x{address:08X}...")
self.wait_until_ready()
self.enable_write()
self.sfi_mac_cmd_write(0xD8, addr=address)
self.wait_until_ready()
print("Block erased")
def page_program(self, address: int, data: bytes):
"""Program data within a single page (chunked to fit 160-byte SFI_GPRAM)"""
self.validate_address_range(address, len(data))
# Max chunk size is 64 bytes payload (command 1 + addr 3 + data 64 = 68 bytes <= 160 SFI_GPRAM)
max_chunk = 64
offset = 0
while offset < len(data):
chunk = data[offset:offset+max_chunk]
curr_addr = address + offset
self.wait_until_ready()
self.enable_write()
self.sfi_mac_cmd_write(0x02, addr=curr_addr, data=chunk)
self.wait_until_ready()
offset += len(chunk)
def read_flash(self, address: int, length: int) -> bytes:
"""Read data from flash memory via BROM 0xD1 memory mapped read"""
self.validate_address_range(address, length)
data = b''
remaining = length
curr_addr = address
blk_size = 1024
while remaining > 0:
rsize = min(remaining, blk_size)
# Send 0xD1 command for sz words
words_cnt = rsize >> 2
cmd = b'\xD1' + struct.pack('>II', curr_addr, words_cnt)
self.send_raw(cmd)
echo = self.serial.read(len(cmd))
resp = self.serial.read((words_cnt * 4) + 4)
if len(resp) >= (words_cnt * 4) + 4:
# Unpack words (skip 2-byte header status and 2-byte footer status)
words = struct.unpack('<' + words_cnt * 'I', resp[2:-2])
chunk_data = struct.pack('>' + words_cnt * 'I', *words)
data += chunk_data
else:
raise FlashError(f"Short read from BROM at 0x{curr_addr:08X}")
curr_addr += rsize
remaining -= rsize
progress = (length - remaining) / length * 100
sys.stdout.write(f"\rReading: {progress:.1f}%")
sys.stdout.flush()
print("\nRead complete")
return data
def write_flash(self, address: int, data: bytes):
"""Write data to flash memory (handles page boundaries)"""
self.validate_address_range(address, len(data))
self.check_write_protection()
print(f"Writing {len(data)} bytes to 0x{address:08X}...")
offset = 0
curr_addr = address
remaining = len(data)
while remaining > 0:
# Calculate page boundary (256 bytes)
page_offset = curr_addr % self.page_size
chunk_size = min(self.page_size - page_offset, remaining)
# Check if we're at a page boundary (full page write)
if page_offset == 0:
# Check if we have a full page
if chunk_size == self.page_size:
# Direct page program
self.page_program(curr_addr, data[offset:offset + chunk_size])
else:
# Partial page at end - read-modify-write
page_start = curr_addr
existing_data = self.read_flash(page_start, self.page_size)
new_page = data[offset:offset + chunk_size] + existing_data[len(data[offset:offset + chunk_size]):]
self.page_program(curr_addr, new_page)
else:
# Cross-page boundary - read-modify-write
page_start = curr_addr - page_offset
existing_data = self.read_flash(page_start, self.page_size)
# Modify existing page data
new_page = bytearray(existing_data)
new_page[page_offset:page_offset + chunk_size] = data[offset:offset + chunk_size]
# Program modified page
self.page_program(page_start, bytes(new_page))
offset += chunk_size
curr_addr += chunk_size
remaining -= chunk_size
progress = offset / len(data) * 100
sys.stdout.write(f"\rWriting: {progress:.1f}%")
sys.stdout.flush()
print("\nWrite complete")
def verify_flash(self, address: int, data: bytes) -> bool:
"""Verify data written to flash"""
self.validate_address_range(address, len(data))
print(f"Verifying {len(data)} bytes at 0x{address:08X}...")
read_data = self.read_flash(address, len(data))
if read_data == data:
print("Verification PASSED")
return True
else:
# Find all mismatches
mismatches = []
for i in range(min(len(data), len(read_data))):
if data[i] != read_data[i]:
mismatches.append((i, data[i], read_data[i]))
if len(mismatches) > 10:
break
if len(mismatches) > 0:
print(f"Verification FAILED ({len(mismatches)}+ mismatches)")
for offset, expected, got in mismatches[:5]:
print(f" Offset 0x{offset:04X}: Expected 0x{expected:02X}, Got 0x{got:02X}")
else:
if len(data) != len(read_data):
print(f"Length mismatch: expected {len(data)}, got {len(read_data)}")
else:
print("Verification PASSED")
return True
return False
# ==================== HIGH-LEVEL OPERATIONS ====================
def flash_bootloader_region(self, address: int, boot_data: bytes):
"""Flash a bootloader region with proper handling"""
# Validate boot region is within safe bounds (top 64KB)
self.check_boot_region_bounds(address, len(boot_data))
print(f"\n{'='*60}")
print(f"Flashing boot region at 0x{address:08X}")
print(f"Size: {len(boot_data)} bytes")
print(f"{'='*60}")
# Check and unlock SPI NOR write protection
self.check_write_protection()
# Check if we need to erase (if data is not all 0xFF)
if all(b == 0xFF for b in boot_data):
print("Data is all 0xFF - skipping (already erased)")
else:
# Erase the required sectors
start_sector = address & ~(self.sector_size - 1)
end_addr = address + len(boot_data)
curr_sector = start_sector
while curr_sector < end_addr:
self.erase_sector(curr_sector)
curr_sector += self.sector_size
# Write data
self.write_flash(address, boot_data)
# Verify
if not self.verify_flash(address, boot_data):
raise VerificationError(f"Boot region verification failed at 0x{address:08X}")
print(f"Boot region at 0x{address:08X} flashed successfully")
def flash_full_dump(self, flash_dump_path: str, verify: bool = True):
"""Flash a complete 4MB flash dump"""
print(f"\n{'='*60}")
print("Flashing Full 4MB Flash Dump")
print(f"{'='*60}")
# Check file exists
if not os.path.exists(flash_dump_path):
raise BROMFlashingError(f"Flash dump file not found: {flash_dump_path}")
# Check file size
file_size = os.path.getsize(flash_dump_path)
print(f"Flash dump file size: {file_size} bytes")
if file_size != self.flash_size:
raise BROMFlashingError(
f"Flash dump size mismatch: expected {self.flash_size}, got {file_size}"
)
# Read flash dump
with open(flash_dump_path, 'rb') as f:
flash_data = f.read()
print(f"MD5: {hashlib.md5(flash_data).hexdigest()}")
# Check if we should do full chip erase or selective
print("\nOptions:")
print("1. Full chip erase (slow but safest)")
print("2. Selective erase (faster but riskier)")
choice = input("Select option (1 or 2): ").strip()
if choice == "1":
print("Performing full chip erase...")
self.chip_erase()
# Write entire flash
self.write_flash(0x000000, flash_data)
else:
# Selective erase: only erase regions that need changing
print("Performing selective erase...")
# Check which sectors need erasing
sectors_to_erase = []
for sector_start in range(0, self.flash_size, self.sector_size):
sector_data = flash_data[sector_start:sector_start + self.sector_size]
if not all(b == 0xFF for b in sector_data):
# Read current sector
current_data = self.read_flash(sector_start, self.sector_size)
if current_data != sector_data:
sectors_to_erase.append(sector_start)
print(f"Sectors to erase: {len(sectors_to_erase)}")
# Erase needed sectors
for sector in sectors_to_erase:
self.erase_sector(sector)
# Write entire flash
self.write_flash(0x000000, flash_data)
# Verify if requested
if verify:
print("\nVerifying entire flash...")
read_data = self.read_flash(0x000000, self.flash_size)
if read_data == flash_data:
print("Full flash verification PASSED")
else:
# Find mismatches
mismatches = sum(1 for i in range(self.flash_size) if read_data[i] != flash_data[i])
print(f"Full flash verification FAILED ({mismatches} mismatches)")
raise VerificationError("Full flash verification failed")
print(f"\nFull flash dump written successfully!")
# ==================== BAUDRATE MANAGEMENT ====================
def change_baudrate(self, new_baud: int):
"""Change communication baudrate"""
print(f"Changing baudrate to {new_baud}...")
self.send_brom_command(BROM_CMD_BAUD_CHANGE, struct.pack('<I', new_baud))
status, _, _ = self.receive_brom_response()
self.check_brom_error(status)
self.serial.baudrate = new_baud
self.current_baud = new_baud
time.sleep(0.2)
# Verify connection
self.send_brom_command(BROM_CMD_GET_CHIP_ID)
status, _, _ = self.receive_brom_response()
self.check_brom_error(status)
print(f"Baudrate changed to {new_baud}")
# ==================== FILE OPERATIONS ====================
def extract_boot_regions_from_flash(flash_dump_path: str, output_dir: str = "./bootloaders") -> List[Dict]:
"""Extract boot regions from a full flash dump"""
print("\n" + "="*60)
print("Extracting Boot Regions from Full Flash Dump")
print("="*60)
if not os.path.exists(flash_dump_path):
raise BROMFlashingError(f"Flash dump file not found: {flash_dump_path}")
# Check file size
file_size = os.path.getsize(flash_dump_path)
if file_size != FLASH_SIZE_4MB:
print(f"WARNING: Flash dump is {file_size} bytes, expected {FLASH_SIZE_4MB} for 4MB flash")
with open(flash_dump_path, 'rb') as f:
flash_data = f.read()
print(f"Flash dump size: {len(flash_data)} bytes ({len(flash_data) // (1024*1024)} MB)")
os.makedirs(output_dir, exist_ok=True)
boot_regions = [
('boot_region1', BOOTLOADER_BD_ADDRESS, BOOT1_SIZE, 'boot_region1.bin'),
('boot_region2', BOOTLOADER_EB_ADDRESS, BOOT2_SIZE, 'boot_region2.bin'),
('boot_region3', BOOTLOADER_REGION3_ADDRESS, BOOT3_SIZE, 'boot_region3.bin')
]
extracted_files = []
for name, start, size, filename in boot_regions:
if start + size > len(flash_data):
print(f"Warning: {name} extends beyond flash dump")
continue
region_data = flash_data[start:start+size]
filepath = os.path.join(output_dir, filename)
with open(filepath, 'wb') as f:
f.write(region_data)
info = {
'name': name,
'address': start,
'size': size,
'file': filepath,
'md5': hashlib.md5(region_data).hexdigest()
}
extracted_files.append(info)
print(f"\nExtracted {name}:")
print(f" Address: 0x{start:08X}")
print(f" Size: 0x{size:X} ({size} bytes)")
print(f" File: {filepath}")
print(f" MD5: {info['md5']}")
# Save extraction info
info_file = os.path.join(output_dir, 'extraction_info.json')
with open(info_file, 'w') as f:
json.dump({
'flash_dump': flash_dump_path,
'flash_size': FLASH_SIZE_4MB,
'extracted_regions': extracted_files
}, f, indent=2)
print(f"\nExtraction info saved to {info_file}")
return extracted_files
# ==================== MAIN FUNCTION ====================
def main():
parser = argparse.ArgumentParser(description='MT6261D BROM Direct SPI NOR Flasher - 4MB Flash')
# Required
parser.add_argument('--port', default='auto', help='Serial port (or auto)')
# Optional
parser.add_argument('--baud', type=int, default=115200, help='Initial baud rate')
parser.add_argument('--high-baud', type=int, help='High speed baud rate')
# Operation modes
parser.add_argument('--flash-dump', help='Full 4MB flash dump file')
parser.add_argument('--boot1', help='Bootloader region 1 file')
parser.add_argument('--boot2', help='Bootloader region 2 file')
parser.add_argument('--boot3', help='Bootloader region 3 file')
parser.add_argument('--extract-only', action='store_true', help='Only extract boot regions')
parser.add_argument('--read-backup', help='Read flash data to file')
parser.add_argument('--read-address', type=lambda x: int(x, 0), help='Address to read from')
parser.add_argument('--read-size', type=lambda x: int(x, 0), help='Number of bytes to read')
parser.add_argument('--verify', action='store_true', help='Verify after write')
# Advanced
parser.add_argument('--chip-erase', action='store_true', help='Erase entire chip')
parser.add_argument('--force-4mb', action='store_true', help='Force 4MB flash size detection')
parser.add_argument('--verbose', action='store_true', help='Enable verbose output')
args = parser.parse_args()
# Validate
if not args.flash_dump and not args.boot1 and not args.boot2 and not args.boot3 and \
not args.read_backup and not args.extract_only and not args.chip_erase:
print("Error: No operation specified")
sys.exit(1)
port_to_use = args.port
if port_to_use == 'auto':
ports = glob.glob('/dev/ttyUSB*') + glob.glob('/dev/ttyACM*')
if ports:
port_to_use = ports[0]
print(f"Auto-detected serial port: {port_to_use}", flush=True)
else:
port_to_use = '/dev/ttyUSB0'
print(f"No port active yet, defaulting to {port_to_use} wait loop...", flush=True)
flasher = None
try:
# Initialize
print(f"Initializing on {port_to_use} at {args.baud} baud...")
flasher = MT6261BROMFlasher(port_to_use, args.baud)
# Force 4MB if requested
if args.force_4mb:
flasher.flash_size = FLASH_SIZE_4MB
flasher.is_4mb_flash = True
# Wait for BootROM
if not flasher.wait_for_brom_mode():
raise BROMFlashingError("Failed to enter BootROM mode")
# Switch to high baud if requested
if args.high_baud and args.high_baud > args.baud:
flasher.change_baudrate(args.high_baud)
# Verify flash size
flasher.is_4mb_flash = True
flasher.flash_size = FLASH_SIZE_4MB
# Print flash info
print(f"\nFlash Information:")
print(f" Size: {flasher.flash_size} bytes ({flasher.flash_size // (1024*1024)} MB)")
print(f" Page Size: 0x{flasher.page_size:X}")
print(f" Sector Size: 0x{flasher.sector_size:X}")
print(f" Block Size: 0x{flasher.block_size:X}")
# Perform operations
if args.read_backup:
# Read flash data
if args.read_address is None or args.read_size is None:
print("Error: --read-backup requires --read-address and --read-size")
sys.exit(1)
# Validate address range
flasher.validate_address_range(args.read_address, args.read_size)
print(f"\nReading 0x{args.read_size:X} bytes from 0x{args.read_address:08X}...")
data = flasher.read_flash(args.read_address, args.read_size)
with open(args.read_backup, 'wb') as f:
f.write(data)
print(f"Data saved to {args.read_backup}")
print(f"MD5: {hashlib.md5(data).hexdigest()}")
elif args.chip_erase:
# Chip erase
flasher.chip_erase()
elif args.extract_only and args.flash_dump:
# Extract only
extract_boot_regions_from_flash(args.flash_dump)
elif args.flash_dump:
# Flash full dump
flasher.flash_full_dump(args.flash_dump, verify=args.verify)
else:
# Flash individual boot files
boot_files = [
(BOOTLOADER_BD_ADDRESS, args.boot1, 'Boot Region 1'),
(BOOTLOADER_EB_ADDRESS, args.boot2, 'Boot Region 2'),
(BOOTLOADER_REGION3_ADDRESS, args.boot3, 'Boot Region 3')
]
for address, filepath, name in boot_files:
if filepath and os.path.exists(filepath):
print(f"\nFlashing {name}...")
with open(filepath, 'rb') as f:
data = f.read()
flasher.flash_bootloader_region(address, data)
elif filepath:
print(f"Error: File not found: {filepath}")
sys.exit(1)
print("\n" + "="*60)
print("Operation completed successfully!")
print("="*60)
except (BROMFlashingError, FlashError, SPIError, VerificationError, UnsupportedFlashError) as e:
print(f"\nError: {e}")
sys.exit(1)
except serial.SerialException as e:
print(f"\nSerial error: {e}")
sys.exit(1)
except KeyboardInterrupt:
print("\nCancelled by user")
sys.exit(1)
finally:
if flasher:
flasher.close()
print("Serial connection closed")
if __name__ == "__main__":
main()