라즈베리파이를 시리얼 통신 중계 서버로 사용한 현장 문제 해결기

language

kor

date

Jan 7, 2026

slug

raspberry-pi-serial-bridge-workaround

author

status

Public

tags

Hardware

Error

summary

PC–아두이노 간 시리얼 통신이 3~4m 거리에서 불안정해지는 현장 문제를 해결하기 위해, 라즈베리파이를 중계 서버로 구성하고 가상 COM 포트를 활용한 우회적인 해결 방법을 정리했다. 비효율적이지만 기존 Host 애플리케이션 수정 없이 빠르게 문제를 해결할 수 있었던 사례이다.

type

Post

thumbnail

category

🐞TroubleShooting

updatedAt

Jan 6, 2026 05:59 PM

PC 설정

hub4com --baud=9600 COM4 --use-driver=tcp 192.168.0.10:5000

COM to TCP 바인딩 *.bat 파일 스크립트

@echo off
REM ---------------------------------------------
REM 1) 작업 디렉터리로 이동
REM    com0com hub4com 실행 파일이 있는 경로
REM ---------------------------------------------
cd /d "C:\Users\Downloads\hub4com-2.1.0.0-386\hub4com-2.1.0.0-386"

REM ---------------------------------------------
REM 2) hub4com 실행
REM    COM4 포트(9600 bps) ↔ TCP 클라이언트 모드(192.168.0.10:5000)
REM ---------------------------------------------
hub4com --baud=9600 COM4 --use-driver=tcp 192.168.0.10:5000

REM ---------------------------------------------
REM (선택) 창이 바로 닫히지 않도록 일시 정지
REM  필요 없으면 다음 줄을 지우세요.
REM ---------------------------------------------
pause

라즈베리파이 코드

#!/usr/bin/env python3
import socket
import threading
import serial
import time

# ============================
# 시리얼 포트 설정 (아두이노 측)
# ============================
SERIAL_PORT    = '/dev/ttyACM0'   # 실제 연결된 아두이노 시리얼 포트
BAUD_RATE      = 9600             # 아두이노 통신 속도
SERIAL_TIMEOUT = 1                # 시리얼 읽기 타임아웃 (초)

# ============================
# TCP 서버 설정 (라즈베리파이 측)
# ============================
TCP_IP   = '0.0.0.0'   # 모든 인터페이스에서 수신
TCP_PORT = 5000        # 클라이언트가 접속할 포트

def handle_client(conn, ser):
    """
    클라이언트 소켓(conn)을 통해 들어오는 바이트를 1바이트씩 읽어들여
    '$' 구분자가 나올 때까지 조각을 모아 full_cmd로 간주한 뒤 시리얼로 전송합니다.
    아두이노 응답은 다시 TCP 클라이언트로 송신합니다.
    """
    peer = conn.getpeername()
    print(f"[+] 클라이언트 연결: {peer}")

    # 1) 소켓을 바이너리 모드 파일 객체로 감싸기
    conn_file = conn.makefile('rb')
    partial_chunks = []  # '$' 나오기 전까지 받은 조각을 저장할 리스트

    try:
        while True:
            # 2) 한 번에 1바이트씩 읽어서 '$'를 기다린다
            byte = conn_file.read(1)
            if not byte:
                # 연결이 종료된 경우
                break

            partial_chunks.append(byte)
            if byte == b'$':
                # 3) '$'까지 모인 조각 전체를 하나의 명령으로 합친다
                full_cmd_bytes = b''.join(partial_chunks)
                partial_chunks.clear()

                # 4) UTF-8로 디코딩 (옵션: 로그 출력)
                try:
                    full_cmd_str = full_cmd_bytes.decode('utf-8', errors='replace')
                except Exception:
                    full_cmd_str = ''
                print(f"[완전수신] {full_cmd_str.strip()}")

                # 5) 시리얼 포트로 한 번에 전송
                ser.write(full_cmd_bytes)

                # 6) 아두이노 응답을 잠시 대기 후 읽어서 TCP로 송신
                time.sleep(0.1)
                bytes_to_read = ser.in_waiting
                if bytes_to_read:
                    response_bytes = ser.read(bytes_to_read)
                    conn.sendall(response_bytes)
                else:
                    # 응답이 없으면 넘어감
                    pass

    except Exception as e:
        print(f"[!] 클라이언트 처리 중 예외 발생 ({peer}): {e}")
    finally:
        conn_file.close()
        conn.close()
        print(f"[-] 클라이언트 연결 종료: {peer}")

def main():
    # 1) 아두이노 시리얼 포트 열기
    ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=SERIAL_TIMEOUT)
    # 일부 아두이노 보드는 시리얼 포트 오픈 후 안정화 대기가 필요합니다.
    time.sleep(2)
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 시리얼 포트 열림: {SERIAL_PORT} @ {BAUD_RATE}bps")

    # 2) TCP 소켓 서버 생성
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((TCP_IP, TCP_PORT))
    srv.listen(1)
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] TCP 서버 시작: {TCP_IP}:{TCP_PORT} - 클라이언트 접속 대기 중...")

    try:
        while True:
            conn, addr = srv.accept()
            # 클라이언트가 연결되면 별도 스레드에서 처리
            client_thread = threading.Thread(target=handle_client, args=(conn, ser), daemon=True)
            client_thread.start()

    except KeyboardInterrupt:
        print("\n[!] 서버를 종료합니다...")

    finally:
        srv.close()
        ser.close()
        print("[*] 시리얼 연결 및 소켓 서버 종료")

if __name__ == '__main__':
    main()