116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""Legacy BALZAM/TMS parallel-memory dump protocol.
|
|
|
|
The TMS320F2812 addresses external XINTF memory in 16-bit words, while
|
|
``CMD_UPLOAD`` transfers a byte count. This module keeps those units explicit
|
|
so a complete dump cannot silently become half or twice the requested size.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from protocan import tms2812 as shared_tms2812
|
|
|
|
CMD_UPLOAD = 52
|
|
DEFAULT_CONTROLLER_ADDRESS = 5
|
|
PARALLEL_NVM_START_WORD = 0x100000
|
|
PARALLEL_NVM_WORDS = 0x80000
|
|
PARALLEL_NVM_BYTES = PARALLEL_NVM_WORDS * 2
|
|
MT29F1G08_WORDS = 0x4000000
|
|
MT29F1G08_BYTES = MT29F1G08_WORDS * 2
|
|
DEFAULT_CHUNK_BYTES = 4096
|
|
NAND_CHUNK_BYTES = 2048
|
|
UPLOAD_TRAILER_BYTES = 6 # CRC16 + four legacy padding bytes
|
|
|
|
|
|
class TmsFlashError(ValueError):
|
|
"""Malformed command, reply, or dump range."""
|
|
|
|
|
|
def build_upload_request(address: int, word_address: int, byte_count: int) -> bytes:
|
|
"""Build the 12-byte legacy ``CMD_UPLOAD`` request."""
|
|
if not 0 <= address <= 0xFF:
|
|
raise TmsFlashError("адрес контроллера вне диапазона 0..255")
|
|
if not 0 <= word_address <= 0xFFFFFFFF:
|
|
raise TmsFlashError("адрес слова вне диапазона uint32")
|
|
if not 1 <= byte_count <= 0xFFFFFFFF:
|
|
raise TmsFlashError("размер блока вне диапазона 1..0xFFFFFFFF")
|
|
try:
|
|
return shared_tms2812.build_upload_request(
|
|
address, word_address, byte_count
|
|
)
|
|
except ValueError as error:
|
|
raise TmsFlashError(str(error)) from error
|
|
|
|
|
|
def decode_upload_reply(raw: bytes, address: int, byte_count: int) -> bytes:
|
|
"""Validate one complete upload reply and return its memory bytes."""
|
|
expected = 2 + byte_count + UPLOAD_TRAILER_BYTES
|
|
if len(raw) != expected:
|
|
raise TmsFlashError(
|
|
f"ответ CMD_UPLOAD: ожидалось {expected} байт, получено {len(raw)}"
|
|
)
|
|
try:
|
|
return shared_tms2812.decode_upload_reply(raw, address, byte_count)
|
|
except ValueError as error:
|
|
raise TmsFlashError(str(error)) from error
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DumpChunk:
|
|
"""One request in a word-addressed, byte-transferred dump."""
|
|
|
|
word_address: int
|
|
byte_count: int
|
|
|
|
|
|
class FlashDumpPlan:
|
|
"""Deterministic chunk plan used by the GUI and host tests."""
|
|
|
|
def __init__(self, start_word: int, word_count: int, chunk_bytes: int) -> None:
|
|
if start_word < 0 or start_word > 0xFFFFFFFF:
|
|
raise TmsFlashError("начальный адрес слова вне диапазона uint32")
|
|
if word_count <= 0:
|
|
raise TmsFlashError("число слов должно быть положительным")
|
|
if start_word + word_count > 0x100000000:
|
|
raise TmsFlashError("конец диапазона выходит за uint32")
|
|
if chunk_bytes < 2 or chunk_bytes > 0xFFFF or chunk_bytes % 2:
|
|
raise TmsFlashError("блок должен быть чётным числом байт 2..65534")
|
|
self.start_word = start_word
|
|
self.word_count = word_count
|
|
self.chunk_bytes = chunk_bytes
|
|
self.total_bytes = word_count * 2
|
|
|
|
def chunks(self) -> list[DumpChunk]:
|
|
result = []
|
|
offset = 0
|
|
while offset < self.total_bytes:
|
|
size = min(self.chunk_bytes, self.total_bytes - offset)
|
|
result.append(DumpChunk(self.start_word + offset // 2, size))
|
|
offset += size
|
|
return result
|
|
|
|
|
|
class UploadReplyBuffer:
|
|
"""Collect arbitrary serial fragments for exactly one upload reply."""
|
|
|
|
def __init__(self, address: int, byte_count: int) -> None:
|
|
self.address = address
|
|
self.byte_count = byte_count
|
|
self.expected_size = 2 + byte_count + UPLOAD_TRAILER_BYTES
|
|
self._buffer = bytearray()
|
|
|
|
@property
|
|
def received_size(self) -> int:
|
|
return len(self._buffer)
|
|
|
|
def feed(self, data: bytes) -> tuple[bytes | None, bytes]:
|
|
"""Return ``(payload, trailing_data)`` when the reply is complete."""
|
|
self._buffer.extend(data)
|
|
if len(self._buffer) < self.expected_size:
|
|
return None, b""
|
|
raw = bytes(self._buffer[: self.expected_size])
|
|
trailing = bytes(self._buffer[self.expected_size :])
|
|
self._buffer.clear()
|
|
return decode_upload_reply(raw, self.address, self.byte_count), trailing
|