클라우드 동적 메시지 보드

1. 아이디어

광고나 공지사항 등을 전달하는 메시지 보드를 클라우드 환경을 적용한다면 기존의 버스나 건물에 사용하는 메시지 보드의 리모컨 등을 사용하여 메시지를 설정하는 불변함을 해소하고 사용자가 자주 메시지 내용을 변경하고자 하는 경우에도 인터넷 환경에 있는 장소라면 어디든지 상관없이 메시지 보드의 내용을 업데이트 할 수 있어서 유용하게 활용할 수 있을까? 하는 아이디어로 제작을 하게 되었습니다.

2. 전체 구조 

1. 사용자가 표시하고자 하는 메시지를 클라우드 서버에 Publish

2. 메시지 보드는 클라우스 서버에서 메시지를 Subscribe하여 디스플레이

3. 메시지 주변의 LED Light Strip으로 해당하는 메시지의 중요도 또는 그래픽적 효과 표시

3 하드웨어 구성

4. 소프트웨어 구성

5. 소스 코드

"""
Dynamic Message Board Source File
"""

import os
import sys
import _thread
import time
import machine
from machine import Pin, SPI, PWM
import micropython
from micropython import const

from dot_matrix_MAX7219 import DotMatrix_MAX7219_8x8
from led_WS2812B import LED_WS2812B


DM_INITIAL_BRIGHTNESS = const(0)
SSID = "xxxxxxxx"
PASSWORD = "xxxxxxxx"

# AWS IoT Endpoint
MQTT_BROKER_ENDPOINT = "xxxxxxxxx.iot.us-west-1.amazonaws.com"
FILE_KEY = "private.der"
FILE_CRT = "certificate.der"


_lock_print = _thread.allocate_lock()
_lock_time = _thread.allocate_lock()

_dm = DotMatrix_MAX7219_8x8(use_HW_SPI = True, SPI_select = 1, SPI_frequency_Hz = 64000000)
_led = LED_WS2812B(lock_time = _lock_time, gpio_DIN = 28, LED_count = 66)

_messageData = None                                                            # 메시지 데이터
_LED_Data = None                                                               # LED 데이터


# MQTT 메시지 수신
def callback_mqtt(topic, payload_UTF8, payload_Bytes):
    global _messageData                                                        # 메시지 데이터
    global _LED_Data                                                           # LED 데이터
    
    data = json.loads(payload_UTF8)
    _messageData = (data["message"], data["message_brightness"], data["message_speed"], data["message_timeout"])
    _LED_Data = (data["LED"], (data["LED_color"]["red"], data["LED_color"]["green"], data["LED_color"]["blue"]), data["LED_speed"], data["LED_timeout"])


# 메시지 표시 업데이트
def update_MessageDisplay():
    global _messageData                                                        # 메시지 데이터
    
    if _messageData is not None:
        (message, message_brightness, message_speed, message_timeout) = _messageData
        _messageData = None
        
        # 메시지 표시 초기화
        _ms.time_ms = 0
        _ms.period_ms = _pico.constrain(round(5 / max(float(message_speed), 0.1)), 5, 50)      # 메시지 표시 주기 (범위: 5 ~ 50)
        lines = message.split("\n")
        _ms.messageList = [line for line in lines if line.strip() != ""]   # 공백 라인 제거
        if len(_ms.messageList) < 1: _ms.messageList = [""]
        _ms.lineCount = len(_ms.messageList)
        _ms.timeout = max(int(message_timeout) * 1000, 0)                  # 유지 시간 (단위: msec, 0인 경우 계속 유지)
        _ms.line = -1
        _ms.sequence = _ms.sequenceLength
        _dm.set_Brightness(_pico.constrain(int(float(message_brightness) * 15), 0, 15))        # 밝기 설정 (범위: 0 ~ 15)
    
    _ms.time_ms += 1
    if _ms.time_ms >= _ms.period_ms:
        _ms.time_ms = 0
        MESSAGE_START_OFFSET = 25
        MESSAGE_START_DELAY = 15
        
        _ms.sequence += 1
        if _ms.sequence >= _ms.sequenceLength:
            _ms.line += 1
            if _ms.line >= len(_ms.messageList): _ms.line = 0
            _ms.sequence = 0
            _ms.sequenceLength = len(_ms.messageList[_ms.line]) * 8 + 8 + MESSAGE_START_DELAY + MESSAGE_START_OFFSET
            
        if _ms.timeout >= 0:
            if _ms.sequence <= 8: 
                _dm.clear()
                _dm.draw_Text(_ms.messageList[_ms.line], MESSAGE_START_OFFSET, 8 - _ms.sequence)
                _dm.update()
            elif _ms.sequence <= (8 + MESSAGE_START_DELAY): 
                pass
            else: 
                _dm.clear()
                _dm.draw_Text(_ms.messageList[_ms.line], MESSAGE_START_DELAY + MESSAGE_START_OFFSET + 8 -_ms.sequence, 0)
                _dm.update()
        

# LED 표시 업데이트
def update_LED_Display():
    global _LED_Data                                                           # LED 데이터
    global _ld
    
    if _LED_Data is not None:
        (LED, LED_color, LED_speed, LED_timeout) = _LED_Data
        _LED_Data = None
        _ld.time_ms = 0
        _ld.period_ms = _pico.constrain(round(5 / max(float(LED_speed), 0.1)), 5, 50)          # LED 표시 주기 (범위: 5 ~ 50)
        
        LED_timeout = None if int(LED_timeout) <= 0 else float(LED_timeout)
        _led.start_PatternLoop(str(LED), fade_in_out = True, direction = True, color = tuple(LED_color), timeout = LED_timeout)        # 패턴 루프 시작 (패턴: "BREATH" "COLOR_OVER_RAINBOW" "RAINBOW_CYCLE" "THEATER_CHASE_RAINBOW")

    _ld.time_ms += 1
    if _ld.time_ms >= _ld.period_ms:
        _ld.time_ms = 0
        _led.process_PatternLoop()

# main 함수
def main():
    global _messageData                                                        # 메시지 데이터
    global _LED_Data                                                           # LED 데이터
    
    _pico.UART0_init(baudrate = 115200, gpio_tx = 0, gpio_rx = 1, rx_invert = True)      # UART0 초기화
    _pico.UART1_init(baudrate = 115200, gpio_tx = 8, gpio_rx = 9)                        # UART1 초기화
    
    _led.clear(update = True)

    _dm.set_Brightness(DM_INITIAL_BRIGHTNESS)                                  # 밝기 설정 (범위: 0 ~ 15)
    _dm.clear()
    _dm.draw_Text("[ INITIALIZING ]", 0, 0, 1)
    _dm.update()
    
    # 네트워크 연결
    print("Connnecting Network (SSID: " + SSID + ")", end = "")
            
    wifi.connect(SSID, PASSWORD)                             # 네트워크 연결
    
    _dm.clear()
    _dm.draw_Text("[NETWORK ONLINE]", 0, 0, 1)
    _dm.update()
    
    # MQTT Broker 연결
    
    if mqttClient.connect_To_AWS_MQTT_Broker(MQTT_BROKER_ENDPOINT, FILE_KEY, FILE_CRT):  # MQTT Broker 연결 (AWS MQTT Broker)
        if DEBUG: print(" → Success")
    else:
        if DEBUG: print(" → Failure [System Reset]")
        machine.reset()                                                        # 시스템 리셋
    
    # MQTT Subscribe
    public_IP = _http_api.get_GeolocationData()[1]
    topic = "dynamic message to Pico W"
    mqttClient.subscribe(topic)                                                # MQTT Subscribe (Topic: weather data 59.6.230.229)
    
    _dm.clear()
    
    while True:
        mqttClient.process_CheckMessage()    # MQTT 메시지 처리
        update_LED_Display()                 # LED 표시 업데이트
        
    sys.exit()                                                  
    

def thread_SubCore():
    while True:
        update_MessageDisplay()
        
    _thread.exit()

if __name__ == "__main__":
    main()

6. 회로도

7. 시연 동영상

8. 제작 소감

 우선 적용된 Raspberry Pi Pico W를 사용하면서 느낀 장점은 파이썬 언어를 사용하여 보다 친숙하게 개발이 가능하였던 것과 2개의 스레드를 사용할 수 있어서 작업 프로세스를 적절히 분할하여 사용할 수 있고 서로 간섭없이 처리가 이루어져서 속도에 큰 장점이 있다는 것입니다. 특히 Pico W에는 와이파이 기능이 탑재되어 매우 저렴한 가격으로 인터넷을 활용하는 어플리케이션이 제작 가능한 것이 큰 경쟁력이라고 생각합니다. 또한 이번 프로젝트에는 사용하지 않았지만 일반적인 MCU와는 달리 기본적으로 파일시스템이 적용되어 저장 공간에 파일을 생성, 입출력이 가능하다는 것이 타 MCU와 구별되는 장점이라고 생각되었습니다. 느꼈던 가장 큰 단점으로는 실시간 디버깅이 지원되지 않기 때문에 print 함수 만을 사용하여 프로그램의 작동과 오류를 검출하는 것이 시간이 오래 걸리고 불편했다는 점입니다. 이번 프로젝트를 통해서 파이썬 언어를 펌웨어 개발에 활용하는데 많은 기법을 익힐 수 있었으며 다중 스레드에서 효과적으로 하드웨어를 제어하고 처리 속도를 향상시키는데 좋은 경험과 공부가 되었습니다. 앞으로도 이러한 MAKE 콘테스트 행사가 더 활발해진다면 공부하는 학생들과 현업의 엔지니어들에게 동기부여가 되어 기술발전에 도움이 될 것이라고 생각합니다.