UART로 센서 값을 읽을 때 한 번의 read()가 한 메시지라고 가정하면, 간헐적으로 값이 빠지거나 두 메시지가 붙어 보일 수 있습니다. 수신 코드는 도착한 바이트를 모으고, 응용 프로토콜이 정한 경계에서 메시지를 꺼내야 합니다.
이 글은 LF로 끝나는 텍스트 메시지를 예로 들어 분할 수신, 연속 수신, 지나치게 긴 입력에서의 복구를 설명합니다. Modbus RTU나 임의의 바이너리 프로토콜에 그대로 적용하는 규격은 아닙니다.
1. 수신 청크와 메시지는 크기가 다를 수 있습니다
장치가 TEMP,23.5 뒤에 CR·LF를 붙여 보냈다고 가정해 보겠습니다. PC에서는 첫 번째 읽기에 TEMP,2, 두 번째 읽기에 나머지가 들어올 수 있습니다. 반대로 온도와 습도 메시지가 같은 읽기 결과에 함께 들어올 수도 있습니다.
pySerial의 read(size)는 타임아웃이 설정되어 있으면 요청한 개수보다 적은 바이트를 반환할 수 있습니다. read_until()도 구분자 외에 크기 제한이나 타임아웃으로 끝날 수 있으므로 반환됐다는 사실만으로 메시지 완성을 판단해서는 안 됩니다. pySerial 공식 읽기 API
예제에서 사용하는 규칙은 다음과 같습니다.
- LF 바이트 하나를 메시지 끝으로 사용합니다.
- LF 바로 앞의 CR 하나는 제거해 CR·LF도 받아들입니다.
- 길이 제한은 LF를 제외한 바이트 수이며, CR이 있다면 CR도 포함합니다.
- 본문에 실제 LF를 넣어야 하는 데이터는 별도의 이스케이프 규칙이나 길이 기반 프레임을 설계해야 합니다.
2. 조립 중인 바이트를 다음 읽기까지 보관합니다
파서는 완성된 메시지를 여러 개 반환할 수 있고, 아직 끝나지 않은 바이트는 내부 버퍼에 남깁니다. 예제의 길이 한도는 1~65,536바이트 사이의 정수로 설정합니다. 아래 코드를 line_framer.py로 저장해 실행할 수 있습니다.
"""LF-delimited demo protocol; limit counts all bytes before LF, including CR."""
class LineFramer:
def __init__(self, limit=128):
if type(limit) is not int or limit not in range(1, 65537):
raise ValueError("limit must be an integer from 1 to 65536")
self.limit = limit
self.buffer = bytearray()
self.dropping = False
self.dropped = 0
def feed(self, chunk):
frames = []
for value in chunk:
if self.dropping:
if value == 10:
self.dropping = False
continue
if value == 10:
frame = bytes(self.buffer)
self.buffer.clear()
if frame.endswith(b"\r"):
frame = frame[:-1]
frames.append(frame)
elif len(self.buffer) == self.limit:
self.buffer.clear()
self.dropping = True
self.dropped += 1
else:
self.buffer.append(value)
return frames
if __name__ == "__main__":
parser = LineFramer(limit=16)
print(parser.feed(b"TEMP,2"))
print(parser.feed(b"3.5\r\nHUM,48\n"))
실행 결과는 다음과 같습니다.
[] [b'TEMP,23.5', b'HUM,48']
첫 입력에는 LF가 없으므로 완성 메시지가 없습니다. 두 번째 입력에서 첫 메시지를 마무리하고 같은 청크에 있던 습도 메시지도 꺼냅니다. LF만 입력되면 빈 메시지 b''가 만들어집니다. 빈 메시지를 허용할지는 이후의 데이터 검증 단계에서 결정합니다.
버퍼에는 bytearray를 쓰고, 완성 시에는 독립된 bytes로 반환합니다. 바이트 단위 순회와 두 자료형의 차이는 Python 공식 바이너리 시퀀스 문서에서 확인할 수 있습니다.
3. 길이를 넘겼다면 다음 경계까지 버립니다
버퍼가 가득 찼을 때 내용을 지우기만 하면 초과 메시지의 뒷부분이 새 메시지처럼 해석될 수 있습니다. 예를 들어 제한이 4바이트인데 ABCDEOK와 LF가 오면, OK만 정상 메시지로 처리하는 오류가 생길 수 있습니다.
예제는 초과를 감지하면 dropping 상태로 바꾸고 다음 LF까지 전부 소비합니다. LF를 만난 뒤에야 새 메시지를 받습니다. dropped는 이 이유로 버린 메시지 수를 셉니다. 다음 실습은 앞의 클래스 정의 뒤에 붙여 실행할 수 있습니다.
parser = LineFramer(limit=4) print(parser.feed(b"ABCDEOK\nYES\n")) print(parser.dropped)
결과는 [b'YES']와 1입니다. 초과 프레임의 일부가 출력되지 않고 다음 정상 프레임부터 복구됩니다. 한도와 정확히 같은 길이의 메시지는 받을 수 있습니다. 예를 들어 제한이 4라면 ABCD 뒤의 LF는 허용하지만, ABCD 뒤의 CR·LF는 LF 앞에 5바이트가 있으므로 초과입니다.
이 구현이 제한하는 것은 조립 버퍼입니다. feed()에 거대한 입력을 한꺼번에 주면 입력 자체와 반환 목록이 메모리를 차지합니다. 실제 읽기 크기도 제한하고, 고속·대용량 환경에서는 완성 메시지를 즉시 소비하는 구조를 검토하세요.
4. 실제 포트에서는 파서 인스턴스를 유지합니다
포트를 읽는 반복문 안에서 매번 파서를 만들면 미완성 데이터가 사라집니다. 연결을 시작할 때 한 번 만들고 같은 연결의 다음 읽기에도 사용합니다. 아래 코드는 위의 클래스가 line_framer.py에 저장되어 있고, 장치가 115200 bps·8N1·흐름 제어 없음으로 데이터를 송신한다고 가정합니다.
import serial
from line_framer import LineFramer
parser = LineFramer(limit=128)
with serial.Serial("COM3", 115200, timeout=0.2) as port:
try:
while True:
chunk = port.read(128)
for frame in parser.feed(chunk):
print(repr(frame))
except KeyboardInterrupt:
pass
COM3과 통신 설정은 실제 장치에 맞게 바꾸며, 종료는 Ctrl+C입니다. 수신 타임아웃으로 빈 바이트가 반환되더라도 파서는 조립 중인 내용을 유지합니다. 응용 규칙상 오래된 조각을 버려야 한다면 마지막 수신 시각과 별도의 프레임 제한 시간을 관리하고, 폐기 뒤 어느 경계에서 다시 읽기 시작할지도 정해야 합니다.
텍스트 디코딩은 메시지를 조립한 다음 수행하세요. UTF-8 한 글자가 두 청크로 갈라질 수 있기 때문입니다. 완성된 바이트에도 잘못된 인코딩이나 잘못된 필드 값이 들어올 수 있으므로, 프레임 경계 확인과 내용 검증은 각각 필요합니다.
5. 현장 적용 전에 확인할 경계 조건
| 입력 상황 | 기대 동작 |
|---|---|
| 메시지를 모든 가능한 위치에서 둘로 분할 | 한 번에 입력한 것과 같은 메시지 출력 |
| 메시지 여러 개를 한 번에 입력 | 순서대로 모두 출력 |
| CR과 LF 사이에서 분할 | CR을 제거한 메시지 하나 출력 |
| LF 없이 한도 초과 | 조립 버퍼를 비우고 다음 LF까지 폐기 |
| 초과 프레임 다음에 정상 프레임 | 정상 프레임부터 복구 |
| 연결 해제 후 재접속 | 이전 연결의 조각과 새 연결의 데이터를 합치지 않음 |
재접속 시에는 파서를 새로 만들고 동기화 규칙을 적용합니다. 메시지 중간부터 수신할 가능성이 있다면 첫 LF까지 버린 뒤 다음 줄부터 처리하거나, 장치와 명시적인 시작 절차를 사용하세요. LF만으로는 내용의 손상 여부나 장치의 명령 실행 완료까지 확인할 수 없습니다. 응용 요구에 따라 체크섬, 순서 번호, 응답 규칙을 별도로 설계합니다.
이 글의 예제는 PC에서 분할 위치·연속 프레임·길이 초과 복구를 검사한 학습용 코드입니다. 특정 MCU의 UART 인터럽트, DMA, 링 버퍼 또는 물리 배선을 실측한 결과는 아닙니다.
함께 읽기
- pySerial 설치·루프백·HEX 수신 실습 자료: 예제 파일과 한국어 안내서 다운로드.
- 센서 JSON 스키마 설계: 조립한 메시지의 필드와 값을 검증하는 다음 단계.
- STM32 펌웨어 개발: 장치 통신과 펌웨어 개발 안내.
작성 기준: 2026-09-17. 파서와 점검 항목은 이 글의 예제 프로토콜을 위한 설계 제안입니다.