byun/esp32s3_canlogger_mcp2515
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

수정

Browse Source
This commit is contained in:
byun
2026-03-10 23:04:52 +00:00
parent e251acd6e3
commit 314d05b15e
1 changed files with 71 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
can_converter_gui.py
View File

@@ -32,22 +32,42 @@ from PyQt5.QtGui import QFont, QTextCursor, QIcon
# ============================================================================ # ============================================================================
class CANMessage: class CANMessage:
"""ESP32 CAN Logger의 CAN 메시지 구조체""" """ESP32 CAN Logger의 CAN 메시지 구조체
STRUCT_FORMAT = '<QIB8s' # little-endian: uint64, uint32, uint8, 8*uint8
STRUCT_SIZE = struct.calcsize(STRUCT_FORMAT)
def __init__(self, timestamp_us: int, can_id: int, dlc: int, data: bytes): 펌웨어 구조체 (22 bytes, packed):
uint64_t timestamp_us (8 bytes)
uint32_t id (4 bytes)
uint8_t dlc (1 byte)
uint8_t flags (1 byte) ← RingBuffer 버전에서 추가 (EFF/RTR 보존)
uint8_t data[8] (8 bytes)
구버전 (21 bytes, flags 없음):
STRUCT_FORMAT_V1 = '<QIB8s'
"""
STRUCT_FORMAT = '<QIBb8s' # 22 bytes (현재 펌웨어 - flags 포함)
STRUCT_FORMAT_V1 = '<QIB8s' # 21 bytes (구버전 - flags 없음)
STRUCT_SIZE = struct.calcsize(STRUCT_FORMAT) # 22
STRUCT_SIZE_V1 = struct.calcsize(STRUCT_FORMAT_V1) # 21
def __init__(self, timestamp_us: int, can_id: int, dlc: int, data: bytes, flags: int = 0):
self.timestamp_us = timestamp_us self.timestamp_us = timestamp_us
self.can_id = can_id self.can_id = can_id
self.dlc = dlc self.dlc = dlc
self.flags = flags # bit7=EFF, bit6=RTR
self.data = data[:dlc] self.data = data[:dlc]
@classmethod @classmethod
def from_bytes(cls, data: bytes) -> 'CANMessage': def from_bytes(cls, data: bytes) -> 'CANMessage':
"""바이너리 데이터에서 CANMessage 파싱""" """바이너리 데이터에서 CANMessage 파싱 (22/21 bytes 자동 감지)"""
timestamp_us, can_id, dlc, msg_data = struct.unpack(cls.STRUCT_FORMAT, data) if len(data) == cls.STRUCT_SIZE:
return cls(timestamp_us, can_id, dlc, msg_data) timestamp_us, can_id, dlc, flags, msg_data = struct.unpack(cls.STRUCT_FORMAT, data)
return cls(timestamp_us, can_id, dlc, msg_data, flags)
elif len(data) == cls.STRUCT_SIZE_V1:
timestamp_us, can_id, dlc, msg_data = struct.unpack(cls.STRUCT_FORMAT_V1, data)
return cls(timestamp_us, can_id, dlc, msg_data, 0)
else:
raise ValueError(f"Invalid data size: {len(data)} (expected 22 or 21)")
def __repr__(self): def __repr__(self):
return f"CANMessage(id=0x{self.can_id:X}, dlc={self.dlc}, ts={self.timestamp_us})" return f"CANMessage(id=0x{self.can_id:X}, dlc={self.dlc}, ts={self.timestamp_us})"
# ============================================================================ # ============================================================================
@@ -69,22 +89,40 @@ class CANBinToMDF:
print(f" - 시그널 수: {sum(len(msg.signals) for msg in self.db.messages)}") print(f" - 시그널 수: {sum(len(msg.signals) for msg in self.db.messages)}")
def read_bin_file(self, bin_path: str) -> List[CANMessage]: def read_bin_file(self, bin_path: str) -> List[CANMessage]:
"""ESP32 CAN Logger의 bin 파일 읽기""" """ESP32 CAN Logger의 bin 파일 읽기 (22/21 bytes 자동 감지)"""
messages = [] messages = []
with open(bin_path, 'rb') as f: with open(bin_path, 'rb') as f:
data = f.read() data = f.read()
if not data:
print(f"⚠️ 파일이 비어있습니다: {bin_path}")
return messages
# 버전 자동 감지: 22 bytes(flags 포함) vs 21 bytes(구버전)
# 파일 크기가 22의 배수면 신버전, 21의 배수면 구버전
is_v2 = (len(data) % CANMessage.STRUCT_SIZE == 0)
is_v1 = (len(data) % CANMessage.STRUCT_SIZE_V1 == 0)
if is_v2:
record_size = CANMessage.STRUCT_SIZE # 22
print(f"✓ 펌웨어 버전: RingBuffer (22 bytes/record, flags 포함)")
elif is_v1:
record_size = CANMessage.STRUCT_SIZE_V1 # 21
print(f"✓ 펌웨어 버전: 구버전 (21 bytes/record)")
else:
# 22 bytes 우선 시도
record_size = CANMessage.STRUCT_SIZE
print(f"⚠️ 파일 크기({len(data)})가 22/21의 배수가 아님, 22 bytes로 파싱 시도")
offset = 0 offset = 0
while offset + CANMessage.STRUCT_SIZE <= len(data): while offset + record_size <= len(data):
msg_bytes = data[offset:offset + CANMessage.STRUCT_SIZE] msg_bytes = data[offset:offset + record_size]
try: try:
msg = CANMessage.from_bytes(msg_bytes) msg = CANMessage.from_bytes(msg_bytes)
messages.append(msg) messages.append(msg)
except Exception as e: except Exception as e:
print(f"⚠️ 메시지 파싱 오류 at offset {offset}: {e}") print(f"⚠️ 메시지 파싱 오류 at offset {offset}: {e}")
offset += record_size
offset += CANMessage.STRUCT_SIZE
print(f"✓ bin 파일 읽기 완료: {bin_path}") print(f"✓ bin 파일 읽기 완료: {bin_path}")
print(f" - 총 메시지: {len(messages)}") print(f" - 총 메시지: {len(messages)}")
@@ -286,7 +324,7 @@ class CANBinToCSV:
print(f" - 시그널 수: {sum(len(msg.signals) for msg in self.db.messages)}") print(f" - 시그널 수: {sum(len(msg.signals) for msg in self.db.messages)}")
def read_bin_file(self, bin_path: str) -> List[CANMessage]: def read_bin_file(self, bin_path: str) -> List[CANMessage]:
"""ESP32 CAN Logger의 bin 파일 읽기""" """ESP32 CAN Logger의 bin 파일 읽기 (22/21 bytes 자동 감지)"""
messages = [] messages = []
try: try:
@@ -298,20 +336,33 @@ class CANBinToCSV:
return messages return messages
print(f" 파일 크기: {len(data)} bytes") print(f" 파일 크기: {len(data)} bytes")
# 버전 자동 감지: 22 bytes(flags 포함) vs 21 bytes(구버전)
is_v2 = (len(data) % CANMessage.STRUCT_SIZE == 0)
is_v1 = (len(data) % CANMessage.STRUCT_SIZE_V1 == 0)
if is_v2:
record_size = CANMessage.STRUCT_SIZE
print(f" 펌웨어 버전: RingBuffer (22 bytes/record)")
elif is_v1:
record_size = CANMessage.STRUCT_SIZE_V1
print(f" 펌웨어 버전: 구버전 (21 bytes/record)")
else:
record_size = CANMessage.STRUCT_SIZE
print(f"⚠️ 파일 크기가 22/21의 배수가 아님, 22 bytes로 파싱 시도")
offset = 0 offset = 0
parse_errors = 0 parse_errors = 0
while offset + CANMessage.STRUCT_SIZE <= len(data): while offset + record_size <= len(data):
msg_bytes = data[offset:offset + CANMessage.STRUCT_SIZE] msg_bytes = data[offset:offset + record_size]
try: try:
msg = CANMessage.from_bytes(msg_bytes) msg = CANMessage.from_bytes(msg_bytes)
messages.append(msg) messages.append(msg)
except Exception as e: except Exception as e:
parse_errors += 1 parse_errors += 1
if parse_errors <= 5: # 처음 5개 에러만 출력 if parse_errors <= 5:
print(f"⚠️ 메시지 파싱 오류 at offset {offset}: {e}") print(f"⚠️ 메시지 파싱 오류 at offset {offset}: {e}")
offset += CANMessage.STRUCT_SIZE offset += record_size
if parse_errors > 5: if parse_errors > 5:
print(f"⚠️ 총 {parse_errors}개의 파싱 오류 발생") print(f"⚠️ 총 {parse_errors}개의 파싱 오류 발생")