[MQTT] pc 2개 통신 (python, 양방향 통신)

개요

PC 2개가 있다.

하나는 User가 업무를 요청하고, 나머지 하나는 Robot이 그 업무를 받아서 실시간 상태를 User가 알 수 있도록 업데이트를 해줄 것이다.

전 프로젝트에서는 Robot이 하나 뿐이라서 socket으로 연결을 했다.

이번에는 Robot이 추가로 더 늘어날 가능성을 염두해서 MQTT를 활용하려고 한다.

 

편히 PC, Robot이라고 칭하겠다.

• PC → Robot : 수행할 작업 (텍스트 형식) 전달
• Robot → PC : 작업 상태 (실시간 업데이트) 전달

 

구조

• PC는 작업지시를 발행하고, 로봇은 이를 구독
• 로봇은 상태를 발행하고, PC는 이를 구독하여 실시간으로 상태를 확인

Topic 설계

• PC → Robot: "robot/task"
• Robot → PC: "robot/status"

 

간단한 코드 구현

 

1. mqtt 설치 (버전 1.6.1)

pip3 install paho-mqtt==1.6.1

 

PC 측 코드

import paho.mqtt.client as mqtt
import threading
import time

broker_ip = "브로커_IP"
received_status = []

# 상태 메시지 수신 콜백
def on_message(client, userdata, msg):
    status = msg.payload.decode()
    print(f"[로봇 상태 업데이트]: {status}")
    received_status.append(status)
    
    if "작업 완료" in status:
        print("모든 작업 완료! 프로그램 종료 ...")
        client.loop_stop()
        client.disconnect()

# MQTT 클라이언트 설정
client = mqtt.Client("PC_Client")
client.connect(broker_ip, 1883)

# 로봇 상태 업데이트 수신 구독
client.subscribe("robot/status")
client.on_message = on_message

# MQTT 수신 대기 시작
client.loop_start()

# 로봇에게 작업 지시 보내기
task_message = "작업 시작: 콜라 옮기기"
print(f"[작업 전송]: {task_message}")
client.publish("robot/task", task_message)

# 프로그램이 종료되지 않도록 대기
while client.is_connected():
    time.sleep(1)

 

로봇 측 코드

(1) 상태 전송용 모듈 (robot_status.py)

import paho.mqtt.client as mqtt

class RobotStatusUpdater:
    def __init__(self, broker_ip, topic="robot/status"):
        self.client = mqtt.Client("Robot_Status_Updater")
        self.client.connect(broker_ip, 1883)
        self.topic = topic

    def send_status(self, message):
        self.client.publish(self.topic, message)
        print(f"[상태 전송됨]: {message}")

# 전역적으로 하나만 생성
status_updater = RobotStatusUpdater("브로커_IP")

def send_status(message):
    status_updater.send_status(message)

 

(2) 실제 로봇 코드 (robot.py)

• send_status 호출하는 방법

import paho.mqtt.client as mqtt
import threading
import time
from robot_status import send_status

broker_ip = "브로커_IP"

# PC에서 작업 지시 받는 콜백 함수
def on_message(client, userdata, msg):
    task = msg.payload.decode()
    print(f"[받은 작업 지시]: {task}")
    execute_task(task)

# 실제 로봇의 작업 수행 함수
def execute_task(task):
    # 기존 로봇 코드의 예시 (단계별로 상태 전송 추가)
    time.sleep(1)
    send_status("초기화 완료")

    time.sleep(1)
    send_status("목표 위치 도착 완료")

    time.sleep(1)
    send_status("물체 픽업 성공")

    time.sleep(1)
    send_status("목표 위치 이동 완료")

    time.sleep(1)
    send_status("물체 내려놓기 완료")

    send_status("전체 작업 완료")

# MQTT 클라이언트 설정 (작업지시 받기용)
client = mqtt.Client("Robot_Client")
client.connect(broker_ip, 1883)
client.subscribe("robot/task")
client.on_message = on_message

# MQTT 수신 루프 시작
threading.Thread(target=client.loop_forever, daemon=True).start()

# 로봇은 메시지를 기다리며 대기 중...
while True:
    time.sleep(1)

 

+ 메세지 브로커는 PC 측에서 띄어서 테스트했다. 아마 계속 PC 측에서 띄어서 사용할 듯 ..?

 

실제 테스트 (동영상)

(1) MQTT Explorer : 로컬 mqtt 서버에 연결하고

(2) python3 robot.py : robot은 대기 상태에 놓은 상태 유지하고

(3) python3 pc.py : pc에서 task 요청 들어가면

(4) robot에서 받아서 task 실행하며 실시간 status을 공유한다 (pc측에서 실시간으로 받아서 확인 가능)