ดึงข้อมูลคุณภาพอากาศประเทศไทย

ลองมือดึงข้อมูลคุณภาพอากาศปัจจุบันในประเทศไทยด้วยโค้ด Python

โค้ดทั้งหมดในบทความนี้ใช้ Apache Software License 2.0

ปัจจุบัน มี 3 เว็บที่แสดงข้อมูลคุณภาพอากาศปัจจุบันในประเทศไทยที่ใช้ข้อมูลจากสถานีวัดอากาศที่ไม่ใช้ข้อมูลจากดาวเทียมดังนี้

• AIR4THAI - เป็นข้อมูลจากกองจัดการคุณภาพอากาศและเสียง กรมควบคุมมลพิษ ที่ให้บริการบนเว็บ air4thai.pcd.go.th

• Bangkok Air Quality - เป็นข้อมูลจากกองจัดการคุณภาพอากาศและเสียงกรุงเทพมหานคร ที่ให้บริการคุณภาพอากาศตามจุดต่าง ๆ ของกรุงเทพมหานคร บนเว็บ bangkokairquality.com

• ศูนย์เฝ้าระวังคุณภาพอากาศ วช. - เป็นข้อมูลที่ทางสำนักงานการวิจัยแห่งชาติ (วช.) รวบรวมมาจัดทำเป็นเว็บให้บริการตรวจสอบคุณภาพอากาศทั่วโลก โดยให้บริการบนเว็บ pm2_5.nrct.go.th

ทั้งสามเว็บปรับปรุงข้อมูลทุก ๆ 1 ชั่วโมง โดยมีเพียง AIR4THAI เท่านั้นที่มี API ทั้ง JSON และ XML ให้ดึงไปใช้งานได้ ผมจึงเขียนโปรแกรมเพื่อดึงข้อมูลจากทั้งหมด 3 แหล่ง โดยจะใช้วิธีการดักข้อมูลแบบง่าย ๆ ก่อนอื่น เราออกแบบว่าข้อมูลที่เราจะจัดเก็บเป็นแบบไหน โดยกำหนด

data = []

โดยเก็บข้อมูลแต่ละ index เป็น index ละสถานีตรวจคุณภาพอากาศ เป็น dict ที่มีรายละเอียดข้อมูลดังนี้

{
            'Lat' : '',
            'Lng' : '',
            'aqi' : '',
            'pm2.5' : '',
            'pm10' : '',
            'CO' : '',
            'NO2' : '',
            'O3' : '',
            'SO2' : '',
            'title' : '',
            'time' : '',
            'source' : ''
  }

โดยข้อมูลตาม key ที่เราสร้างเก็บทั้ง PM2.5 , PM 10, AQI, CO, NO2, O3, SO2 และที่ตั้งของสถานีตรวจคุณภาพอากาศ เวลา รวมถึงแหล่งที่มา โดยข้อมูลส่วนไหนขาดจะเป็น '' เริ่มแรกที่ AIR4THAI กันก่อน ของ AIR4THAI มี API แบบ JSON ให้เราสามารถดึงไปใช้งานได้ Link

เรามาเขียน Python ดึงข้อมูลคุณภาพอากาศในประเทศไทยจาก AIR4THAI โดยใช้ requests โดยแรกให้เราดึงข้อมูลมาแล้วแปลงให้เป็น json ตามโค้ดข้างล่าง แล้วแปลงให้เป็น dict

import requests, json
url3 = "http://air4thai.pcd.go.th/services/getNewAQI_JSON.php"
r3 = requests.get(url3)
temp_data = dict(r3.json())

ต่อมา เราจะต้องจัดเก็บข้อมูลให้อยู่ในรูป dict ใน list ที่เรากำหนด แต่ก่อนที่เราจะเอาข้อมูลไปใส่ ต้องตรวจสอบวันเดือนปีของข้อมูลก่อน เนื่องจากมีบางเซ็นเซอร์ไม่ได้อัพเดตข้อมูลส่วนนี้ เราจึงสร้างฟังก์ชันตรวจสอบวันเดือนปีก่อน แบบง่าย ๆ

from datetime import datetime
import pytz
tz = pytz.timezone('Asia/Bangkok')

d = datetime.now(tz).strftime('%Y-%m-%d')
def check_datenow(t):
  return if d in t

มาลูปจัดเก็บลง data

for i in temp_data['stations']:
  d_temp = {}
  d_temp['Lat'] = i['lat']
  d_temp['Lng'] = i['long']
  l_temp=i['LastUpdate']
  d_temp['aqi'] = l_temp['AQI']['aqi']
  if 'PM25'  in list(l_temp.keys()):
    pm25=l_temp['PM25']['value']
    if pm25 != 'n/a' and pm25!='' and pm25!='-' and pm25!='N/A':
      d_temp['pm2.5'] = pm25
    else:
      d_temp['pm2.5'] = ''
      #continue
  else:
    d_temp['pm2.5'] = ''
    #continue
  if 'PM10'  in list(l_temp.keys()):
    pm10=l_temp['PM10']['value']
    if pm10 != 'n/a' and pm10!='' and pm10!='-' and pm10!='N/A':
      d_temp['pm10'] = pm10
  if 'CO'  in list(l_temp.keys()):
    CO=l_temp['CO']['value']
    if CO != 'n/a' and CO!='' and CO!='-' and CO!='N/A':
      d_temp['CO'] = CO
  if 'NO2'  in list(l_temp.keys()):
    NO2=l_temp['NO2']['value']
    if NO2 != 'n/a' and NO2!='' and NO2!='-' and NO2!='N/A':
      d_temp['NO2'] = NO2
  if 'O3'  in list(l_temp.keys()):
    O3=l_temp['O3']['value']
    if O3 != 'n/a' and O3!='' and O3!='-' and O3!='N/A':
      d_temp['O3'] = O3
  if 'SO2'  in list(l_temp.keys()):
    SO2=l_temp['SO2']['value']
    if SO2!= 'n/a' and SO2!='' and SO2!='-' and SO2!='N/A':
      d_temp['SO2'] = SO2
  d_temp['title'] = i['areaTH']
  d_temp['time'] = l_temp['date']+" "+l_temp['time']
  d_temp['source'] = "air4thai"
  if check_datenow(d_temp['time']):
    data.append(d_temp)

ต่อมา เรามาจัดการดึงข้อมูลคุณภาพอากาศในเขตกรุงเทพมหาครโดยดึงผ่าน Bangkok Air Quality กัน

Bangkok Air Quality มีข้อดีตรงที่มี XML ที่สามารถดึงมาใช้งานได้ โดยดึงจาก https://bangkokairquality.com/bma/marker.php เป็น XML หลาย ๆ คน คงใช้ไลบารี XML ของ Python ในการจัดการ XML แต่ผมจะใช้วิธีที่ง่ายกว่านั้น โดยใช้ xmltodict สามารถติดตั้ง xmltodict ได้ด้วยคำสั่ง xmltodict ก่อนดึงข้อมูล เนื่องจาก Bangkok Air Quality ใช้รูปแบบวันเดือนปีที่ไม่เหมือนกับ AIR4THAI จึงต้องสร้างฟังก์ชันใหม่ขึ้นมา

d2 = datetime.now(tz).strftime('%d-%m-%Y')
def check_datenow2(t):
  return if d2 in t
  
# getdata
import xmltodict
data_bangkok=dict(json.loads(json.dumps(xmltodict.parse(r2.text),ensure_ascii=False)))
for i in data_bangkok['markers']['marker']:
  d_temp = {}
  d_temp['Lat'] = i['@lat']
  d_temp['Lng'] = i['@lng']
  d_temp['aqi'] = ''
  if '@pm25' in list(i.keys()):
    if i['@pm25'] != 'n/a' and i['@pm25']!='' and i['@pm25']!='-':
      d_temp['pm2.5'] = i['@pm25']
    else:
      d_temp['pm2.5'] = ''#continue
  else:
    d_temp['pm2.5'] =''#continue
  if '@pm10' in list(i.keys()):
    d_temp['pm10'] = i['@pm10']
  d_temp['title'] = i['@district_th'].strip() + " กรุงเทพฯ"
  d_temp['CO'] = ''
  d_temp['NO2'] = ''
  d_temp['O3'] = ''
  d_temp['SO2'] = ''
  d_temp['time'] = i['@date_time']
  d_temp['source'] = "bangkokairquality"
  if check_datenow2(d_temp['time']):
    d_temp['time'] = d_temp['time'].replace(d2,d) # เปลี่ยนวันเดือนปี ให้เหมือนกัน
    data.append(d_temp)

สุดท้าย เรามาดึงข้อมูลคุณภาพอากาศจากศูนย์เฝ้าระวังคุณภาพอากาศ วช. ด้วย Python กัน เนื่องจากข้อมูลเป็นแบบหน้าเว็บ ไม่ได้ใช้ JSON หรือ XML เราจึงเขียนตัวสกัดเอาข้อมูลจาก HTML Code กัน หลาย ๆ คนคงคิดว่าคงจะใช้ regex ใช่ไหม? คำตอบของบทความนี้ คือ ไม่! เราจะเขียนแบบง่าย ๆ ที่สุดคือ ใช้ split ไปจนได้ข้อมูลตามที่ต้องการมาได้ครบถ้วน ข้อมูลใน pm2_5.nrct.go.th/map เราสามารถดักเป็นรูปแบบได้ โดยตามรูปแบบ

var locationXX = ...  # ไปเรื่อย ๆ แล้วค่อยดักหลัง
google.maps.event.addListener(
# เป็นต้นไป

url = "http://pm2_5.nrct.go.th/map"
r = requests.get(url)
html = r.text.strip()

do2 = html.split(" var location")[1:]
do3 = []
for i in do2:
  do3.append(i.split("google.maps.event.addListener(")[0])

for i in do3:
  d_temp = {}
  temp1= i.split("new google.maps.LatLng(")
  LatLng = temp1[1].split(");")[0]
  d_temp['Lat'] = LatLng.split(",")[0]
  d_temp['Lng'] = LatLng.split(",")[1]
  temp1 = temp1[1]
  if "var aqi = '" in temp1:
    d_temp['aqi'] = temp1.split("var aqi = '")[1].split("'")[0]
  #if "var aqival = '" in temp1:
  #  d_temp['aqival'] = temp1.split("var aqival = '")[1].split("'")[0]
  if 'var pm = "' in temp1:
    d_temp['pm2.5'] = temp1.split('var pm = "')[1].split('"')[0]
  #if 'var pmint = "' in temp1:
  #  d_temp['pmint'] = temp1.split('var pmint = "')[1].split('"')[0]
  d_temp['pm10'] = ''
  d_temp['CO'] = ''
  d_temp['NO2'] = ''
  d_temp['O3'] = ''
  d_temp['SO2'] = ''
  d_temp['title'] = temp1.split("data: '")[1].split("'")[0].strip()
  if "Air4Thai" in d_temp['title']:
    continue
  d_temp['time'] = temp1.split("time: '")[1].split("'")[0]
  d_temp['source'] = "nrct"
  if check_datenow(d_temp['time']):
    data.append(d_temp)

เรียบร้อย เพียงเท่านี้เราก็ได้ข้อมูลข้อมูลคุณภาพอากาศในประเทศไทยด้วย Python ไว้ใช้ประมวลผลต่อไปแล้ว

เรามาลองโชว์ข้อมูล PM2.5 บนแผนที่ตามจุดสถานีตรวจคุณภาพอากาศกัน โดยใช้ folium

import folium # ติดตั้งได้ด้วย pip install folium
map_osm = folium.Map(location=[15.0000, 100.0000], zoom_start=5) # [ละติจูด,ลองติดจูด] และการซูมลำดับ 5
tooltip = 'Click me!'

for i in data:
  if i['Lat']=='' and i['Lng']=='':
    continue
  #print([(i['Lat'].strip()), (i['Lng'].strip())])
  folium.Marker([float(i['Lat'].strip()), float(i['Lng'].strip())], popup=i['pm2.5']+" µg/m³", tooltip=tooltip).add_to(map_osm)

map_osm.save('thailandmap.html') # บันทึกเป็นไฟล์ html

ลองมาเปิดไฟล์ thailandmap.html

แผนที่โดย OpenStreetMap

ดึงข้อมูลคุณภาพอากาศปัจจุบันในประเทศไทยด้วย PyThaiAIR

ก่อนที่เราจะไปไกลกันกว่านี้ เชื่อว่าผู้อ่านคงอยากดึงข้อมูลด้วยโมดูลง่าย ๆ ไม่ต้องเขียนโค้ดดึงเอง ผมจึงได้ทำโมดูลใหม่ที่ชื่อ PyThaiAIR ขึ้นมา เพื่อดึงข้อมูลคุณภาพอากาศในประเทศไทยปัจจุบัน สามารถติดตั้งได้ด้วยคำสั่ง

pip install pythaiair

การใช้งาน ให้เรียกใช้

from pythaiair import Air
air = Air()

โดยมีคำสั่งดังนี้

• Air().get_data() - รับข้อมูลคุณภาพอากาศจากสถานีตรวจอากาศออกมาเป็น list โดยข้อมูลเป็น dict ตามโค้ดข้างบน

• Air().get_data_provinces() - รับข้อมูลคุณภาพอากาศจากสถานีตรวจอากาศออกมาเป็น dict แยกตามชื่อจังหวัดในประเทศไทย

• Air().update_data() - ใช้ปรับปรุงข้อมูลให้ดึงข้อมูลมาใหม่ แนะนำให้ทำทุก 1 ชั่วโมง

สร้างกราฟแท่งแสดงค่า PM2.5 ที่สูงที่สุด 5 อันดับแรกของกรุงเทพมหานคร

พอเราติดตั้งเสร็จแล้ว ต่อไปเรามาลองสร้างกราฟแท่งแสดงค่า PM2.5 ที่สูงที่สุด 5 อันดับแรกของกรุงเทพมหานครกัน

from pythaiair import Air
air = Air()
กรุงเทพpm25_x = [i['title'] for i in air.get_data_provinces()['กรุงเทพฯ']]
กรุงเทพpm25_y = [i['pm2.5'] for i in air.get_data_provinces()['กรุงเทพฯ']]
กรุงเทพpm25 = sorted(list(zip(กรุงเทพpm25_x, กรุงเทพpm25_y)), key=lambda x: x[1],reverse=True) # เรียงจากมากที่สุดไปน้อยที่สุด
pm25_x = [i[0] for i in กรุงเทพpm25]
pm25_y = [i[1] for i in กรุงเทพpm25]

from nvd3 import discreteBarChart # ติดตั้งด้วยคำสั่ง pip install python-nvd3
chart = discreteBarChart(name='discreteBarChart', height=400, width=1280)

xdata = pm25_x[:5]
ydata = pm25_y[:5]

chart.add_serie(y=ydata, x=xdata)
chart.buildhtml()
with open('bkk.html','w') as f:
 f.write('chart.htmlcontent')

ผลลัพธ์ เมื่อเปิดไฟล์ bkk.html

สามารถลองเล่น Colab บทความนี้ได้ที่ https://colab.research.google.com/drive/10C-MPTQTQYljKCuoyf2O73CYI7C29b7Y และเข้าไปร่วมพัฒนา PyThaiAIR ได้ที่ https://github.com/wannaphong/pythaiair

Reference : python3.wannaphong.com