Добавить загрузку F28335 по ProtoCAN и сервисные команды клиента
This commit is contained in:
@@ -37,6 +37,8 @@ class CanBootTransfer:
|
||||
def __init__(self, image, target: CanBootTarget):
|
||||
if not 0 < len(image.data) <= 512 * 1024:
|
||||
raise ValueError("ProtoCAN Boot: размер образа 1…512 КиБ")
|
||||
if target.product_type == 0x2833 and len(image.data) > 0x1FE00:
|
||||
raise ValueError("F28335 M95M02 A/B: образ не должен превышать 130560 байт")
|
||||
if not 0 <= image.version <= 0xFFFFFFFF:
|
||||
raise ValueError("Версия образа вне диапазона u32")
|
||||
self.image, self.target = image, target
|
||||
@@ -46,6 +48,10 @@ class CanBootTransfer:
|
||||
self.finished = False
|
||||
self._window_end = 0
|
||||
|
||||
@property
|
||||
def response_timeout_ms(self):
|
||||
return 15000 if self.stage in ("enter", "verify", "commit") else 1500
|
||||
|
||||
@property
|
||||
def percent(self):
|
||||
return min(100, self.next_block * 8 * 100 // len(self.image.data))
|
||||
@@ -128,4 +134,93 @@ class CanBootTransfer:
|
||||
self.stage = "reboot"
|
||||
return self._control(9), "Перезапуск прибора"
|
||||
self.stage, self.finished = "finished", True
|
||||
if self.target.product_type == 0x2833:
|
||||
return [], "Образ принят; после перезапуска приложение подтвердит пробный запуск"
|
||||
return [], "Прошивка ProtoCAN Boot завершена"
|
||||
|
||||
|
||||
class CanBootServices:
|
||||
"""Read-only passport queries and explicit session service commands.
|
||||
|
||||
Replies use message D, body=(page<<8)|command. IDENTIFY/discovery reply body=1.
|
||||
All multibyte values are little endian; version is major:8/minor:8/patch:16.
|
||||
"""
|
||||
def __init__(self, target: CanBootTarget):
|
||||
self.target = target
|
||||
|
||||
def request(self, command: int, page: int = 0):
|
||||
if command not in (1, 8, 11, 12, 13, 14, 15):
|
||||
raise ValueError("Unsupported service command")
|
||||
if not 0 <= page <= 255 or (page and command not in (14, 15)):
|
||||
raise ValueError("Invalid service page")
|
||||
t = self.target
|
||||
payload = bytes([page]) if command in (14, 15) else b""
|
||||
return build_frame(ProtoCanId.build(1, 0, t.device_type, t.device, 9,
|
||||
t.session_id << 8 | command), payload, to_can=True)
|
||||
|
||||
@staticmethod
|
||||
def discover():
|
||||
return build_frame(ProtoCanId.build(1, 0, 7, 15, 13, 0), b"", to_can=True)
|
||||
|
||||
def decode(self, frame, *, discovery=False):
|
||||
p = ProtoCanId.parse(frame.can_id)
|
||||
if (frame.to_can or not frame.ide or frame.rtr or frame.is_error or p.pm != 1
|
||||
or p.msg_type not in (12, 13) or len(frame.data) != 8):
|
||||
return None
|
||||
if discovery:
|
||||
if p.msg_type != 13 or p.body != 1:
|
||||
return None
|
||||
elif (p.device_type, p.device) != (self.target.device_type, self.target.device):
|
||||
return None
|
||||
if p.msg_type == 12:
|
||||
if p.body >> 8 != self.target.session_id:
|
||||
return None
|
||||
status, slot, next_block, crc32 = struct.unpack("<BBHI", frame.data)
|
||||
return {"kind": "status", "device_type": p.device_type, "device": p.device,
|
||||
"command": p.body & 255, "status": status, "target_slot": slot,
|
||||
"next_block": next_block, "crc32": crc32}
|
||||
command, page = p.body & 255, p.body >> 8
|
||||
d = frame.data
|
||||
result = {"kind": "passport", "device_type": p.device_type, "device": p.device, "command": command, "page": page}
|
||||
if command == 1 and page == 0:
|
||||
product, hardware, protocol, version = struct.unpack("<HBBI", d)
|
||||
result.update(product_type=product, hardware_revision=hardware, protocol_version=protocol,
|
||||
firmware_version=version, version=(version >> 24, (version >> 16) & 255, version & 65535))
|
||||
elif command == 12 and page == 0:
|
||||
result.update(protocol_version=d[0], state=d[1], active_slot=d[2], target_slot=d[3],
|
||||
session_id=d[4], slot_count=d[5], last_status=d[6])
|
||||
elif command == 13 and page == 0:
|
||||
flags, capacity = struct.unpack("<II", d)
|
||||
result.update(capabilities=flags, slot_capacity=capacity)
|
||||
elif command == 14:
|
||||
result["name_fragment"] = d.split(b"\0", 1)[0].decode("ascii", errors="replace")
|
||||
elif command == 15:
|
||||
if page == 0:
|
||||
size, page_size, mode, running = struct.unpack("<IHBB", d)
|
||||
result.update(memory_bytes=size, page_bytes=page_size, bootloader=bool(mode), running_slot=running)
|
||||
elif page == 1:
|
||||
result["slot_offsets"] = struct.unpack("<II", d)
|
||||
elif page == 2:
|
||||
active, pending, attempted, healthy, timeout = struct.unpack("<BBBBI", d)
|
||||
result.update(active_slot=active, pending_slot=pending, trial_attempted=bool(attempted),
|
||||
storage_healthy=bool(healthy), session_timeout_ms=timeout)
|
||||
elif page == 3:
|
||||
result["parameters_offset"], result["parameters_bytes"] = struct.unpack("<II", d)
|
||||
else:
|
||||
return None
|
||||
else:
|
||||
return None
|
||||
return result
|
||||
|
||||
|
||||
class CanBootPingExchange:
|
||||
"""PING adapter with the same request/feed shape as the SETProtocol client."""
|
||||
def __init__(self, target: CanBootTarget):
|
||||
self.services = CanBootServices(target)
|
||||
|
||||
def request(self):
|
||||
return [self.services.request(12)]
|
||||
|
||||
def feed(self, frame):
|
||||
reply = self.services.decode(frame)
|
||||
return reply if reply and reply["kind"] == "passport" and reply["command"] == 12 else None
|
||||
|
||||
70
python/tests/test_can_boot_services.py
Normal file
70
python/tests/test_can_boot_services.py
Normal file
@@ -0,0 +1,70 @@
|
||||
from types import SimpleNamespace
|
||||
import struct
|
||||
import pytest
|
||||
|
||||
from protocan.can_boot import CanBootServices, CanBootTarget, CanBootTransfer, CanBootPingExchange
|
||||
from protocan.protocan import ProtoCanId
|
||||
from protocan.transport import build_frame
|
||||
|
||||
|
||||
def response(command, data, device=7):
|
||||
return build_frame(ProtoCanId.build(1, 1, 3, device, 13, command), data, to_can=False)
|
||||
|
||||
|
||||
def test_passport_and_discovery_address_filters():
|
||||
service = CanBootServices(CanBootTarget(device_type=3, device=7, product_type=0x2833))
|
||||
identify = response(1, struct.pack('<HBBI', 0x2833, 1, 1, 0x01010002))
|
||||
assert service.decode(identify)['version'] == (1, 1, 2)
|
||||
foreign = response(1, identify.data, device=8)
|
||||
assert service.decode(foreign) is None
|
||||
assert service.decode(foreign, discovery=True)['device'] == 8
|
||||
capabilities = service.decode(response(13, struct.pack('<II', 0xFFF, 130560)))
|
||||
assert capabilities['slot_capacity'] == 130560
|
||||
assert capabilities['capabilities'] & (1 << 8)
|
||||
assert service.decode(response(14, b'F28335-M'))['name_fragment'] == 'F28335-M'
|
||||
assert service.decode(response(0x10E, b'95M02\0\0\0'))['name_fragment'] == '95M02'
|
||||
memory = service.decode(response(15, struct.pack('<IHBB', 262144, 256, 1, 255)))
|
||||
assert memory['bootloader'] and memory['memory_bytes'] == 262144
|
||||
assert service.decode(response(0x10F, struct.pack('<II', 512, 131072)))['slot_offsets'] == (512, 131072)
|
||||
pending = service.decode(response(0x20F, struct.pack('<BBBBI', 0, 1, 1, 1, 30000)))
|
||||
assert pending['trial_attempted'] and pending['pending_slot'] == 1
|
||||
assert service.decode(response(0x30F, struct.pack('<II', 0x3FEFC, 258)))['parameters_bytes'] == 258
|
||||
assert service.decode(response(0x40F, bytes(8))) is None
|
||||
discover = ProtoCanId.parse(service.discover().can_id)
|
||||
assert (discover.device_type, discover.device, discover.msg_type, discover.body) == (7, 15, 13, 0)
|
||||
|
||||
|
||||
def test_service_requests_and_transfer_limits():
|
||||
target = CanBootTarget(device_type=3, device=7, product_type=0x2833, session_id=15)
|
||||
service = CanBootServices(target)
|
||||
request = service.request(15, 2)
|
||||
assert request.data == b'\x02' and ProtoCanId.parse(request.can_id).body == 0xF0F
|
||||
assert ProtoCanId.parse(service.request(8).can_id).body == 0xF08
|
||||
with pytest.raises(ValueError):
|
||||
service.request(5)
|
||||
with pytest.raises(ValueError):
|
||||
service.request(1, 3)
|
||||
with pytest.raises(ValueError):
|
||||
CanBootTransfer(SimpleNamespace(data=bytes(130561), version=1), target)
|
||||
transfer = CanBootTransfer(SimpleNamespace(data=bytes(40), version=1), target)
|
||||
assert transfer.response_timeout_ms == 15000
|
||||
transfer.stage = 'data'
|
||||
assert transfer.response_timeout_ms == 1500
|
||||
|
||||
|
||||
def test_ping_exchange_only_accepts_addressed_ping_reply():
|
||||
ping = CanBootPingExchange(CanBootTarget(device_type=3, device=7))
|
||||
assert ProtoCanId.parse(ping.request()[0].can_id).body & 255 == 12
|
||||
assert ping.feed(response(1, bytes(8))) is None
|
||||
assert ping.feed(response(12, bytes([1, 0, 0, 255, 0, 2, 0, 0]), device=8)) is None
|
||||
assert ping.feed(response(12, bytes([1, 0, 0, 255, 0, 2, 0, 0])))['slot_count'] == 2
|
||||
|
||||
|
||||
def test_confirm_and_progress_status_correlate_session():
|
||||
service = CanBootServices(CanBootTarget(device_type=3, device=7, session_id=5))
|
||||
frame = build_frame(ProtoCanId.build(1, 1, 3, 7, 12, 0x508),
|
||||
struct.pack('<BBHI', 0, 1, 123, 0x12345678), to_can=False)
|
||||
reply = service.decode(frame)
|
||||
assert reply['status'] == 0 and reply['command'] == 8 and reply['next_block'] == 123
|
||||
assert service.decode(frame, discovery=True) is None
|
||||
assert CanBootServices(CanBootTarget(device_type=3, device=7, session_id=6)).decode(frame) is None
|
||||
Reference in New Issue
Block a user