μžλ°” μ–Έμ–΄κ°€ IoT λ””λ°”μ΄μŠ€λ₯Ό μœ„ν•΄ λ§Œλ“€μ–΄μ§„ κ²ƒμœΌλ‘œ μ•Œλ €μ Έ μžˆμ§€λ§Œ μ‹€μ œλ‘œλŠ” λΌμ¦ˆλ² λ¦¬νŒŒμ΄μ™€ 같은 μ†Œν˜• λ””λ°”μ΄μŠ€μ— 파이썬 λ˜λŠ” C둜 μž‘μ„±λœ μ• ν”Œλ¦¬μΌ€μ΄μ…˜μœΌλ‘œ μž‘μ„±λ˜μ–΄ ν¬ν•¨λ˜λŠ” 것 κ°™λ‹€. 예λ₯Ό λ“€μ–΄, 파이썬 μ–Έμ–΄λ‘œ AWS IoT 와 ν†΅μ‹ ν•˜λŠ” μ• ν”Œλ¦¬μΌ€μ΄μ…˜μ„ λ§Œλ“€λ•ŒλŠ” AWS IoT Device SDK v2 for Pythonλ₯Ό μ‚¬μš©ν•  수 μžˆλ‹€. basic-connect.py λ˜λŠ” pubsub.py μƒ˜ν”Œ μ½”λ“œλ₯Ό μ°Έκ³ ν•˜μ—¬ ν΄λΌμ΄μ–ΈνŠΈ λ””λ°”μ΄μŠ€λ‘œ μˆ˜μ§‘λœ 데이터λ₯Ό AWS IoTλ₯Ό 톡해 λͺ¨λ‹ˆν„°λ§ μ‹œμŠ€ν…œμœΌλ‘œ 전달할 수 μžˆμŒμ„ ν…ŒμŠ€νŠΈ ν•΄λ³Ό 수 μžˆλ‹€. 파이썬 μ–Έμ–΄λŠ” ν…ŒμŠ€νŠΈ μ—”μ§€λ‹ˆμ–΄λ„ μ‚¬μš©ν•˜λŠ” λ²”μš©μ μΈ 언어이기 λ•Œλ¬Έμ— κ°œλ°œμžμ™€ QA λͺ¨λ‘ μ‚¬μš©ν•  수 μžˆλŠ” ν…ŒμŠ€νŠΈ μ½”λ“œκ°€ λœλ‹€.

파이썬 ν™˜κ²½ μ€€λΉ„ν•˜κΈ°

파이썬 μ½”λ“œλ₯Ό μ‹€ν–‰ν•˜κΈ° μœ„ν•΄μ„œλŠ” 파이썬 가상 ν™˜κ²½μ„ μ€€λΉ„ν•΄μ•Όν•œλ‹€. 파이썬의 가상 ν™˜κ²½μ„ κ΅¬μ„±ν•˜λŠ” 건 생각보닀 쉽지 μ•Šμ€ 뢀뢄일 수 있기 λ•Œλ¬Έμ— νŒŒμ΄μ¬μ„ 주둜 닀루지 μ•ŠλŠ” 개발자라면 러슀트둜 μž‘μ„±λœ UVλ₯Ό 파이썬 νŒ¨ν‚€μ§€ κ΄€λ¦¬μžλ‘œ μ„€μΉ˜ν•˜μ—¬ 파이썬 가상 ν™˜κ²½(.venv)λ₯Ό μ‰½κ²Œ κ΅¬μ„±ν•˜λŠ”κ²Œ 쒋을 것 κ°™λ‹€. 참고둜, μ—¬λŸ¬κ°€μ§€ 파이썬으둜 μž‘μ„±λœ MCPλ₯Ό μ„€μΉ˜ν•  λ•Œμ—λ„ UV λͺ…λ Ήμ–΄κ°€ μ‚¬μš©λ˜κ³  μžˆλ‹€.

Windows Terminal
PS C:\Users\Mambo> uv python install 3.12 Installed Python 3.12.9 in 7.79s + cpython-3.12.9-windows-x86_64-none PS C:\Users\Mambo> uv python pin 3.12 Pinned `.python-version` to `3.12` PS C:\Users\Mambo> uv python dir C:\Users\Mambo\AppData\Roaming\uv\python PS C:\Users\Mambo> uv init --app awsiot Initialized project `awsiot` at `C:\Users\Mambo\awsiot` PS C:\Users\Mambo\awsiot> uv run main.py Using CPython 3.12.9 Creating virtual environment at: .venv Hello from awsiot!

일반적으둜 PyCharm을 μ‚¬μš©ν•΄μ„œ 파이썬 ν”„λ‘œμ νŠΈλ₯Ό μ—΄κ²Œ 되면 Python 인터프리터λ₯Ό μ„€μ •ν•΄μ•Ό λ©λ‹ˆλ‹€.
UV λ₯Ό μ‚¬μš©ν•˜λŠ” ν”„λ‘œμ νŠΈλŠ” 파이썬 가상 ν™˜κ²½μ΄ μžλ™μœΌλ‘œ κ΅¬μ„±λ˜μ—ˆκΈ° λ•Œλ¬Έμ— Python 인터프리터가 κΈ°λ³Έ 섀정됨을 확인할 수 μžˆμŠ΅λ‹ˆλ‹€.

파이썬 νŒ¨ν‚€μ§€ μ„€μΉ˜ν•˜κΈ°

파이썬 μ½”λ“œμ—μ„œ ν•„μˆ˜μ μœΌλ‘œ μ‚¬μš©ν•΄μ•Όν•  AWS IoT Device SDK v2 for Python (awsiotsdk)와 ν•¨κ»˜ ν™˜κ²½ λ³€μˆ˜λ₯Ό 파일둜 μ μš©ν•  수 μžˆλŠ” python-dotenv λ₯Ό μ„€μΉ˜ν•˜μž. python-dotenv λŠ” λ°˜λ“œμ‹œ ν•„μš”ν•˜μ§€ μ•Šμ§€λ§Œ AWS IoT Endpoint 와 ClientID 와 같은 일뢀 정보듀을 ν•˜λ“œμ½”λ”©ν•˜λŠ” κ²ƒλ³΄λ‹€λŠ” λ‚˜μ„ 것이닀.

Windows Terminal
PS C:\Users\Mambo\awsiot> uv add awsiotsdk python-dotenv Resolved 4 packages in 57ms Installed 3 packages in 16ms + awscrt==0.27.4 + awsiotsdk==1.24.0 + python-dotenv==1.1.1

mqtt_connection_builder 둜 AWS IoT에 μ—°κ²°ν•˜κΈ°

예제 μ½”λ“œ 쀑 basic_connect.py λ₯Ό 보면 awsiotsdk 의 mqtt_connection_builderλ₯Ό μ‚¬μš©ν•˜λ©° MQTT 연결을 생성할 수 μžˆμŒμ„ μ•Œ 수 μžˆλ‹€. μ•„λž˜μ™€ 같이 μ½”λ“œλ₯Ό μž‘μ„±ν•˜κ³  파이썬 μ½”λ“œλ₯Ό 싀행해보면 AWS IoT μ—”λ“œν¬μΈνŠΈμ— μ—°κ²°λ˜κ³ λ‚˜μ„œ μ’…λ£Œλ¨μ„ μ•Œ 수 μžˆλ‹€.

main.py
import os from awscrt import mqtt from awsiot import mqtt_connection_builder from dotenv import load_dotenv load_dotenv() ENDPOINT = os.getenv("ENDPOINT") # iot:Data-ATS endpoint CLIENT_ID = os.getenv("CLIENT_ID") TOPIC = os.getenv("TOPIC") CERTIFICATE_PATH = os.getenv("CERTIFICATE_PATH") PRIVATE_KEY_PATH = os.getenv("PRIVATE_KEY_PATH") ROOT_CA_PATH = os.getenv("ROOT_CA_PATH") if __name__ == '__main__': print("\nStarting MQTT PubSub sample...") # Create a MQTT connection mqtt_connection = mqtt_connection_builder.mtls_from_path( endpoint=ENDPOINT, ca_filepath=ROOT_CA_PATH, cert_filepath=CERTIFICATE_PATH, pri_key_filepath=PRIVATE_KEY_PATH, client_id=CLIENT_ID, ) # Connect MQTT Broker... print(f"Connecting to {ENDPOINT} with client ID '{CLIENT_ID}'...") connect_future = mqtt_connection.connect() connect_future.result() # Future.result() waits until a result is available print("Connected!")
Windows Terminal
(awsiot) PS C:\Users\Mambo\awsiot> uv run main.py Starting MQTT PubSub sample... Connecting to $prefix$-ats.iot.ap-northeast-2.amazonaws.com with client ID '$client_id$'... Connected!

Topic κ΅¬λ…ν•˜κ³  MQTT λ©”μ‹œμ§€ κ²Œμ‹œν•˜κΈ°

예제 μ½”λ“œ 쀑 pubsub.py λ₯Ό μ°Έκ³ ν•˜λ©΄ mqtt_connection 의 subscribe ν•¨μˆ˜λ‘œ λ©”μ‹œμ§€λ₯Ό κ΅¬λ…ν•˜κ³  publish ν•¨μˆ˜λ‘œ λ©”μ‹œμ§€λ₯Ό κ²Œμ‹œν•˜λŠ” 것을 μ•Œ 수 μžˆλ‹€.

main.py
import os import time from awscrt import mqtt from awsiot import mqtt_connection_builder from dotenv import load_dotenv load_dotenv() # AWS IoT Core ENDPOINT = os.getenv("ENDPOINT") # iot:Data-ATS endpoint CLIENT_ID = os.getenv("CLIENT_ID") # Recommended Thing ID TOPIC = os.getenv("TOPIC") # AWS IoT Core - Thing Certificate CERTIFICATE_PATH = os.getenv("CERTIFICATE_PATH") PRIVATE_KEY_PATH = os.getenv("PRIVATE_KEY_PATH") ROOT_CA_PATH = os.getenv("ROOT_CA_PATH") # Optional def on_message_received(topic, payload): print(f"Received message from {topic}: \n<<< {payload.decode()}") if __name__ == '__main__': print("\nStarting MQTT PubSub sample...") # Create a MQTT connection mqtt_connection = mqtt_connection_builder.mtls_from_path( endpoint=ENDPOINT, ca_filepath=ROOT_CA_PATH, cert_filepath=CERTIFICATE_PATH, pri_key_filepath=PRIVATE_KEY_PATH, client_id=CLIENT_ID, ) # Connect MQTT Broker... print(f"Connecting to {ENDPOINT} with client ID '{CLIENT_ID}'...") connect_future = mqtt_connection.connect() connect_future.result() # Future.result() waits until a result is available print("Connected!") # Subscribe print(f"Subscribing to topic '{TOPIC}'...") subscribe_future, packet_id = mqtt_connection.subscribe( topic=TOPIC, qos=mqtt.QoS.AT_LEAST_ONCE, callback=on_message_received) subscribe_result = subscribe_future.result() print(f"Subscribed with {subscribe_result}") # Publish print(f"Publishing message to topic '{TOPIC}'...") payload = "PING" mqtt_connection.publish( topic=TOPIC, payload=payload.encode('ascii'), qos=mqtt.QoS.AT_LEAST_ONCE) print(f"Published message to {TOPIC}: \n>>> {payload}") time.sleep(2) # Disconnect print("Disconnecting...") disconnect_future = mqtt_connection.disconnect() disconnect_future.result() print("Disconnected!")
Windows Terminal
(awsiot) PS C:\Users\Mambo\awsiot> uv run main.py Starting MQTT PubSub sample... Connecting to $prefix$-ats.iot.ap-northeast-2.amazonaws.com with client ID '$client_id$'... Connected! Subscribing to topic '$topic$'... Subscribed with {'packet_id': 1, 'topic': '$topic$', 'qos': <QoS.AT_LEAST_ONCE: 1>} Publishing message to topic '$topic$'... Published message to $topic$: >>> PING Received message from $topic$: <<< PING Disconnecting... Disconnected!

λ©”μ‹œμ§€ κ²Œμ‹œμ™€ λ™μΌν•œ 토픽을 κ΅¬λ…ν•œ κ²ƒμœΌλ‘œ MQTT λ©”μ‹œμ§€ 브둜컀인 AWS IoT μ—μ„œ λ©”μ‹œμ§€λ₯Ό 전달됨은 ν™•μΈν–ˆλ‹€. λ‹€λ§Œ, AWS IoT 둜 κ²Œμ‹œλœ λ©”μ‹œμ§€κ°€ λͺ¨λ‹ˆν„°λ§ μ‹œμŠ€ν…œμ—μ„œ μš”κ΅¬ν•˜λŠ” νŽ˜μ΄λ‘œλ“œ ν˜•μ‹μ— λ§žλŠ”μ§€λŠ” μ•Œ 수 μ—†λ‹€. AWS IoT 의 λ©”μ‹œμ§€ λΌμš°νŒ… κ·œμΉ™μ— λ§žλŠ” μ£Όμ œμ— λ©”μ‹œμ§€λ₯Ό κ²Œμ‹œν•˜κ³  (λ©”μ‹œμ§€λ₯Ό SQS λŒ€κΈ°μ—΄λ‘œ μ „μ†‘ν•˜λ„λ‘ μ„€μ •ν–ˆλ‹€λ©΄) Amazon SQS 에 κ²Œμ‹œλœ λ©”μ‹œμ§€κ°€ μ „λ‹¬λ˜λŠ”μ§€κΉŒμ§€ ν™•μΈν•΄μ•Όν•œλ‹€. λ˜ν•œ, κ²Œμ‹œν•œ λ©”μ‹œμ§€μ— ν¬ν•¨λœ 데이터가 λͺ¨λ‹ˆν„°λ§ μ‹œμŠ€ν…œμ—μ„œ 확인할 수 μžˆλŠ” κ°„λ‹¨ν•œ μ›ΉνŽ˜μ΄μ§€λ₯Ό κ°œλ°œν•˜λŠ” 것도 쒋은 방법이닀.

AWS IoT 루트 CA μΈμ¦μ„œ

ν…ŒμŠ€νŠΈ μ½”λ“œμ—μ„œ 루트 CA μΈμ¦μ„œκ°€ λ°˜λ“œμ‹œ ν•„μš”ν•œ 것은 μ•„λ‹ˆμ§€λ§Œ 컴퓨터 ν™˜κ²½λ§ˆλ‹€ μ‹ λ’° κ°€λŠ₯ν•œ μΈμ¦μ„œ λͺ©λ‘μ— ν¬ν•¨λ˜μ–΄μžˆμ§€ μ•Šμ„ 수 있기 λ•Œλ¬Έμ— μƒν˜Έ 인증(mTLS)을 μˆ˜ν–‰ν•˜κΈ° μœ„ν•΄μ„œ λͺ…μ‹œμ μœΌλ‘œ μ§€μ •ν•  ν•„μš”κ°€ μžˆλ‹€. AWS IoT μ—μ„œ μΈμ¦μ„œλ₯Ό λ‹€μš΄λ‘œλ“œ λ°›κ²Œ 되면 Amazon Trust Services Repository μ—μ„œ μ œκ³΅ν•˜λŠ” AmazonRootCA1.pem 와 AmazonRootCA3.pemκ°€ ν¬ν•¨λ˜λŠ”λ° RSA 2048λΉ„νŠΈ 킀에 ν•΄λ‹Ήλ˜λŠ” AmazonRootCA1 κ°€ 일반적으둜 μ‚¬μš©λœλ‹€. νšŒμ‚¬ λ‚΄ 개발 ν™˜κ²½μœΌλ‘œ κ΅¬μ„±λœ AWS IoT 에 μ—°κ²°ν•΄λ³Έ κ²°κ³Όλ‘œλŠ” ECC 256λΉ„νŠΈ 킀에 ν•΄λ‹Ήλ˜λŠ” AmazonRootCA3λ₯Ό μ‚¬μš©ν•˜λ‹ˆ awscrt.exceptions.AwsCrtError: AWS_IO_TLS_ERROR_NEGOTIATION_FAILURE: TLS (SSL) negotiation failed λΌλŠ” 였λ₯˜κ°€ λ°œμƒν–ˆλ‹€.