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 ได้แก่

สำหรับ 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 มาใช้งาน

  1. ทำขั้นตอน Login เข้าใช้งานด้วย email address และ password แล้วยืนยันผ่าน LINE App บน Smartphone

  2. ไปที่เมนู MyPage

  3. กดปุ่ม Generate Token ในส่วนที่เรียกว่า "Generate access token (For developers)"

  4. จากนั้นจะมี Pop-up window ให้กรอกข้อมูล โดยจะต้องระบุ Token Name (ชื่อที่จะปรากฎเมื่อส่งข้อมูลไปยัง LINE) และเลือกว่า จะส่งข้อความ Chat ไปยังกลุ่ม LINE ใด ซึ่งผู้ใช้ได้สร้างหรืออยู่ในกลุ่มดังกล่าว หรือจะส่งหาตัวเองก็ได้โดยให้เลือก "1-on-1 chat with LINE notify" แล้วจึงกดปุ่ม Generate Token

  5. เมื่อมีการจับคู่และสร้าง 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

เผยแพร่ภายใต้ลิขสิทธิ์ Attribution-ShareAlike 4.0 International (CC BY-SA 4.0)

PreviousESP32 with Dual-Channel DAC OutputNextArduino C/C++ versus MicroPython

Last updated