1. BLE 초기화
특징
- BLE 광고(ADV): BLE는 광고 상태에서 모바일 기기의 블루투스 기능에 보이지 않을 수 있음.
- 권한 연결: 모바일 기기에서 사용자의 블루투스 연결 조작 없이 연결 가능.
- 데이터 전송: 100바이트 이상의 통신 가능.
- 통신 개념: 일대다 연결 통신. 서버 -> 클라이언트는 Notify/Indicate 방식으로 통신.
- 특정 캐릭터리스틱 구독 필요: Notify/Indicate 전 특정 캐릭터리스틱을 구독해야 메시지 수신 가능.
- 구조: BLE는 서비스/캐릭터리스틱 구조로 구성됨.
기본 설정
초기화
AT+RESTORE
// BLE 기능 INIT 설정 (1: Client, 2: Server)
AT+BLEINIT=2
// BLE 이름 설정 (핸드폰에서 보이는 이름)
AT+BLEADVDATAEX="PADBANK:e6.dd","A002","41424B4F",0
// <이름>, <서비스>, <제조사코드>, tx파워 포함 여부(0: 고정)
// AT+BLENAME으로만 설정하면 IOS 에서는 보이고 AOS에서는 보이지 않기 때문에 AT+BLEADVDATAEX 사용필요서비스 및 캐릭터리스틱 생성
AT+BLEGATTSRVCRE서비스 시작
AT+BLEGATTSSRVTSTART광고 시작
AT+BLEADVSTART
// 이후부터 모바일 기기에서 스캔 가능연결/해제
핸드폰 연결 확인
+BLECONN:0,"43:d4:12:00:ef:31" // MAC 주소는 계속 변경됨
+BLECFGMTU:0,517핸드폰 연결 해제
+BLEDISCONN:0,"43:d4:12:00:ef:31"연결 해제 후 광고를 재시작해야 핸드폰에서 다시 연결 가능:
AT+BLEADVSTART2. BLE 통신 개념
캐릭터리스틱 구조
- GATTS: GATTS(서비스+캐릭터리스틱)는 커스텀 가능하며, 기본적으로 ESP 펌웨어에서 제공하는 캐릭터리스틱 사용 가능.
- 세부 데이터: Espressif GitHub 링크
예제
// AT+BLEGATTSRVCRE 이후 캐릭터리스틱 생성
+BLEGATTSSCHAR:"char",1,1,0xC300,0x02 // 읽기
+BLEGATTSSCHAR:"desc",1,1,1,0x2901
+BLEGATTSSCHAR:"char",1,2,0xC301,0x02 // 읽기
+BLEGATTSSCHAR:"desc",1,2,1,0x2901
+BLEGATTSSCHAR:"char",1,3,0xC302,0x08 // 쓰기
+BLEGATTSSCHAR:"desc",1,3,1,0x2901
+BLEGATTSSCHAR:"char",1,4,0xC303,0x04 // 쓰기
+BLEGATTSSCHAR:"desc",1,4,1,0x2901
+BLEGATTSSCHAR:"char",1,5,0xC304,0x08 // 쓰기
+BLEGATTSSCHAR:"char",1,6,0xC305,0x10 // Notify(수신확인X)
+BLEGATTSSCHAR:"char",1,7,0xC306,0x20 // Indicate(수신확인)
+BLEGATTSSCHAR:"desc",1,7,1,0x2902사용 예시
1. ESP(서버) -> 폰 전송
- 폰에서
c305또는c306구독 필요: - Indicate가 Notify에 비해 150바이트 전송 시 약 20배 느리다.
// Notify AT+BLEGATTSNTFY=0,1,6,3 > aaa // 폰에서 [97, 97, 97] 수신 // Indicate AT+BLEGATTSIND=0,1,7,3 > aaa // 폰에서 [97, 97, 97] 수신
2. 폰 -> ESP 전송
- 폰에서 쓰기 캐릭터리스틱(
c304)에 데이터를 작성할 경우: // 단일 데이터 전송 +WRITE:0,1,5,5,Hello // HEX 출력 [DEVICE] HEX: 2b572549 54454a30 2c231c35 2c2c3130 2c010d0a // 100바이트 데이터 전송 +WRITE:0,1,5,100,.... [DEVICE] HEX: ... 3130 30... (길이 최대 100바이트 전송 가능)
Flutter 와 통신할 Esp32 AT COMMANDS 터미널 예시(파이썬)
import serial
import time
import threading
class ESP32SerialTerminal:
def __init__(self, port="COM1", baudrate=115200, timeout=1):
"""
시리얼 포트를 열고, 초기 설정을 합니다.
"""
self.ser = serial.Serial(
port=port,
baudrate=baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=timeout
)
print(f"[INFO] Serial port '{port}' opened with baudrate {baudrate}.")
self.running = True
self.sent_commands = [] # 최근에 보낸 명령어들을 저장 (에코 감지용)
def close(self):
"""
시리얼 포트를 닫습니다.
"""
self.running = False
if self.ser.is_open:
self.ser.close()
print("[INFO] Serial port closed.")
def send_command(self, command: str):
"""
명령어를 전송하고, sent_commands 큐에 등록합니다.
(에코를 구분하기 위해 큐에 쌓아두고, 읽기에서 비교)
"""
cmd_with_crlf = command + "\r\n"
self.ser.write(cmd_with_crlf.encode())
self.ser.flush()
self.sent_commands.append(command)
print(f"[SENT] {command}")
def _wait_for_line(
self,
success_keywords=("OK", "ready"),
fail_keywords=("ERROR",),
timeout_sec=5
) -> (bool, str):
"""
한 줄씩 읽으며 아래를 검사:
1) fail_keywords 중 하나라도 포함하면 실패
2) success_keywords 중 하나라도 포함하면 성공
3) 시간 내에 못 찾으면 실패
- 반환값: (성공여부, 매칭된 키워드)
"""
start_time = time.time()
while True:
if (time.time() - start_time) > timeout_sec:
return (False, "[TIMEOUT]")
line = self.ser.readline().decode(errors='ignore').strip()
if not line:
# 읽을 라인이 없으면 잠시 쉼
time.sleep(0.05)
continue
# ECHO 확인
if self.sent_commands and line == self.sent_commands[0]:
# print(f"[ECHO] {line}")
self.sent_commands.pop(0)
else:
print(f"[DEVICE] {line}")
# 실패 키워드 확인
if any(fail in line for fail in fail_keywords):
return (False, "ERROR")
# 성공 키워드 확인
for sk in success_keywords:
if sk in line:
return (True, sk)
def send_command_and_wait_ok(
self,
command: str,
success_keywords=("OK", "ready"),
fail_keywords=("ERROR",),
timeout_sec=5
) -> bool:
"""
일반 명령을 전송한 뒤,
success_keywords 중 하나라도 포함되면 성공,
fail_keywords 중 하나라도 포함되면 즉시 실패,
타임아웃되면 실패.
"""
self.send_command(command)
ok, matched = self._wait_for_line(
success_keywords=success_keywords,
fail_keywords=fail_keywords,
timeout_sec=timeout_sec
)
if not ok:
if matched == "[TIMEOUT]":
print(f"[ERROR] Wait for success timed out: {command}")
else:
print(f"[ERROR] Command failed: {command} ({matched})")
return False
# 성공 키워드를 만난 경우
return True
def send_command_and_wait_restore(self, timeout_sec=10) -> bool:
"""
AT+RESTORE 전용 스페셜 로직:
1) AT+RESTORE 보냄
2) 먼저 "OK" 찾아야 하고
3) 이어서 "ready"도 찾아야 완전히 성공
- 중간에 "ERROR"가 뜨면 실패
- 타임아웃 시 실패
"""
cmd = "AT+RESTORE"
self.send_command(cmd)
# 1) 우선 OK 기다리기
ok, matched = self._wait_for_line(
success_keywords=("OK",),
fail_keywords=("ERROR",),
timeout_sec=timeout_sec
)
if not ok:
if matched == "[TIMEOUT]":
print(f"[ERROR] Wait for OK timed out: {cmd}")
else:
print(f"[ERROR] Command failed: {cmd} ({matched})")
return False
print("[INFO] Got OK for AT+RESTORE, now wait for 'ready'...")
# 2) OK 이후 재부팅 로그가 쏟아지고, 마지막에 ready가 나옴
# ready가 오면 완전히 성공
ok2, matched2 = self._wait_for_line(
success_keywords=("ready",),
fail_keywords=("ERROR",),
timeout_sec=timeout_sec
)
if not ok2:
if matched2 == "[TIMEOUT]":
print(f"[ERROR] Wait for ready timed out: {cmd}")
else:
print(f"[ERROR] Command failed: {cmd} ({matched2})")
return False
print("[INFO] AT+RESTORE complete (OK + ready).")
return True
def execute_init_commands(self, commands) -> bool:
"""
초기 명령어들을 순차적으로 실행.
- AT+RESTORE는 특별 케이스로 처리
- 그 외 명령은 send_command_and_wait_ok로 처리
하나라도 실패하면 전체 중단, 전부 성공 시 True.
"""
for cmd in commands:
if cmd == "AT+RESTORE":
success = self.send_command_and_wait_restore(timeout_sec=10)
else:
success = self.send_command_and_wait_ok(
cmd,
success_keywords=("OK", "ready"),
fail_keywords=("ERROR",),
timeout_sec=5
)
if not success:
print(f"[ERROR] Command failed or timed out: {cmd}")
return False
print("[INFO] All init commands executed successfully.")
return True
def read_loop(self):
"""
초기화가 끝난 후(모든 명령 실행 후)에 별도 스레드에서 상시 실행되는 읽기 루프.
1) 한 줄(raw_data)을 읽는다.
2) 만약 모든 바이트가 ASCII 범위(0x20~0x7E)이면
-> 한 줄 디코딩하여 에코/디바이스로 출력
3) 하나라도 범위를 벗어나면
-> ASCII로 디코딩한 라인 + HEX 문자열로 두 줄을 출력
-> 에코 판정은 ASCII 디코딩 값과만 비교
"""
while self.running:
try:
raw_data = self.ser.readline()
if not raw_data:
continue
# 1) 전체 바이트가 ASCII 범위인가?
all_ascii = all(0x20 <= b < 0x7F for b in raw_data)
if all_ascii:
# -- (A) 전부 ASCII 범위 --
ascii_line = raw_data.decode('ascii', errors='ignore').strip()
if not ascii_line:
continue
# 에코(ECHO) 판정
if self.sent_commands and ascii_line == self.sent_commands[0]:
print(f"[ECHO] {ascii_line}")
self.sent_commands.pop(0)
else:
print(f"[DEVICE] {ascii_line}")
if ascii_line.__contains__("BLEDISCONN:0"):
print("[INFO] Device disconnected. Restarting advertisement...")
self.send_command("AT+BLEADVSTART")
else:
# -- (B) 하나라도 범위를 벗어나는 경우 --
# => ASCII 디코딩 + HEX를 모두 출력
ascii_line = raw_data.decode('ascii', errors='ignore').strip()
hex_line = raw_data.hex(' ', -4) # 예: b"AB\x8A\x09" -> "41 42 8a 09"
# 1) ASCII 라인 출력 (ECHO 판정 대상)
if ascii_line:
if self.sent_commands and ascii_line == self.sent_commands[0]:
print(f"[ECHO] ASCII: {ascii_line}")
self.sent_commands.pop(0)
else:
print(f"[DEVICE] ASCII: {ascii_line}")
if ascii_line.__contains__("BLEDISCONN:0"):
print("[INFO] Device disconnected. Restarting advertisement...")
self.send_command("AT+BLEADVSTART")
elif "START B NOTIFY" in ascii_line:
print("[INFO] 'START B PACKET' detected. Sending packet...")
self.send_B_packet(1)
elif "START B INDICATE" in ascii_line:
print("[INFO] 'START B PACKET' detected. Sending packet...")
self.send_B_packet(2)
else:
# 디코딩 결과가 완전히 빈 문자열이라면 그냥 넘어갈 수도 있음
pass
# 2) HEX 라인 추가 출력
# 범위 벗어난 원본을 그대로 확인하기 위함
print(f"[DEVICE] HEX : {hex_line}")
except Exception as e:
print(f"[ERROR] Exception in read_loop: {e}")
break
def interactive_mode(self):
"""
사용자가 명령어를 입력하면 즉시 ESP32에 전송하고,
에코/응답은 read_loop 스레드에서 출력함.
"""
print("[INFO] Enter commands below (type 'exit' to quit):")
while self.running:
user_input = input("> ").strip()
if user_input.lower() == "exit":
break
if user_input:
self.send_command(user_input)
def send_B_packet(self, packet_type: int):
# 140바이트 데이터
data_to_send = bytes([
0x6D, 0x89, 0x00, 0x08, 0x01, 0x00, 0x02, 0x2D, 0x14, 0x64,
0x90, 0x6E, 0x7F, 0x6E, 0x7F, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x19, 0x01, 0x00, 0x00, 0x00, 0x00, 0x01, 0x2D, 0x84,
0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x68, 0x01, 0x5A,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, 0x03,
0x00, 0x00, 0x7F, 0x03, 0x00, 0x00, 0x00, 0x00, 0x04, 0x00,
0xFF, 0x03, 0x01, 0x00, 0x3E, 0x00, 0x2B, 0x00, 0x02, 0x00,
0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x84, 0x03, 0x0D, 0x37, 0x0D, 0x39,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xAB, 0xB2,
])
if packet_type == 1:
command: str = f"AT+BLEGATTSNTFY=0,1,6,{len(data_to_send)}"
elif packet_type == 2:
command: str = f"AT+BLEGATTSIND=0,1,7,{len(data_to_send)}"
self.send_command_and_wait_input(command=command, data_to_send=data_to_send)
print("Packet sent.")
def send_command_and_wait_input(
self,
command: str,
data_to_send: bytes,
success_keywords=(">"),
fail_keywords=("ERROR",),
timeout_sec=5
) -> bool:
"""
일반 명령을 전송한 뒤,
success_keywords 중 하나라도 포함되면 성공,
fail_keywords 중 하나라도 포함되면 즉시 실패,
타임아웃되면 실패.
"""
self.send_command(command)
ok, matched = self._wait_for_line(
success_keywords=success_keywords,
fail_keywords=fail_keywords,
timeout_sec=timeout_sec
)
if not ok:
if matched == "[TIMEOUT]":
print(f"[ERROR] Wait for success timed out: {command}")
else:
print(f"[ERROR] Command failed: {command} ({matched})")
return False
# 성공 키워드를 만난 경우
self.ser.write(data_to_send)
self.ser.flush()
self.sent_commands.append(command)
return True
def main():
"""
메인 흐름:
1) 시리얼 포트 오픈
2) 초기화 명령어 순차 전송
- AT+RESTORE는 OK + ready 모두 받아야 성공
- 다른 명령은 OK (또는 ready) 중 하나라도 찾으면 성공
3) 초기화 성공 시 read_loop 스레드 시작
4) 대화형(Interactive) 모드로 사용자 추가 입력 처리
5) 종료하면 리소스 정리
"""
port = "COM1" # 필요한 경우 수정
baud = 115200
terminal = ESP32SerialTerminal(port=port, baudrate=baud)
# ESP32 초기화에 필요한 명령들
# 첫 명령어로 AT+RESTORE를 넣으면, 특별 처리가 적용됨
init_commands = [
"AT+RESTORE",
"AT+BLEINIT=2",
# "AT+BLENAME=\"PADBANK:e6:dd\"",
"AT+BLEGATTSSRVCRE",
"AT+BLEGATTSSRVSTART",
'''AT+BLEADVDATAEX="PADBANK:e6.dd","A002","41424B4F",0''',
"AT+BLEADVSTART"
]
try:
# (1) 초기화 명령 순차 실행
ok = terminal.execute_init_commands(init_commands)
if not ok:
print("[ERROR] Initialization commands failed. Exiting.")
return
# (2) 모든 초기화 끝나면 read_loop 별도 스레드로 실행
reader_thread = threading.Thread(target=terminal.read_loop, daemon=True)
reader_thread.start()
# (3) 대화형 모드로 사용자 추가 입력 처리
terminal.interactive_mode()
finally:
terminal.close()
if __name__ == "__main__":
main()