ESP32를 뒤로하고, RP2040과 CC1101로 구현한 초장거리 무선 제어

작성자

micro2iot

안녕하세요! 오늘은 무선 통신 프로젝트의 ‘진한 맛’을 느껴보고 싶어 시작한 RP2040 + CC1101 장거리 통신 프로젝트 제작기를 공유합니다.

1. 왜 ESP32가 아닌 CC1101인가?

많은 분이 묻습니다. “ESP32 쓰면 Wi-Fi도 있고 블루투스도 있는데, 왜 귀찮게 외부 모듈을 쓰나요?” 하지만 엔지니어라면 공감할 CC1101만의 대체 불가능한 매력이 있습니다.

압도적인 장애물 투과력: 2.4GHz(Wi-Fi)는 벽 하나만 있어도 신호가 급감하지만, CC1101이 사용하는 Sub-1GHz(433MHz) 대역은 파장이 길어 벽을 타고 넘는 회절성이 뛰어납니다.
미친 통신 거리: 설정만 잘하면 개활지에서 수백 미터, 심지어 1km 가까이 신호를 보낼 수 있습니다.
날것의 전파(RAW RF): 표준 프로토콜에 갇히지 않고 주파수부터 데이터 레이트까지 내 입맛대로 주무를 수 있는 진정한 무선의 재미를 줍니다.

2. CC1101의 명암: 장점과 단점

직접 만져보며 느낀 CC1101의 특징을 솔직하게 정리합니다.

👍 장점

유연성: 300MHz부터 928MHz까지 광범위한 주파수 설정 가능.
저전력: 전송 시 소모 전류가 매우 적어 배터리 구동에 최적화.
호환성: 아두이노, STM32, 그리고 이번에 사용한 RP2040까지 SPI 통신만 있다면 어디든 연결 가능.

👎 단점

까다로운 초기 설정: 레지스터 값이 수백 개입니다. 하나만 틀려도 통신이 안 됩니다.
안테나 의존성: 안테나 길이나 방향에 따라 성능 차이가 극명합니다.
간섭 문제: 433MHz는 자동차 리모컨, 무선 벨 등이 많이 써서 나만의 ‘주소(Address)’ 설정이 필수입니다.

3. 하드웨어 연결 (SPI 핀 맵)

RP2040은 하드웨어 SPI 핀이 지정되어 있습니다. 매뉴얼에 명시된 SPI0 또는 SPI1 전용 핀을 반드시 사용해야 합니다.
RP2040배선을 아래 중 하나로 반드시 바꾸셔야 에러가 해결됩니다.

기능	권장 (SPI 0)	대안 (SPI 1)
SCK	GP2 또는 GP6	GP10 또는 GP14
MOSI	GP3 또는 GP7	GP11 또는 GP15
MISO	GP0 또는 GP4	GP8 또는 GP12

4. 양방향 통신 구현

“CC1101의 기본 동작은 송신 또는 수신 중 한 방향으로만 작동하는 반이중(Half-Duplex) 방식입니다. 이를 극복하기 위해 멀티쓰레딩(Multithreading) 기법을 도입했습니다. 평상시에는 수신 대기 상태를 상시 유지하다가, 송신 이벤트 발생 시에만 즉시 모드를 전환하여 데이터를 전송하고 다시 수신 모드로 복귀하는 로직을 구현함으로써, 사용자 관점에서의 **심리스(Seamless)한 양방향 통신(Bidirectional Communication)**을 완성했습니다.”

테스트의 편의를 위해 하나의 브레드보드 위에 구성했지만, **좌우측의 전원 라인까지 완전히 분리하여** 전기적으로 독립된 두 개의 개별 장치(Standalone Device) 환경을 구축했습니다. 물리적 연결 없이 오직 **CC1101의 RF 신호만을 이용해 데이터를 주고받는 실제 무선 통신 환경**을 재현했습니다.

5. 장거리 통신

Python cc1101.py

from machine import Pin, SPI
import _thread
import time

# 레지스터 주소 및 상수
WRITE_BURST     = 0x40
READ_SINGLE     = 0x80
READ_BURST      = 0xC0
CONFIG_PKTLEN   = 0x06
IDLE_STATE      = 0x36
TX_STATE        = 0x35
RX_STATE        = 0x34
FLUSH_TX        = 0x3B
FLUSH_RX        = 0x3A

class CC1101:
    def __init__(self, spi, cs):
        self.spi = spi
        self.cs = cs
        self.reset()
        self.self_test()
        self.default_config()
        self.packet_len = 16
        self.lock = _thread.allocate_lock()
        self.on_receive_callback = None # 수신 시 실행할 함수 저장용
        
        self.set_base_frequency(433.0)
        self.set_packet_length(16)

    def write_reg(self, addr, val):
        self.cs.value(0)
        self.spi.write(bytearray([addr, val]))
        self.cs.value(1)

    def read_reg(self, addr):
        self.cs.value(0)
        self.spi.write(bytearray([addr | READ_SINGLE]))
        data = self.spi.read(1)
        self.cs.value(1)
        return data[0]

    def reset(self):
        self.cs.value(1)
        time.sleep_ms(1); self.cs.value(0); time.sleep_ms(1); self.cs.value(1)
        self.cs.value(0)
        self.spi.write(bytearray([0x30])) # SRES
        self.cs.value(1)
        time.sleep_ms(10)

    def self_test(self):
        # 상태 레지스터 읽기는 반드시 READ_BURST(0xC0) 사용
        partnum = self.read_reg(0x30 | READ_BURST)
        version = self.read_reg(0x31 | READ_BURST)
        print(f"CC1101 확인 - PartNum:{partnum}, Version:{version}")

    def default_config(self):
        self.write_reg(0x0B, 0x06) # FSCTRL1
        self.set_base_frequency(433.0)
        self.write_reg(0x10, 0x56) # MDMCFG4 (DRate: 4.8kbps)
        self.write_reg(0x11, 0xF8) # MDMCFG3
        self.write_reg(0x12, 0x03) # MDMCFG2 (2-FSK, 16/16 sync word bits)
        self.write_reg(0x18, 0x18) # MCSM0 (Auto Calibration)
        self.write_reg(0x08, 0x04) # PKTCTRL0 (Fixed Length Mode)
        self.write_reg(0x06, 16)   # PKTLEN (기본 16바이트)

    def set_base_frequency(self, freq_mhz):
        f = int(freq_mhz * 65536 / 26)
        self.write_reg(0x0D, (f >> 16) & 0xFF)
        self.write_reg(0x0E, (f >> 8) & 0xFF)
        self.write_reg(0x0F, f & 0xFF)

    def set_packet_length(self, length):
        self.write_reg(0x06, length)

    def send_data(self, data):
        if isinstance(data, str): data = data.encode()
        self.write_reg(IDLE_STATE, 0)
        self.write_reg(FLUSH_TX, 0)
        self.cs.value(0)
        self.spi.write(bytearray([0x3F | WRITE_BURST]))
        self.spi.write(data) # 고정 길이 모드에서는 길이를 쓰지 않음
        self.cs.value(1)
        self.write_reg(TX_STATE, 0)
        time.sleep_ms(20) # 전송 시간 보장

    def listen(self):
        """수신 모드로 전환"""
        self.write_reg(RX_STATE, 0)

    def data_received(self):
        # RXBYTES 레지스터(0x3B) 확인
        rx_bytes = self.read_reg(0x3B | READ_BURST)
        return rx_bytes & 0x7F # 하위 7비트가 실제 바이트 수

    def receive_data(self, length=16):
        self.cs.value(0)
        self.spi.write(bytearray([0x3F | READ_BURST]))
        payload = self.spi.read(length)
        self.cs.value(1)
        self.write_reg(IDLE_STATE, 0)
        self.write_reg(FLUSH_RX, 0)
        return payload
    
    def flush_and_listen(self):
        self.write_reg(0x36, 0) # IDLE
        self.write_reg(0x3A, 0) # FLUSH_RX
        self.write_reg(0x3B, 0) # FLUSH_TX
        self.listen()

    def send_chat(self, msg_str):
        """외부에서 호출할 송신 메서드"""
        full_msg = (msg_str + " " * self.packet_len)[:self.packet_len]
        with self.lock:
            self.write_reg(0x36, 0)
            self.send_data(full_msg)
            while (self.read_reg(0xF5 | 0xC0) & 0x1F) == 0x13: # TX 상태 체크
                time.sleep_ms(1)
            self.flush_and_listen()

    def _receiver_thread(self):
        """내부 수신 쓰레드 루프"""
        self.flush_and_listen()
        while True:
            if not self.lock.locked():
                with self.lock:
                    status = self.read_reg(0x3B | 0xC0)
                    if status & 0x80:
                        self.flush_and_listen()
                    elif (status & 0x7F) >= self.packet_len:
                        data = self.receive_data(self.packet_len)
                        if data:
                            try:
                                msg = data.decode('ascii').strip()
                                if msg and self.on_receive_callback:
                                    self.on_receive_callback(msg) # 콜백 함수 실행
                            except: pass
                        self.flush_and_listen()
                    
                    if (self.read_reg(0xF5 | 0xC0) & 0x1F) == 0x01:
                        self.listen()
            time.sleep_ms(20)

    def start_dual_mode(self, callback):
        """쌍방향 통신 시작 (수신 시 실행할 함수를 인자로 받음)"""
        self.on_receive_callback = callback
        _thread.start_new_thread(self._receiver_thread, ())

Python left.py

import machine
from machine import Pin, SPI, I2C
import cc1101
import ssd1306
import time
import _thread

# ================= [ 핀 설정 구역 ] =================
# 1. CC1101 (SPI)
RF_SPI_ID = 1
RF_SCK    = 14
RF_MOSI   = 15
RF_MISO   = 8
RF_CS     = 6

# 2. SSD1306 (I2C)
OLED_I2C_ID = 0
OLED_SCL    = 5
OLED_SDA    = 4

# 3. 버튼 (내부 풀업)
BTN_PIN     = 2
# =================================================

# --- 객체 초기화 ---
# OLED
i2c = I2C(OLED_I2C_ID, scl=Pin(OLED_SCL), sda=Pin(OLED_SDA), freq=400000)
display = ssd1306.SSD1306_I2C(128, 32, i2c)

# 버튼
btn = Pin(BTN_PIN, Pin.IN, Pin.PULL_UP)

# CC1101
spi = SPI(RF_SPI_ID, baudrate=5000000, polarity=0, phase=0, 
          sck=Pin(RF_SCK), mosi=Pin(RF_MOSI), miso=Pin(RF_MISO))
cs = Pin(RF_CS, Pin.OUT, value=1)

rf = cc1101.CC1101(spi, cs)
rf.set_base_frequency(433.0)
rf.set_packet_length(16)

# 전역 변수
tx_display = "READY"
rx_display = "WAITING..."

def draw_ui(invert=False):
    """화면을 그리는 함수 (invert=True면 반전효과)"""
    display.fill(0)
    if invert:
        display.fill(1) # 배경을 하얗게
        color = 0       # 글씨를 검게
    else:
        color = 1       # 글씨를 하얗게

    # 상단: TX 상태
    display.text("TX > " + tx_display, 5, 4, color)
    display.hline(0, 16, 128, color)
    # 하단: RX 상태
    display.text("RX < " + rx_display, 5, 22, color)
    
    display.show()

def my_receive_handler(message):
    global rx_display
    rx_display = message
    # [쇼츠 연출] 수신 시 화면 2번 깜빡이기
    for _ in range(2):
        draw_ui(invert=True)
        time.sleep_ms(100)
        draw_ui(invert=False)
        time.sleep_ms(100)

# 쌍방향 모드 시작
rf.start_dual_mode(my_receive_handler)

draw_ui()
print("🎬 쇼츠 촬영 준비 완료!")

last_btn_state = 1
msg_count = 0

try:
    while True:
        btn_state = btn.value()
        
        # 버튼 눌림 감지 (Falling Edge)
        if last_btn_state == 1 and btn_state == 0:
            msg_count += 1
            msg_to_send = f"Left {msg_count}"
            tx_display = "SENDING..."
            
            # [쇼츠 연출] 송신 시 '슈슉' 지나가는 애니메이션
            for i in range(0, 128, 32):
                display.scroll(16, 0)
                draw_ui(invert=True)
                time.sleep_ms(20)
            
            rf.send_chat(msg_to_send)
            tx_display = msg_to_send
            draw_ui()
            
            time.sleep_ms(300) # 디바운싱
            
        last_btn_state = btn_state
        time.sleep_ms(10)

except KeyboardInterrupt:
    print("\nSTOP")

Python right.py

import machine
from machine import Pin, SPI, I2C
import cc1101
import ssd1306
import time
import _thread

# ================= [ 핀 설정 구역 ] =================
# 1. CC1101 (SPI)
RF_SPI_ID = 0
RF_SCK    = 2
RF_MOSI   = 3
RF_MISO   = 4
RF_CS     = 15

# 2. SSD1306 (I2C)
OLED_I2C_ID = 1
OLED_SCL    = 7
OLED_SDA    = 6

# 3. 버튼 (내부 풀업)
BTN_PIN     = 28
# =================================================

# --- 객체 초기화 ---
# OLED
i2c = I2C(OLED_I2C_ID, scl=Pin(OLED_SCL), sda=Pin(OLED_SDA), freq=400000)
display = ssd1306.SSD1306_I2C(128, 32, i2c)

# 버튼
btn = Pin(BTN_PIN, Pin.IN, Pin.PULL_UP)

# CC1101
spi = SPI(RF_SPI_ID, baudrate=5000000, polarity=0, phase=0, 
          sck=Pin(RF_SCK), mosi=Pin(RF_MOSI), miso=Pin(RF_MISO))
cs = Pin(RF_CS, Pin.OUT, value=1)

rf = cc1101.CC1101(spi, cs)
rf.set_base_frequency(433.0)
rf.set_packet_length(16)

# 전역 변수
tx_display = "READY"
rx_display = "WAITING..."

def draw_ui(invert=False):
    """화면을 그리는 함수 (invert=True면 반전효과)"""
    display.fill(0)
    if invert:
        display.fill(1) # 배경을 하얗게
        color = 0       # 글씨를 검게
    else:
        color = 1       # 글씨를 하얗게

    # 상단: TX 상태
    display.text("TX < " + tx_display, 5, 22, color)
    display.hline(0, 16, 128, color)
    # 하단: RX 상태
    display.text("RX > " + rx_display, 5, 4, color)
    
    display.show()

def my_receive_handler(message):
    global rx_display
    rx_display = message
    # [쇼츠 연출] 수신 시 화면 2번 깜빡이기
    for _ in range(2):
        draw_ui(invert=True)
        time.sleep_ms(100)
        draw_ui(invert=False)
        time.sleep_ms(100)

# 쌍방향 모드 시작
rf.start_dual_mode(my_receive_handler)

draw_ui()
print("🎬 쇼츠 촬영 준비 완료!")

last_btn_state = 1
msg_count = 0

try:
    while True:
        btn_state = btn.value()
        
        # 버튼 눌림 감지 (Falling Edge)
        if last_btn_state == 1 and btn_state == 0:
            msg_count += 1
            msg_to_send = f"Right {msg_count}"
            tx_display = "SENDING..."
            
            # [쇼츠 연출] 송신 시 '슈슉' 지나가는 애니메이션
            for i in range(0, 128, 32):
                display.scroll(16, 0)
                draw_ui(invert=True)
                time.sleep_ms(20)
            
            rf.send_chat(msg_to_send)
            tx_display = msg_to_send
            draw_ui()
            
            time.sleep_ms(300) # 디바운싱
            
        last_btn_state = btn_state
        time.sleep_ms(10)

except KeyboardInterrupt:
    print("\nSTOP")

코멘트