Python 空气质量指数计算 [ 日升于东 ]

计算方法

分指数

IAQI_p = (IAQI_Hi - IAQI_Lo)/(BP_Hi - BP_Lo)(C_p - BP_Lo) + IAQI_Lo

IAQI_p：污染物项目P的空气质量分指数

C_p：污染物项目P的质量浓度值

BP_Hi：C_p相近的污染物浓度限值的高位值

BP_Lo：C_p相近的污染物浓度限值的低位值

IAQI_Hi：BP_Hi对应的空气质量分指数

IAQI_Lo：BP_Lo对应的空气质量分指数

指数

AQI = max｛IAQI₁，IAQI₂，···，IAQI_n｝

IAQI：单项污染物的空气质量指数
AQI：定量描述空气质量状况的无量纲指数

本质：线性缩放

AQI

#线性缩放
def cal_linear(C_p, BP_Lo, BP_Hi, IAQI_Lo, IAQI_Hi):
    return (IAQI_Hi-IAQI_Lo)/(BP_Hi-BP_Lo)*(C_p-BP_Lo) + IAQI_Lo
#分指数计算
def cal_iaqi(val, type):
    dict = {
        "pm":["0-0","35-50","75-100","115-150","150-200","250-300","350-400","500-500"],
        "co":["0-0","5-50","10-100","35-150","60-200","90-300","120-400","150-500"]
        }
    C_p = val
    for i in dict[type]:
        currIndex = dict[type].index(i)
        try:
            if dict[type][currIndex+1] != '':
                _lo = float(i.split('-')[0])
                _hi = float(dict[type][currIndex+1].split('-')[0])
                if _lo <= C_p < _hi:
                    BP_Lo = _lo
                    BP_Hi = _hi

                    IAQI_Lo = float(i.split('-')[1])
                    IAQI_Hi = float(dict[type][currIndex+1].split('-')[1])
                    break
            else:
                print("IAQI 超过 500！")
        except TypeError:
            print("数值类型错误！")
        except:
            print("分指数计算错误！")

    return cal_linear(C_p, BP_Lo, BP_Hi, IAQI_Lo, IAQI_Hi)
#指数计算
def cal_AQI(params):
    
    pm_iaqi = cal_iaqi(params[0], "pm")
    co_iaqi = cal_iaqi(params[1], "co")

    return max(pm_iaqi,co_iaqi)

def main():
    input("请输入以下信息，用空格分割。");
    str = input("(1)PM2.5 (2)CO:");
    try:
        pmVal = float(str.split(' ')[0])
        coVal = float(str.split(' ')[1])

        params = []
        params.append(pmVal)
        params.append(coVal)

        aqiVal = cal_AQI(params)

        print("空气质量指数为：{}".format(aqiVal))
    except IndexError:
        print("输入参数值数量不够！")
    except:
        print("主程序报错！")

if __name__ == "__main__":
    main()

JSON 数据文件

轻量级数据交换格式
对复杂数据进行表达和存储，易于阅读和理解
示例：
json = {
    'a':[1,2,3,4,5],
    'b':"123"
}

JSON 库

import json
编码（encoding） 将 python 数据转化成 JSON 数据
解码（decoding） 将 JSON 数据解析到 python 数据类型

dumps()  //编码
loads()  //解码
dump()   //输出到文件
load()   //从文件读入

# 从文件读入
f = open(filepath, mode = "r", encoding = "utf-8")
_list = json.load(f)
# -> 输出结果为字典类型

_list.sort(key = lambda city: city['api'])
# 输出到文件
json.dump(_list[:5], f, ensure_ascii = False)

CSV

通用，简单文本文件
适用商业、科学领域
规则：
    以行为单位，每行表示一条数据，以英文逗号分割每列数据，列名通常放置在第一行

import csv
csv.writerow(list)  //将列表中的元素写入文件的一行中
csv.reader()        //将每行记录作为列表返回
with open(filepath) as somefile:
    for line in somefile:
        print(line)
*使用 with 语句，不管是否发生异常，最后都会关闭文件，不用写 close()*

示例：
api, area, pm2_5
47, 北京, 32
63, 上海, 37

#写入CSV
lines =[]
lines.append(list(_list[0].keys()))
for item in _list:
    lines.append(list(item.values()))

f = open(csvfilepath, "w", encoding = "utf-8", newline = "")
writer = csv.writer(f)

for line in lines
    csv.writerow(line)

f.close()

os 模块

os.remove()          //删除文件
os.makedirs()        //创建多层目录
os.rmdir()           //删除单级目录
os.remame()          //重命名文件
os.path.isfile()     //判断是否为文件
os.path.isdir()      //判断是否为目录
os.path.join()       //连接目录
# -> os.path.join(path1,path2)
# -> path1/path2
os.path.splitext()   //分割文件名和扩展名
# -> os.path.splitext(tmp.txt)
# -> tmp，.txt

import json
import csv
import os

class os_switch:
    #初始化
    def __init__(self, params):
        self.funs = {
            "json": self.jsonFun,
            "csv": self.csvFun,
            }
        self.params = params
        self.funs.get(self.params["moden"], self.other)()

    def jsonFun(self):
        try:
            with open(self.params["filepath"], mode="r", encoding="utf-8") as f:
                # load 方法
                _list = json.load(f, strict=False)
                # loads 方法
                # data = f.read();
                # _list = json.loads(data)
                print(_list)
        except json.decoder.JSONDecodeError:
            print("jsonFun 运行 JSONDecodeError 报错！")
        except:
            print("jsonFun 运行报错！")

    def csvFun(self):
        with open(self.params["filepath"], mode="r", encoding="utf-8", newline="") as f:
            reader = csv.reader(f)
            for row in reader:
                print('，'.join(row))

    def other(self):
        print("暂不支持当前格式的文件！")

def osFun():
    filepath = input("请输入文件路径：");
    file_name, file_ext = os.path.splitext(filepath);

    try:
        params = {
            'moden': file_ext[1:],
            'filepath': filepath
            }
        os_switch(params)
    except:
        print("osFun 运行错误！")

def main():
    osFun()

if __name__ == "__main__":
    main()

自定义 switch

类模式

class os_switch:
    def __init__(self, params):
        self.funs = {
            "json": self.jsonFun,
            "csv": self.csvFun,
            }
        self.params = params
        self.funs.get(self.params["moden"], self.other)()
    def json():
        pass
    def csv():
        pass
    def other():
        pass

字典模式

switch = {
    'a' = a,
    'b' = b
}
def a():
    pass
def b():
    pass

requests 模块

简洁、简单的处理 HTTP 请求的工具
支持丰富的链接访问功能，包括 URL 获取，HTTP 回话，Cookie 记录等
get() 
post()
status_code    // HTTP 请求的返回状态，200表示成功，400表示失败
text           // HTTP 相应内容的字符串形式，即 URL 对应的页面内容

def getHtmlText(url):
    r = requests.get(url, timeout = 30);
    print(r.status_code)
    r.encoding='utf-8'
    return r.text

def main():
    city = input("请输入城市拼音：")
    url = 'http://www.86kongqi.com/city/' + city + '.html'
    text = getHtmlText(url)

    print(text)

if __name__ == "__main__":
    main()

中文乱码问题

参考解决方案

#方法1
r = requests.get(url)
html=r.content
html_doc=str(html,'utf-8')

#方法2
r=requests.get(url)
r.encoding='utf-8'

beautifulsoup4

解析 HTML 或者 XML
pip install beautifulsoup4
import bs4

find       //查找第一个满足条件的节点
find_all   //查找所有满足条件的节点

bs = BeautifulSoup(
    url,
    html_parser   //指定解析器
    enoding       //编码格式，确保和网页编码格式一致
)

# <a href="a.html" class="a_link">next page</a>
# 可按节点类型，属性或内容访问

bs.find_all('a');
bs.find_all('a', href = 'a.html')
bs.find_all('a', href = 'a.html', string = 'next page')

bs.find_all('a', class_ = 'a_link')
bs.find_all('a', {'class':'a_link'})

import bs4

r = requests.get(url, timeout = 30);
r.encoding='utf-8'

soup = bs4.BeautifulSoup(r.text, 'lxml')
return soup.find_all('div', {'class':'weilai'})[0].find('table')

批量抓取

def getAllCity(url):
    r = requests.get(url, timeout = 30);
    r.encoding='utf-8'
    soup = bs4.BeautifulSoup(r.text, 'lxml')

    cities = soup.find_all('dl', {'id':'clist'})[0].find('dd')
    return cities.find_all('a')

def getHtmlText(url):
    r = requests.get(url, timeout = 30);
    r.encoding='utf-8'

    soup = bs4.BeautifulSoup(r.text, 'lxml')
    return soup.find_all('div', {'class':'weilai'})[0].find('table')

def main():
    
    url = 'http://www.86kongqi.com'

    cities = getAllCity(url)
    for city in cities:
        ciry_url = url + "/" + city['href']
        text = getHtmlText(ciry_url)
        print(ciry_url)

if __name__ == "__main__":
    main()

pandas

分析结构化数据的工具集
基础是 Numpy 提供了高性能的矩阵运算
应用：数据挖掘，数据分析，数据清洗功能

数据分析

ser_obj = pd.Series(range(10))  // 一维数组对象，索引自动创建，索引在左，数据在右
ser_obj.index                   // 获取索引
ser_obj.values                  // 获取值
ser_obj.head(n)                 // 预览数据，n 前几行
ser_obj.tail(n)                 // 预览数据，n 后几行
ser_obj[index]

Data.Frame
通过 ndarray，dict 创建
data['列名']                           // 获取单列数据
data[['列名1'，'列名2']]                // 获取多列数据
sort_values(by = ['列名']).head(10)     // 获取前 10 名，降序 ascending = False

pd.to_csv('topTen.csv'，index = False)  // 保存为 CSV

数据清洗

dropna()    // 丢失缺失数据
fillna()    // 填充缺失数据
df[过滤条件]

数据可视化

plot(kind，x，y，title，figsize)
plt.savefig()

官方地址