NanoPB에 통신 짬때리기 (Pico-SDK + NanoPB + CRC32 + COBS)

// ${데이터종류},{데이터1},{데이터2}\r\n char *token = strtok(arr, ","); token = strtok(NULL, ","); if (token == NULL) { break; } strncpy(type_str, token, 4); switch(type_str[0]) { case 'W': type = FOO_W; break; case 'L': type = FOO_L; break; default: type = FOO_N break; } token = strtok(NULL, ","); if (token == NULL) { break; } strncpy(boo, token, 16);

$ git clone --branch nanopb-0.4.9.1 --depth 1 https://github.com/nanopb/nanopb.git $ python3 -m venv venv $ source venv/bin/activate (venv) $ pip install -r nanopb/requirements.txt (venv) $ python nanopb/generator/nanopb_generator.py message.proto

static void s_sender_task(void *arg) { Example data; char buff_temp[SND_BUFF_SIZE]; char buff_out[SND_BUFF_SIZE]; data.value = 0; for (;;) { // protobuff encoding pb_ostream_t stream = pb_ostream_from_buffer(buff_temp, SND_BUFF_SIZE - 4); // 4 bytes for crc bool pb_res = pb_encode(&stream, &Example_msg, &data); if (pb_res == false) { continue; } // CRC32 calculation uint32_t crc = crc32_calc(buff_temp, stream.bytes_written); memcpy(&buff_temp[stream.bytes_written], &crc, sizeof(crc)); // COBS encoding cobs_encode_result cobs_res = cobs_encode(buff_out, SND_BUFF_SIZE - 1, buff_temp, stream.bytes_written + 4); if (cobs_res.status != COBS_ENCODE_OK) { continue; } buff_out[cobs_res.out_len] = '\0'; // add NULL cobs_res.out_len++; // serial transmit serial_write(buff_out, cobs_res.out_len); data.value++; vTaskDelay(pdMS_TO_TICKS(1000)); } }

typedef struct { uint8_t len; } receiver_task_queue_t; static QueueHandle_t receiver_task_queue_hd = NULL; static void s_serial_callback(size_t toread_len) { if (receiver_task_queue_hd == NULL) { return; } receiver_task_queue_t qdata = {.len = toread_len}; xQueueSend(receiver_task_queue_hd, &qdata, 0); }

static void s_receiver_task(void *arg) { Example data; receiver_task_queue_t qdata; char buff_input[RCV_BUFF_SIZE]; char buff_cobs_decoded[RCV_BUFF_SIZE]; for (;;) { if (xQueueReceive(receiver_task_queue_hd, &qdata, portMAX_DELAY) == pdTRUE) { if (qdata.len > RCV_BUFF_SIZE) { serial_read(NULL, qdata.len); continue; } serial_read(buff_input, qdata.len); if (buff_input[qdata.len - 1] != '\0') { continue; } if (qdata.len == 1) { // only 1byte continue; } qdata.len--; // ignore NULL-terminate (COBS) // COBS decoding cobs_decode_result cobs_res = cobs_decode(buff_cobs_decoded, RCV_BUFF_SIZE, buff_input, qdata.len); if (cobs_res.status != COBS_DECODE_OK) { continue; } // CRC32 check // len=5 [1, 1, 1, 1, 1] uint32_t crc = crc32_calc(buff_cobs_decoded, cobs_res.out_len - 4); uint32_t crc_buff; memcpy(&crc_buff, &buff_cobs_decoded[cobs_res.out_len - 4], sizeof(uint32_t)); if (crc != crc_buff) { continue; } // protobuff decoding pb_istream_t stream = pb_istream_from_buffer(buff_cobs_decoded, cobs_res.out_len); pb_decode(&stream, &Example_msg, &data); // LED if (data.value % 2 == 0) { led_builtin_on(); } else { led_builtin_off(); } } } }

(venv) $ ./nanopb/generator/protoc message.proto --python_out=. --pyi_out=.

import serial import zlib import struct from cobs import cobs import message_pb2 PORT = "/dev/tty.usbmodem123123" BAUD = 115200 TIMEOUT = 1 def read_frame_zero_terminated(ser: serial.Serial) -> bytes: """0x00으로 끝나는 한 프레임을 읽어서, 마지막 0x00은 제거한 raw(COBS 인코딩된) 바이트를 리턴.""" buf = bytearray() while True: b = ser.read(1) if not b: continue if b == b"\x00": return bytes(buf) buf.append(b[0]) def decode_frame(frame_cobs: bytes) -> message_pb2.Example: """ frame_cobs: 0x00 terminator 제거된 COBS 인코딩 데이터 리턴: Protobuf 메시지 """ # 1) COBS decode decoded = cobs.decode(frame_cobs) if len(decoded) < 4: raise ValueError(f"decoded too short: {len(decoded)} bytes") # 2) CRC32 split (마지막 4바이트) payload = decoded[:-4] (crc_recv,) = struct.unpack("<I", decoded[-4:]) # little-endian uint32 # 3) CRC32 verify (CRC-32) crc_calc = zlib.crc32(payload) & 0xFFFFFFFF if crc_calc != crc_recv: raise ValueError(f"CRC mismatch: calc=0x{crc_calc:08X}, recv=0x{crc_recv:08X}") # 4) Protobuf decode msg = message_pb2.Example() msg.ParseFromString(payload) return msg def main(): ser = serial.Serial(port=PORT, baudrate=BAUD, timeout=TIMEOUT) while True: frame_cobs = read_frame_zero_terminated(ser) try: msg = decode_frame(frame_cobs) print("OK:", msg, "LED ON" if msg.value % 2 == 0 else "LED OFF") except Exception as e: print("ERR:", e, " raw_cobs=", frame_cobs) ser.write(frame_cobs + b"\x00") if __name__ == "__main__": main()