ESP32 Networking
ตัวอย่างการใช้งานไมโครไพธอนและบอร์ด ESP32 สำหรับเครือข่ายไร้สาย โดยใช้ MicroPython v1.13 (built on 2020-09-02)
1. ESP32 Networking
ESP32 สามารถเชื่อมต่อกับอุปกรณ์อื่นแบบไร้สายได้ โดยใช้ Wi-Fi ตามมาตรฐาน IEEE 802.11b/g/n (2.4GHz) และ Bluetooth และมีโหมดการทำงานสำหรับ Wi-Fi ดังนี้
STA (Station)
AP (Access Point)
Soft-AP (both STA and AP)
แต่ถ้าจะเชื่อมต่อกับ Ethernet จะต้องมีวงจร Ethernet PHY Transceiver (RMII Interface) นำมาต่อเพิ่ม เช่น LAN8720 หรือ TLK110 (10/100 Mbps) เป็นต้น ตัวอย่างบอร์ด ESP32 ที่มี Ethernet/LAN Port ได้แก่
Espressif Systems ESP32-Ethernet-Kit (ESP32-WROVER-B + IP101GRI Transceiver)
LilyGo / TTGO T-Internet-POE Board (ESP32-WROOM + LAN8720A Transceiver)
Wireless Tag WT32-ETH01 (ESP32-based WT32-S1 Wi-SoC + LAN8720A transceiver)
สำหรับ ESP32 การเชื่อมต่อด้วย Wi-Fi คงเป็นวิธีที่ง่ายและสะดวกที่สุด โดยทั่วไปแล้ว เราจะโปรแกรมให้อุปกรณ์ ESP32 ทำงานในโหมด STA และเชื่อมต่อกับอุปกรณ์ Wi-Fi Access Point หรือ Router ในระบบเครือข่าย WLAN และทำให้สามารถเชื่อมต่อไปยังอินเทอร์เน็ตได้
ถ้าจะให้ ESP32 ทำงานในโหมด AP หรือ Soft-AP ก็อนุญาตให้อุปกรณ์อื่นเชื่อมต่อผ่าน Wi-Fi เข้ามายัง ESP32 ได้ เราอาจจะเห็นตัวอย่างการใช้งาน ESP32 ที่ทำงานในโหมดดังกล่าวและทำหน้าที่เป็น Light-weight Web Server ในตัวด้วย ทำให้คอมพิวเตอร์อื่นสามารถเชื่อมต่อผ่าน Wi-Fi มายัง ESP32 และผู้ใช้สามารถเปิด Web Browser เพื่อเข้าสู่หน้าเว็บของ ESP32 เช่น ใช้ในการตั้งค่าอุปกรณ์ผ่านหน้าเว็บ หรือดูสถานะบางอย่างของระบบ เป็นต้น
อีกตัวเลือกหนึ่งที่น่าสนใจสำหรับ ESP32 คือ การสื่อสารแบบไร้สายในรูปแบบที่เรียกว่า ESP-NOW โดยไม่จำเป็นต้องใช้ Wi-Fi Infrastructure แต่ใช้รูปแบบการทำงานแบบ Peer-to-Peer และในปัจจุบัน เราก็สามารถใช้ ESP-NOW (Micropython Port) ร่วมกับไมโครไพธอนได้เช่นกัน
2. ESP32 Wi-Fi Scan
เริ่มต้นตัวอย่างแรกด้วยการสแกนหาอุปกรณ์หรือเครือข่ายไร้สาย Wi-Fi โดยจะต้องทำให้ ESP32 เปิดใช้งานในโหมด STA ก่อน และสามารถใช้คำสั่งจากไลบรารี network ดังนี้ โดยทำการสแกนซ้ำไปเรื่อย ๆ และถ้าต้องการหยุดหรือจบการทำงาน ให้กด Ctrl+C ในหน้าต่าง Interactive Shell ของ REPL
import network
import utime as time
# open WiFi interface in STA mode
wifi = network.WLAN( network.STA_IF )
# activate the WiFi interface (if_up)
wifi.active( True )
print ('Press Ctrl+C to stop WiFi scanning...')
try:
while True:
# scan WiFi
scan_results = wifi.scan()
# print scan results (a list of tuples)
print(60*'=')
for ap in scan_results:
print( ap )
print(60*'-')
except KeyboardInterrupt:
print('Terminated...')
# deactivate the WiFi interface
wifi.active(False)เมื่อสแกนเครือข่าย Wi-Fi ข้อมูลสำหรับแต่ละอุปกรณ์ที่พบ จะประกอบด้วยชื่ออ้างอิงในการเชื่อมต่อ (SSID), หมายเลขของอุปกรณ์ (MAC address), ช่องสัญญาณความถี่ (Channel), ค่าตัวชี้วัดความแรงของสัญญาณที่รับได้ (RSSI), โหมดการเข้ารหัสสำหรับความปลอดภัย (Authentication Mode) และมีการซ่อนชื่อ SSID หรือไม่ (Hidden) ตามลำดับ
ลองมาดูตัวอย่างการเขียนโค้ดไมโครไพธอนที่ใช้สำหรับสแกนระบบเครือข่าย Wi-Fi ที่อยู่โดยรอบ แล้วแปลงผลที่ได้ให้เป็นข้อมูลแบบ JSON เช่น แสดงชื่อ SSID รหัสเครื่อง (MAC Address) โหมดการตั้งค่าความปลอดภัย (Authentication Mode) และช่องสัญญาณ เป็นต้น
import network
import ujson as json
param_names = ['essid','mac','channel','rssi','authmode','hidden']
authmodes = ['OPEN','WEP','WPA-PSK',
'WPA2-PSK','WPA/WPA2-PSK','MAX']
def mac_bytes_to_hex_str( mac_bytes ):
return ':'.join(['%02X' % x for x in mac_bytes])
def convert_to_json( found_list ):
results = {'count': len(found_list), 'found_list': []}
for ap in found_list:
pairs = dict(zip(param_names,ap))
for name in pairs:
value = pairs.get(name)
if type(value)==bytes:
if name=='mac' and len(value)==6:
value = mac_bytes_to_hex_str( value )
else:
value = value.decode('ascii')
else:
try:
value = int(value)
if name=='authmode' and 0 <= value <= 5:
value = authmodes[value]
except (ValueError, TypeError):
pass
pairs[name] = value
results['found_list'].append( pairs )
return results# open WiFi interface in STA mode
# activate the WiFi interface
wifi = network.WLAN( network.STA_IF )
wifi.active(True)
# scan WiFi and get the results as a list of tuples
scan_results = wifi.scan()
# deactivate the WiFi interface
wifi.active(False)
# convert a list of tuples to json data
results = convert_to_json( scan_results )
if results.get('count') > 0:
for item in results.get('found_list'):
print( json.dumps(item) ) # show JSON string
else:
print( 'No WiFi found')3. Wi-Fi Connection in STA Mode
การเชื่อมต่อกับเครือข่าย Wi-Fi ในโหมด STA จะต้องระบุชื่อ SSID และรหัสผ่าน (Password) สำหรับเชื่อมต่ออุปกรณ์ไร้สาย เช่น Wireless Access Point / Router ตามตัวอย่างโค้ดต่อไปนี้
ในตัวอย่างนี้ มีการสร้างฟังก์ชันชื่อ connect_wifi() เพื่อใช้ในการเชื่อมต่อ Wi-Fi (STA Mode) เมื่อเรียกใช้ จะต้องระบุอาร์กิวเมนต์เป็นโครงสร้างข้อมูลแบบ dictionary อ้างอิงโดยตัวแปร WIFI_CFG ที่มีชื่อสมาชิก ‘ssid’ และ ‘pwd’ ตามลำดับ ดังนั้นให้กำหนดค่าให้ถูกต้องเมื่อนำไปใช้งาน
เมื่อเรียกใช้ฟังก์ชันนี้ และสามารถเชื่อมต่อได้สำเร็จ จะได้ค่ากลับคืนเป็นอ็อบเจกต์ที่ไม่ใช่ None และเราก็สามารถใช้คำสั่ง wifi.ifconfig() เพื่อดูข้อมูลเกี่ยวกับการเชื่อมต่อเครือข่าย เช่น หมายเลขของไอพีที่ได้รับ (IP address) หมายเลขของไอพีของอุปกรณ์ Gateway เป็นต้น
import network
import utime as time
# Specify the SSID and password of your WiFi network
WIFI_CFG = { 'ssid': "XXXXXX", 'pwd': "XXXXXXXXX" }
def connect_wifi( wifi_cfg, max_retries=10 ):
# use WiFi in station mode (not AP)
wifi_sta = network.WLAN( network.STA_IF )
# activate the WiFi interface (up)
wifi_sta.active(True)
# connect to the specified WiFi AP
wifi_sta.connect( wifi_cfg['ssid'], wifi_cfg['pwd'] )
retries = 0
while not wifi_sta.isconnected():
retries = retries + 1
if retries >= max_retries:
return None
time.sleep_ms(500)
return wifi_sta
# try to connect the network
wifi = connect_wifi( WIFI_CFG )
if wifi is None:
print( 'WiFi connection failed' )
else:
ipaddr, netmask, gateway, dns = wifi.ifconfig()
print("============================")
print("IP address :", ipaddr)
print("Net mask :", netmask)
print("Gateway :", gateway)
print("DNS server :", dns)
print("----------------------------")4. Wi-Fi in AP Mode + Simple HTTP Server
ตัวอย่างถัดไป สาธิตการเปิดใช้งาน Wi-Fi ในโหมด AP แบบ Open (ไม่มีการป้องกันด้วยรหัสผ่าน) โดยการสร้างและเรียกใช้ฟังก์ชัน start_wifi_ap() จากนั้นเมื่อ ESP32 ทำงานเป็น Wi-Fi AP แล้ว (จะมีหมายเลขไอพีตรงกับ 192.168.4.1 และเลือกใช้ช่องสัญญาณหมายเลข 11) ยังได้มีการสร้างและเรียกใช้ฟังก์ชัน start_web_server() เพื่อให้สามารถทำงานเป็น Web Server (HTTP protocol) อย่างง่ายได้ด้วย แสดงข้อความเป็นหน้าเว็บได้
ในตัวอย่างนี้ ถ้าต้องการจบการทำงานของ Web Server เราจะใช้วิธีตรวจสอบการกดปุ่ม ดังนั้นจะต้องมีการต่อวงจรกดปุ่มภายนอกแบบ Active-Low และเลือกใช้ขา GPIO23 เป็นอินพุตสำหรับปุ่มกด
import network
import utime as time
import usocket as socket
from machine import Pin
AP_CFG = { 'essid': 'MyAP', # SSID for your AP
'authmode': network.AUTH_OPEN, 'channel': 11, }
def start_wifi_ap( ap_cfg ):
# open WiFi in AP mode
ap = network.WLAN( network.AP_IF )
# activate AP interface (up)
ap.active(True)
# configure the AP interface
try:
ap.config( **ap_cfg )
except Exception as ex:
print ( 'Cannot set AP configuration...' )
if ap.active():
return ap
else:
return False
def start_web_server( btn=None ):
global sock
# create a socket (STREAM TCP socket) for network connection
sock = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
# set socket option: reuse address
sock.setsockopt( socket.SOL_SOCKET, socket.SO_REUSEADDR, 1 )
# use socket in non-blocking mode
sock.setblocking( False )
# get the server address (localhost)
addr = socket.getaddrinfo('0.0.0.0',80)[0][-1]
# bind the socket to port 80 (HTTP)
sock.bind( addr )
# listen for an incoming connection
sock.listen(1)
while True:
try:
# waiting for connection
conn,addr = sock.accept()
except OSError:
if btn and btn.value()==0: # check button
sock.close() # close socket
break # break the loop
else:
continue
# read incoming request data
request = conn.recv(1024)
print( 'Content = %s' % str(request) )
# send HTML as response
html = '<html><head></head><body>'
html += '<h1>Welcome to ESP32...</h1><br>'
html += '<b>Client from {}<b><br>'.format(str(addr))
html += '</body></html>'
conn.send( html ) # send HTML response
conn.close() # close HTTP connection
time.sleep_ms(10)
sock = None
ap = start_wifi_ap( AP_CFG )
ipaddr = ap.ifconfig()[0]
print ( 'IP address:', ipaddr ) # 192.168.4.1
BTN_GPIO = const(23) # use GPIO23 for push button
btn = Pin( BTN_GPIO, Pin.IN, Pin.PULL_UP )
try:
start_web_server( btn ) # start web server
except KeyboardInterrupt:
pass
finally:
if sock:
sock.close()
ap.active(False) # turn off WiFi APถ้าจะเปลี่ยนจาก Open เป็นการตั้งรหัสป้องกันด้วยวิธี WPA2-PSK ก็กำหนดค่าสำหรับ Configuration ตามตัวอย่างดังนี้
AP_CFG = { 'essid': 'MyAP', 'password': 'YOUR_STRONG_PASSWORD',
'authmode': network.AUTH_WPA2_PSK, 'channel':11, }
5. Getting Date & Time From NTP Server
ตัวอย่างถัดไปสาธิตการเปิดใช้งาน ESP32 ในโหมด STA และเชื่อมต่อกับอุปกรณ์ Wi-Fi AP ที่สามารถเชื่อมต่อไปยังอินเทอร์เน็ตได้ และใช้คำสั่ง ntptime.settime() เพื่อเชื่อมต่อกับคอมพิวเตอร์ในอินเทอร์เน็ตที่ทำหน้าที่เป็น NTP Server และอัปเดทเวลาปัจจุบันสำหรับการทำงานของวงจรของ RTC (Real-Time Clock) ของ ESP32
import network
import utime as time
import ntptime
import machine
# Specify the SSID and password of your WiFi network
WIFI_CFG = { 'ssid': "XXXX", 'pwd': "XXXXXXX" }
def connect_wifi( wifi_cfg ):
wifi_sta = network.WLAN( network.STA_IF )
wifi_sta.active(True)
wifi_sta.connect( wifi_cfg['ssid'], wifi_cfg['pwd'] )
while not wifi_sta.isconnected():
time.sleep(1.0)
print(wifi_sta.ifconfig())
def sync_ntp():
while True:
try:
# synchronize RTC with NTP server
ntptime.settime()
break
except OSError:
print('NTP server: connection timeout...')
connect_wifi( WIFI_CFG )
sync_ntp()
rtc = machine.RTC()
tm = rtc.datetime() # get current RTC datetime (now)
tz_offset = +7 # timezone offset (GMT+7)
hh, mm, ss = (tm[4]+tz_offset) % 24, tm[5], tm[6]
print( 'Time: {}:{}:{}'.format( hh, mm, ss ) )
y, m, d = tm[0], tm[1], tm[2]
print( 'Date: {}-{:02d}-{:02d}'.format( y, m, d ) )6. HTTP GET: Getting Datetime Using WorldTime API
ตัวอย่างถัดไปสาธิตการเปิดใช้งาน ESP32 ในโหมด STA และเชื่อมต่อกับอุปกรณ์ Wi-Fi AP ที่สามารถเชื่อมต่อไปยังอินเทอร์เน็ตได้ และร้องขอข้อมูลแบบ HTTP Request / GET Method ไปยัง World Time API ซึ่งเป็นผู้ให้บริการข้อมูลเกี่ยวกับวันและเวลาในปัจจุบันตามโซนเวลา (Timezone) ที่ระบุไว้ ข้อมูลที่ได้รับกลับมานั้น จะอยู่ในรูปแบบของ JSON String
import network
import utime as time
import ujson as json
import urequests as requests
URL = "http://worldtimeapi.org/api/timezone/Asia/Bangkok"
# Specify the SSID and password of your WiFi network
WIFI_CFG = { 'ssid': "XXXX", 'pwd': "XXXXXXX" }
def connect_wifi( wifi_cfg, max_retries=10 ):
# use WiFi in station mode (not AP)
wifi_sta = network.WLAN( network.STA_IF )
# activate the WiFi interface (up)
wifi_sta.active(True)
# connect to the specified WiFi AP
wifi_sta.connect( wifi_cfg['ssid'], wifi_cfg['pwd'] )
retries = 0
while not wifi_sta.isconnected():
retries = retries + 1
if retries >= max_retries:
return None
time.sleep_ms(500)
return wifi_sta
def get_worldtime_data( url ):
resp = requests.get( url )
if resp.status_code==200: # request ok
try:
data = json.loads(resp.text)
except Exception as ex:
print ('JSON data error...' )
return
print( 'Timezone:', data['timezone'] )
print( 'Epoch time (Jan 1, 1970):', int(data['unixtime']) )
print( 'UTC datetime:', data['utc_datetime'] )
print( 'Local datetime:', data['datetime'] )
print( 'UTC offset (hours):', data['utc_offset'] )
print( 'UTC offset (seconds):', int(data['raw_offset']) )
else:
print ('HTTP request error...')
# try to connect the network
wifi = connect_wifi( WIFI_CFG )
if wifi is not None:
get_worldtime_data( URL )
else:
print('No WiFi connection')ตัวอย่างเอาต์พุตที่ได้
Timezone: Asia/Bangkok
Epoch time (Jan 1, 1970): 1609939805
UTC datetime: 2021-01-06T13:30:05.138524+00:00
Local datetime: 2021-01-06T20:30:05.138524+07:00
UTC offset (hours): +07:00
UTC offset (seconds): 25200
7. HTTPS GET: Thailand's COVID-19 Status Update
ตัวอย่างถัดไป สาธิตการดึงข้อมูลเกี่ยวกับผู้ป่วย COVID-19 สำหรับประเทศไทย ด้วยวิธี HTTPS GET จาก API ของเว็บ covid19.th-stat.com ข้อมูลที่ได้จะอยู่ในรูปของ JSON String
import network
import utime as time
import urequests as requests
URL = "https://covid19.th-stat.com/api/open/today"
# Specify the SSID and password of your WiFi network
WIFI_CFG = { 'ssid': "XXXX", 'pwd': "XXXXXXX" }
WIFI_CFG = { 'ssid': "esl_ap", 'pwd': "cnch2687" }
COVID19_NAMES = [
('Confirmed', u'ยอดผู้ป่วยสะสม'),
('Recovered', u'ผู้ป่วยรักษาหายแล้ว'),
('Hospitalized', u'ผู้ป่วยรักษาตัวในโรงพยาบาล'),
('Deaths', u'ผู้เสียชีวิตสะสม'),
('NewConfirmed', u'ผู้ป่วยตรวจพบเพิ่มวันนี้'),
('NewRecovered', u'ผู้ป่วยรักษาหาย'),
('NewDeaths', u'ผู้เสียชีวิตวันนี้'),
('UpdateDate', u'อัปเดตล่าสุด') ]
def get_covid19_data( url ):
data = None
try:
resp = requests.get( url )
data = resp.json()
for name in COVID19_NAMES[:-1]:
if name[0] in data:
value = str(data[name[0]])
print('{}:\t\t\t\t{}'.format(name[1],value) )
name = COVID19_NAMES[-1]
update = data[ name[0] ]
print( '{}: {}'.format(name[1], update) )
except Exception as ex:
print('error:', ex)
return data
def connect_wifi( wifi_cfg ):
wifi_sta = network.WLAN( network.STA_IF )
wifi_sta.active(True)
wifi_sta.connect( wifi_cfg['ssid'], wifi_cfg['pwd'] )
while not wifi_sta.isconnected():
time.sleep(1.0)
print(wifi_sta.ifconfig())
connect_wifi( WIFI_CFG )
import gc
gc.collect()
get_covid19_data( URL )ตัวอย่างข้อความเอาต์พุตที่ได้
ยอดผู้ป่วยสะสม: 9331
ผู้ป่วยรักษาหายแล้ว: 4418
ผู้ป่วยรักษาตัวในโรงพยาบาล: 4847
ผู้เสียชีวิตสะสม: 66
ผู้ป่วยตรวจพบเพิ่มวันนี้: 365
ผู้ป่วยรักษาหาย: 21
ผู้เสียชีวิตวันนี้: 1
อัปเดตล่าสุด: 06/01/2021 12:25
8. Getting Air Quality & Weather Data from AirVisual
ตัวอย่างถัดไปแสดงการขอข้อมูลเกี่ยวกับสภาพอากาศและดัชนีชี้วัดคุณภาพอากาศจาก Airvisual Platform ซึ่งใช้วิธีสื่อสารแบบ HTTPS ตามรูปแบบ AirVisual REST API ที่ทางบริษัทกำหนดไว้ และจะได้ข้อมูลกลับมาเป็น JSON Data
การเรียกใช้ REST API ในกรณีนี้ จะต้องมี API Key โดยจะต้องสมัครสมาชิกเพื่อใช้งาน โดยเลือกแบบฟรี (Community) และระบุสถานที่หรือ Location (ในตัวอย่างนี้คือ เมืองนนทบุรี ประเทศไทย)
import sys
import network
import ujson as json
import utime as time
import urequests as requests
JSON_CONFIG_FILE = 'config.json'
AIRVISUAL_URL = 'https://api.airvisual.com/v2/city'
config = {}
try:
with open( JSON_CONFIG_FILE ) as json_file:
config = json.load(json_file)
except OSError as ex:
print('Cannot open JSON file')
sys.exit(-1)
sta_if = network.WLAN( network.STA_IF )
sta_if.active( True )
sta_if.connect( config['ssid'], config['pwd'] )
while not sta_if.isconnected():
time.sleep(1.0)
# show IP address assigned by DHCP server
#print( 'Connected:', sta_if.ifconfig()[0] )
# Fetch JSON data by using the AirVisual API
url_template = AIRVISUAL_URL + '?city={}&state={}&country={}&key={}'
url = url_template.format(
'Mueang%20Nonthaburi',
'Nonthaburi',
'Thailand',
config['airvisual_api_key'] )
resp = None
try:
resp = requests.get( url )
except OSError:
print('Cannot fetch data')
if resp and resp.status_code == 200: # OK
json_data = resp.json()
if json_data['status'] == 'success':
data = json_data['data']
location = (data['state'],data['city'],data['country'])
polution = data['current']['pollution']
us_aqi = polution['aqius']
ts = polution['ts'] # '2020-05-19T02:00:00.000Z'
weather = data['current']['weather']
tp = weather['tp'] # temperature (deg.C)
hu = weather['hu'] # humidity (%)
pr = weather['pr'] # pressure (mBar or hPa)
print('Location: {}, {}, {}'.format(*location) )
print( 'US AQI : {}'.format(us_aqi) )
print(('Temperature : {} deg.C\n' +
'Humidity : {} %RH\n' +
'Pressure : {} mBar').format(tp,hu,pr))
else:
print (resp.reason)ในการทำงานของโค้ดนี้ จะต้องมีไฟล์ config.json ซึ่งได้บันทึกเก็บไว้ใน MicroPython Flash Storage และมีข้อความแบบ JSON ตามลักษณะดังนี้
{ "ssid": "XXXXXXX", "pwd": "XXXXXXXXX",
"airvisual_api_key": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX" }
ตัวอย่างข้อความเอาต์พุต มีดังนี้
Location: Nonthaburi, Mueang Nonthaburi, Thailand
US AQI : 126
Temperature : 29 deg.C
Humidity : 55 %RH
Pressure : 1011 mBar
9. Getting Weather Data from OpenWeatherMap
OpenWeatherMap (https://openweathermap.org/) มีบริการ API ให้ดึงข้อมูลด้วยโพรโทคอล HTTP/HTTPS (ดูตัวอย่างการใช้ API และตัวอย่างข้อมูลในรูปแบบ JSON ได้จาก https://openweathermap.org/current)
ผู้ใช้จะต้องสมัครสมาชิกก่อนจึงจะสามารถใช้งานได้ฟรี (แต่มีข้อจำกัดและเงื่อนไขในการใช้งาน) เพื่อให้ได้ APPID ซึ่งเป็นข้อความแบบเลขฐานสิบหกจำนวน 32 หลัก
ในตัวอย่างนี้ เราจะดึงข้อมูลสภาพภูมิอากาศ และเลือกนำมาแสดงเฉพาะ อุณหภูมิ (Temperature) ความชื้นสัมพัทธ์ (Relative Humidity) และความกดอากาศ (Pressure) และเลือกเมืองเป็น จังหวัดนนทบุรี (Nonthaburi, TH) แล้วนำมาแสดงผลเป็นข้อความเอาต์พุตผ่านทาง REPL
import sys
import network
import ujson as json
import utime as time
import urequests as requests
# URL for openweathermap
API_URL = 'https://api.openweathermap.org/data/2.5/weather'
JSON_CONFIG_FILE = 'config.json'
config = {}
try:
with open( JSON_CONFIG_FILE ) as json_file:
config = json.load(json_file)
except OSError as ex:
print('Cannot open JSON file')
sys.exit(-1)
sta_if = network.WLAN( network.STA_IF )
sta_if.active( True )
sta_if.connect( config['ssid'], config['pwd'] )
while not sta_if.isconnected():
time.sleep(1.0)
#########################################################
def get_weather_data( APPID, city ):
results = {}
url = API_URL+'?units=metric&APPID='+APPID+'&q='+city
try:
data = requests.get(url).json()
except OSError:
print ('Cannot get data from OpenWeatherMap..')
return results
code = data.get('cod')
#print( json.dumps(data) )
if code == 200: # response OK
city = data['name']
results['p'] = data['main']['pressure'] # hPa
results['t'] = data['main']['temp'] # deg.C
results['h'] = data['main']['humidity'] # %
elif code == 401:
print('Error Code:', code)
print( data['message'] )
return results
city_name = 'Mueang Nonthaburi, TH'
weather_data = {
't': (' - Temperature :','deg.C'),
'h': (' - Humidity :','%RH'),
'p': (' - Pressure :','hPa'), }
appid = config['openweather_appid']
results = get_weather_data( appid, city_name )
print( city_name ) # show city name
for key in weather_data:
name, unit = weather_data[key]
value = results[key]
print('{} {} {}'.format(name,value,unit) )ในการทำงานของโค้ดนี้ จะต้องมีไฟล์ config.json ซึ่งได้บันทึกเก็บไว้ใน MicroPython Flash Storage และมีข้อความแบบ JSON ตามลักษณะดังนี้
{ "ssid": "XXXXXXX", "pwd": "XXXXXXXXX",
"openweather_appid": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" }
ตัวอย่างข้อความเอาต์พุต มีดังนี้
Mueang Nonthaburi, TH
- Temperature : 29.53 deg.C
- Pressure : 1011 hPa
- Humidity : 54 %RH
10. Getting Air Quality Data from Air4Thai
ตัวอย่างถัดไป สาธิตการดึงข้อมูลเกี่ยวกับคุณภาพอากาศในเขตกรุงเทพจากเว็บ Air4Thai ใช้ HTTP GET แล้วนำข้อมูลบางส่วนซึ่งเป็น JSON Data มาแสดงผลเป็นข้อความเอาต์พุตทาง REPL
ในตัวอย่างนี้ได้เลือกสถานีตรวจวัดที่มี Station ID ตรงกับ bkp95t ซึ่งเป็นเขตบางซื่อของกรุงเทพ ฯ
import sys
import network
import ujson as json
import utime as time
import urequests as requests
STATION_ID = 'bkp95t'
API_URL = 'http://air4thai.pcd.go.th/services/getNewAQI_JSON.php'
JSON_CONFIG_FILE = 'config.json'
config = {}
try:
with open( JSON_CONFIG_FILE ) as json_file:
config = json.load(json_file)
except OSError as ex:
print('Cannot open JSON file')
sys.exit(-1)
sta_if = network.WLAN( network.STA_IF )
sta_if.active( True )
sta_if.connect( config['ssid'], config['pwd'] )
while not sta_if.isconnected():
time.sleep(1.0)
#########################################################
def get_air4thai_data( url, station_id ):
try:
api_url = url + '?stationID=' + station_id
json_data = requests.get(api_url).json()
except OSError:
print ('Cannot get data from air4thai...')
return results
#print( json.dumps(data) )
station = json_data['areaEN']
print( 'Station: {}'.format(station) )
data = json_data['LastUpdate']
print( 'Last update:', data['date'], data['time'] )
print( 'PM25:', data['PM25']['value'], data['PM25']['unit'] )
print( 'AQI:', data['AQI']['aqi'] )
get_air4thai_data( API_URL, STATION_ID ) ตัวอย่างข้อความเอาต์พุต มีดังนี้
Station: Bang Sue, Bangkok
Last update: 2021-01-06 00:00
PM25: 41 µg/m³
AQI: 63
11. ส่งข้อความแจ้งเตือนผ่าน LINE Notify API
อีกตัวอย่างหนึ่งพบเห็นได้คือ การให้อุปกรณ์ประเภทไมโครคอนโทรลเลอร์สำหรับงาน IoT สามารถส่งข้อความแจ้งเตือน หรือ Chat Message ไปยังผู้ใช้ LINE App โดยใช้บริการที่เรียกว่า LINE Notify API ซึ่งมีการเริ่มใช้งานมาตั้งแต่ปีค.ศ. 2016
ถ้าจะใช้งานได้นั้น ผู้ใช้จะต้องไปตั้งค่าการใช้งานผ่านหน้าเว็บที่ https://notify-bot.line.me/en/ เพื่อให้ได้ Access Token มาใช้งาน
ทำขั้นตอน Login เข้าใช้งานด้วย email address และ password แล้วยืนยันผ่าน LINE App บน Smartphone
ไปที่เมนู MyPage
กดปุ่ม Generate Token ในส่วนที่เรียกว่า "Generate access token (For developers)"
จากนั้นจะมี Pop-up window ให้กรอกข้อมูล โดยจะต้องระบุ Token Name (ชื่อที่จะปรากฎเมื่อส่งข้อมูลไปยัง LINE) และเลือกว่า จะส่งข้อความ Chat ไปยังกลุ่ม LINE ใด ซึ่งผู้ใช้ได้สร้างหรืออยู่ในกลุ่มดังกล่าว หรือจะส่งหาตัวเองก็ได้โดยให้เลือก "1-on-1 chat with LINE notify" แล้วจึงกดปุ่ม Generate Token
เมื่อมีการจับคู่และสร้าง Token แล้ว ในหน้า MyPage ในส่วนที่เรียกว่า "Connected services" จะมีรายการการเชื่อมต่อ From (ชื่อ Token Name ตามที่ได้ตั้งไว้ ) To (ชื่อกลุ่มไลน์ตามที่ได้เลือก หรือ ชื่อบัญชีของผู้ใช้เอง)
การส่งข้อความไปยัง LINE Notify API จะใช้วิธี HTTPS POST Method โดยระบุ URL เป็น https://notify-api.line.me/api/notify และในส่วน HTTPS Header จะมีต้องการระบุ 'Authorization' ร่วมกับ 'Bearer ' ตามด้วย Access Token
ตัวอย่างโค้ดสำหรับไมโครไพธอนที่ทดลองใช้งานกับ ESP32 มีดังนี้ และเป็นการทดลองส่งข้อความ เช่น ระบุค่าอุณหภูมิและความชื้นในบ้าน (เป็นตัวอย่างสมมุติ)
import utime as time
import ujson as json
import usocket as socket
import ussl as ussl
import network
import urllib.parse
JSON_CONFIG_FILE = 'config.json'
config = {}
try:
with open( JSON_CONFIG_FILE ) as json_file:
config = json.load(json_file)
except OSError as ex:
print('Cannot open JSON file')
sys.exit(-1)
sta_if = network.WLAN( network.STA_IF )
sta_if.active( True )
sta_if.connect( config['ssid'], config['pwd'] )
while not sta_if.isconnected():
print('waiting for Wi-Fi connection')
time.sleep(1.0)
#########################################################
def https_post( url, headers={}, data='' ):
proto, _, host, path = url.split("/", 3)
assert( proto == 'https:' )
port = 443
if headers is None:
headers = {}
required_headers = {
'Host' : host,
'User-Agent' : 'MicroPython',
'Cache-Control' : 'no-cache',
'User-Agent' : 'MicroPython',
'Content-Type' : 'application/x-www-form-urlencoded' }
for key in required_headers.keys():
if key not in headers:
headers[key] = required_headers[key]
# get address info of the host machine
ai = socket.getaddrinfo( host, port, 0, socket.SOCK_STREAM )[0]
# create a secure socket
s = socket.socket( ai[0], ai[1], ai[2] )
s.connect(ai[-1])
s.settimeout(2.0)
s = ussl.wrap_socket(s, server_hostname=host)
# write header lines
s.write(b'POST /%s HTTP/1.1\r\n' % path)
for k in headers:
s.write(k)
s.write(b': ')
s.write(headers[k])
s.write(b'\r\n' )
s.write( b'Content-Length: %d\r\n' % (len(data)) )
s.write( b'\r\n' )
# write data
s.write(data)
# receive the response from the server
status_code = 0
eof_count = 0
resp_text = ''
while True:
try:
line = s.readline().decode().strip()
if line == '':
eof_count += 1
if eof_count >= 2:
break
if line.startswith( 'HTTP' ):
status_code = line.split(' ')[1]
elif line.startswith('{') and eof_count==1:
resp_text += line
except OSError:
break
s.close()
return (status_code, resp_text)
#########################################################
token = config['line_notify_token']
msg = {'message': u'ในบ้าน อุณหภูมิ 27.5 °C ความชื้น 55.7%'}
headers = {'Authorization' : 'Bearer %s' % token}
data = '{}={}'.format( 'message', urllib.parse.quote(msg['message']) )
url = 'https://notify-api.line.me/api/notify'
status_code, resp_text = https_post( url, headers, data )
if status_code == '200':
print( resp_text )ข้อมูลที่ใช้สำหรับการตั้งค่า Wi-Fi และ LINE Token Access จะถูกเก็บอยู่ในไฟล์ config.json ใน Flash File Storage ของตัวอุปกรณ์ไมโครไพธอน ตัวอย่างเช่น
{ "ssid": "my_wifi_ssid",
"pwd": "my_wifi_password",
"line_notify_token": "sSVjJ3qoX17pdw5UETGfi3EhLxaCzTxxxxxxxxxxxx" }
เมื่อส่งข้อความไปยัง LINE Notify Service จะได้ข้อความตอบกลับมาเป็น JSON String ถ้าทำงานได้ถูกต้องจะได้ข้อความ
{"status":200,"message":"ok"}
แต่ถ้าเกิดความผิดพลาด จะได้ข้อความ เช่น ในกรณีที่ไม่ได้ระบุ Authorization ในส่วน Header ของ HTTPS POST
{"status":401,"message":"Missing authorization header"}
หรือถ้าระบุ Access Token ไม่ถูกต้อง ก็จะได้ข้อความ
{"status":401,"message":"Invalid access token"}
ข้อสังเกต: เนื่องจากว่า มีการใช้ข้อความเป็นภาษาไทยด้วย ซึ่งบันทึกเป็น UTF-8 และเมื่อจะส่งออกไปโดยใช้ HTTPS POST จึงต้องมีการแปลงข้อมูลก่อน โดยใช้คำสั่ง urllib.parse.quote()
ดังนั้นจึงจะต้องติดตั้งไลบรารี urllib.parse ของ mciropython-lib โดยสามารถทำได้ง่ายใน MicroPython REPL โดยใช้คำสั่ง upip ดังนี้ (และการทำงานของ MicroPython ใน ESP32 จะต้องเคยเชื่อมต่อกับ Wi-Fi ได้สำเร็จแล้ว)
>>> import upip
>>> upip.install('urllib.parse')
และจะมีการดาวน์โหลดไฟล์ (.tar.gz) ที่เกี่ยวข้องจากเว็บ micropython.org แล้วเก็บลงใน /lib
ปัญหาที่พบ: แต่การใช้งานไลบรารี urllib.parse จะพบปัญหา วิธีแก้คือ ให้ลบไฟล์ re.py และ ffilib.py ออกไป จากนั้นกด Ctrl+D เพื่อเริ่มต้นการทำงานของ MicroPython ใหม่อีกครั้ง
เราสามารถลองใช้ไลบรารี urequests ซึ่งเป็นโมดูลแบบ built-in สำหรับไมโครไพธอน และใช้คำสั่ง urequests.post() สำหรับส่งข้อมูลแบบ POST ได้ แต่จากการที่ได้ลองใช้กับ LINE Notify API พบว่า มีความผิดพลาดเกิดขึ้นในการรับข้อมูลตอบกลับ (Response) แต่ก็ได้รับข้อความใน LINE App
import network
import sys
import ujson as json
import urllib.parse
import urequests as requests
JSON_CONFIG_FILE = 'config.json'
config = {}
try:
with open( JSON_CONFIG_FILE ) as json_file:
config = json.load(json_file)
except OSError as ex:
print('Cannot open JSON file')
sys.exit(-1)
sta_if = network.WLAN( network.STA_IF )
sta_if.active( True )
sta_if.connect( config['ssid'], config['pwd'] )
while not sta_if.isconnected():
print('waiting for Wi-Fi connection')
time.sleep(1.0)
#########################################################
token = config['line_notify_token']
url = 'https://notify-api.line.me/api/notify'
msg = {'message': u'ในบ้าน อุณหภูมิ 29.8 °C ความชื้น 69.0%' }
headers = {'Authorization' : 'Bearer %s' % token,
'Content-Type' : 'application/x-www-form-urlencoded' }
data = '{}={}'.format( 'message', urllib.parse.quote(msg['message']) )
try:
resp = requests.post( url, headers=headers, data=data )
print(resp.status_code, resp.text)
except Exception as ex:
print(ex)12. ตรวจสอบการเชื่อมต่อในเครือข่ายด้วย Ping
ตัวอย่างถัดไปสาธิตการตรวจสอบการเชื่อมต่อกับเครื่องคอมพิวเตอร์หรืออุปกรณ์ในเครือข่าย โดยใช้คำสั่ง ping ซึ่งเป็นการส่งแพคเกจตามรูปแบบ ICMP (Internet Control Message Protocol)
โค้ดตัวอย่างต่อไปนี้ ดัดแปลงมาจากโค้ด uping.py (Micro-Ping for MicroPython) ที่ได้มีการแชร์ไว้ใน gist.github.com
import network
import utime as time
import ujson as json
########################################################
def checksum(data): # CRC16
if len(data) % 2 == 1:
data += b'\x00'
chksum = 0
for pos in range(0, len(data), 2):
hi_byte = data[pos]
lo_byte = data[pos + 1]
chksum += (hi_byte << 8) + lo_byte
while chksum >= 0x10000:
chksum = (chksum & 0xffff) + (chksum >> 16)
chksum = ~chksum & 0xffff
return chksum
def ping(host, count=4, timeout=5000,
interval=10, quiet=False, size=64):
import utime
import uselect
import uctypes
import usocket
import ustruct
import urandom
# prepare packet
assert( (size >= 16), "packet size too small")
assert( (size <= 1000), "packet size too big")
pkt = b'Q'*(size)
pkt_desc = {
"type" : uctypes.UINT8 | 0,
"code" : uctypes.UINT8 | 1,
"checksum" : uctypes.UINT16 | 2,
"id" : uctypes.UINT16 | 4,
"seq" : uctypes.INT16 | 6,
"timestamp": uctypes.UINT64 | 8,
} # packet header descriptor
h = uctypes.struct( uctypes.addressof(pkt),
pkt_desc, uctypes.BIG_ENDIAN )
h.type = 8 # ICMP_ECHO_REQUEST
h.code = 0
h.checksum = 0
h.id = urandom.getrandbits(16)
h.seq = 1
# init socket
sock = usocket.socket(usocket.AF_INET, usocket.SOCK_RAW, 1)
sock.setblocking(0)
sock.settimeout(timeout/1000.0)
addr = usocket.getaddrinfo(host, 1)[0][-1][0] # ip address
sock.connect((addr, 1))
if not quiet:
print("PING %s (%s): %u data bytes" % (host, addr, len(pkt)))
seqs = list(range(1, count+1))
c = 1
t = 0
n_trans = 0
n_recv = 0
finish = False
while t < timeout:
if t==interval and c<=count:
# send packet
h.checksum = 0
h.seq = c
h.timestamp = utime.ticks_us()
h.checksum = checksum(pkt)
if sock.send(pkt) == size:
n_trans += 1
t = 0 # reset timeout
else:
seqs.remove(c)
c += 1
# recv packet
while True:
socks,_,_ = uselect.select([sock],[],[],0)
if socks:
# receive incoming packet as response
resp = socks[0].recv(1024)
# create a memoryview object
resp_mv = memoryview(resp)
h2 = uctypes.struct(uctypes.addressof(resp_mv[20:]),
pkt_desc, uctypes.BIG_ENDIAN)
# validate checksum
calc_chksum = checksum(bytes(resp_mv[24:]))
chksum = ustruct.unpack("!H", resp_mv[22:24])[0]
chksum_ok = chksum == calc_chksum
assert chksum_ok, 'Checksum failed...'
seq = h2.seq
if h2.type==0 and h2.id==h.id and (seq in seqs): # 0: ICMP_ECHO_REPLY
t_elasped = (utime.ticks_us()-h2.timestamp) / 1000
ttl = ustruct.unpack('!B', resp_mv[8:9])[0] # time-to-live
n_recv += 1
if not quiet:
args = (len(resp), addr, seq, ttl, t_elasped)
print("{} bytes from {}: icmp_seq={}, ttl={}, time={:.1f} ms".format(*args))
seqs.remove(seq)
if len(seqs) == 0:
finish = True
break
else:
break
if finish:
break
utime.sleep_ms(1)
t += 1
# close
sock.close()
ret = (n_trans, n_recv)
if not quiet:
args = (n_trans, n_recv)
print("{} packets transmitted, {} packets received".format(*args))
return (n_trans, n_recv)
########################################################
JSON_CONFIG_FILE = 'config.json'
config = {}
try:
with open( JSON_CONFIG_FILE ) as json_file:
config = json.load(json_file)
except OSError as ex:
print('Cannot open JSON file')
sys.exit(-1)
sta_if = network.WLAN( network.STA_IF )
sta_if.active( True )
sta_if.connect( config['ssid'], config['pwd'] )
while not sta_if.isconnected():
print('waiting for Wi-Fi connection')
time.sleep(1.0)
ping('8.8.8.8',count=5, timeout=5000, interval=200,
quiet=False, size=44)
########################################################ตัวอย่างข้อความเอาต์พุต
PING 8.8.8.8 (8.8.8.8): 44 data bytes
64 bytes from 8.8.8.8: icmp_seq=1, ttl=112, time=139.7 ms
64 bytes from 8.8.8.8: icmp_seq=2, ttl=112, time=64.4 ms
64 bytes from 8.8.8.8: icmp_seq=3, ttl=112, time=150.3 ms
64 bytes from 8.8.8.8: icmp_seq=4, ttl=112, time=61.7 ms
64 bytes from 8.8.8.8: icmp_seq=5, ttl=112, time=158.2 ms
5 packets transmitted, 5 packets received
กล่าวสรุป
เราได้เห็นตัวอย่างการเขียนโค้ดภาษาไมโครไพธอนสำหรับ ESP32 เพื่อเชื่อมต่อกับระบบเครือข่ายไร้สายผ่าน Wi-Fi และสามารถใช้โพรโตคอลได้ อย่างเช่น HTTP/HTTPS
Last updated
Was this helpful?