手把手教你用Python实现AQI计算器附完整代码下载最近几年身边关注空气质量的朋友越来越多了。早上出门前看一眼手机上的空气质量指数几乎成了很多人的习惯。但你是否好奇过屏幕上那个简单的数字和“良”、“轻度污染”的标签背后究竟是怎么算出来的市面上的App和网站提供了便捷的查询但作为一个对数据和编程有点兴趣的人我总觉得直接“拿来就用”少了点意思。如果能亲手把原始监测数据“加工”成那个熟悉的AQI值整个过程会清晰得多不仅能更深刻地理解空气质量评价体系还能根据自己的需求定制展示方式。今天我们就抛开现成的工具从最底层的技术规范出发用Python一步步搭建一个属于自己的AQI计算器。这个过程会涉及到数据处理、逻辑判断和结果可视化非常适合想通过实战项目来巩固Python技能同时又对环境数据感兴趣的朋友。1. 理解AQI计算的核心逻辑在动手写代码之前我们必须先把AQI的计算规则吃透。AQI即环境空气质量指数它本身并不是直接测量出来的而是一个通过一套既定公式“计算”出来的综合指数。它的核心思想是“木桶效应”——空气质量的好坏由当时最差的那项污染物决定。1.1 污染物与分指数IAQI我们常说的AQI主要依据六项基本污染物细颗粒物PM₂.₅、可吸入颗粒物PM₁₀、臭氧O₃、一氧化碳CO、二氧化氮NO₂和二氧化硫SO₂。计算的第一步是为每一项污染物计算其单独的“空气质量分指数”即IAQI。IAQI的计算并非简单的线性比例而是采用了分段线性插值法。国家标准为每种污染物在不同浓度区间对应不同的IAQI区间定义了限值。例如PM₂.₅的24小时平均浓度限值单位μg/m³与IAQI的对应关系如下表所示IAQI区间PM₂.₅浓度低值 (BP_Lo)PM₂.₅浓度高值 (BP_Hi)0-5003551-1003575101-15075115151-200115150201-300150250301-400250350401-500350500注意不同污染物的浓度区间划分各不相同且O₃有1小时和8小时平均两个标准CO有1小时和24小时平均两个标准在实际计算中需要根据数据的时间尺度正确选择。计算公式非常直观IAQI_p [(IAQI_Hi - IAQI_Lo) / (BP_Hi - BP_Lo)] * (C_p - BP_Lo) IAQI_Lo其中C_p是污染物p的实测浓度值BP_Lo和BP_Hi是C_p所在浓度区间的下限和上限IAQI_Lo和IAQI_Hi是对应的IAQI下限和上限。1.2 从IAQI到AQI与首要污染物计算出所有六项污染物的IAQI后AQI的确定就很简单了AQI max(IAQI_1, IAQI_2, ..., IAQI_6)即AQI的值等于所有污染物分指数中的最大值。而那个“最大值”对应的污染物就被确定为首要污染物。当AQI大于50时首要污染物需要被报告。有时会出现多个IAQI值相同且均为最大值的情况则它们都被列为首要污染物。1.3 数据准备与项目结构理解了算法我们来看看实现它需要什么。首先你需要一份包含六项污染物浓度的数据。这可以是一个CSV文件、一个Excel表格或者从API实时获取的数据流。为了演示我们将使用一个静态的CSV文件作为输入。我们的项目将包含以下几个核心部分数据加载模块负责读取原始浓度数据。核心计算引擎实现分段线性插值算法计算每个污染物的IAQI。AQI判定模块找出最大IAQI确定AQI级别、类别和首要污染物。结果可视化模块将计算结果以图表形式直观展示。主程序串联整个流程。我会在文章最后提供完整的、可运行的代码包。现在让我们进入具体的编码环节。2. 搭建Python计算环境与数据加载工欲善其事必先利其器。我们选择Python主要是因为它在数据分析和科学计算领域的强大生态。这个项目不需要特别复杂的库。2.1 创建虚拟环境与安装依赖我强烈建议为这个项目创建一个独立的虚拟环境避免污染全局的Python环境。# 创建并激活虚拟环境 (以conda为例) conda create -n aqi_calculator python3.9 conda activate aqi_calculator # 安装必要的库 pip install pandas numpy matplotlib这里我们用到三个核心库pandas用于轻松处理和操作表格型数据我们的浓度数据表。numpy提供高效的数组运算虽然本项目计算不复杂但它是科学计算的基础。matplotlib用于绘制结果图表直观展示AQI和各项IAQI。2.2 定义污染物浓度限值表这是整个计算器的“灵魂”所在。我们需要将国家标准中的限值表严谨地翻译成Python数据结构。我将它定义为一个嵌套字典这样后续查询非常方便。# 定义AQI技术规定HJ 633-2012中的浓度限值表 # 结构污染物 - IAQI区间 - [浓度低值 浓度高值] # 单位CO为mg/m³其余为μg/m³。O3-8h为8小时滑动平均O3-1h为1小时平均。 AQI_BREAKPOINTS { PM2.5_24h: { (0, 50): (0, 35), (51, 100): (35, 75), (101, 150): (75, 115), (151, 200): (115, 150), (201, 300): (150, 250), (301, 400): (250, 350), (401, 500): (350, 500) }, PM10_24h: { (0, 50): (0, 50), (51, 100): (50, 150), (101, 150): (150, 250), (151, 200): (250, 350), (201, 300): (350, 420), (301, 400): (420, 500), (401, 500): (500, 600) }, O3_8h: { (0, 50): (0, 100), (51, 100): (100, 160), (101, 150): (160, 215), (151, 200): (215, 265), (201, 300): (265, 800), # 301-400和401-500区间在标准中未定义通常按上一区间外延处理此处简化 }, CO_24h: { # 单位 mg/m³ (0, 50): (0, 2), (51, 100): (2, 4), (101, 150): (4, 14), (151, 200): (14, 24), (201, 300): (24, 36), (301, 400): (36, 48), (401, 500): (48, 60) }, NO2_24h: { (0, 50): (0, 40), (51, 100): (40, 80), (101, 150): (80, 180), (151, 200): (180, 280), (201, 300): (280, 565), (301, 400): (565, 750), (401, 500): (750, 940) }, SO2_24h: { (0, 50): (0, 50), (51, 100): (50, 150), (101, 150): (150, 475), (151, 200): (475, 800), (201, 300): (800, 1600), (301, 400): (1600, 2100), (401, 500): (2100, 2620) } }2.3 加载实测浓度数据假设我们有一个sample_data.csv文件内容如下date,PM2.5,PM10,O3,CO,NO2,SO2 2023-10-27, 68, 120, 110, 1.2, 45, 30 2023-10-28, 105, 180, 140, 1.8, 60, 25 2023-10-29, 45, 95, 95, 0.9, 30, 15我们用pandas来加载它import pandas as pd def load_concentration_data(file_path): 加载污染物浓度数据CSV文件。 假设文件包含日期和六项污染物浓度列。 try: df pd.read_csv(file_path) # 确保列名与我们的计算逻辑匹配这里可以进行一些清洗和重命名 # 例如确保列名为 PM2.5, PM10, O3, CO, NO2, SO2 required_cols [PM2.5, PM10, O3, CO, NO2, SO2] for col in required_cols: if col not in df.columns: # 尝试一些常见的列名变体 if col PM2.5 and PM2_5 in df.columns: df.rename(columns{PM2_5: PM2.5}, inplaceTrue) elif col PM10 and PM10 not in df.columns: # 其他可能的变体处理... pass print(f数据加载成功共 {len(df)} 条记录。) print(df.head()) return df except FileNotFoundError: print(f错误文件 {file_path} 未找到。) return None except Exception as e: print(f加载数据时发生错误{e}) return None3. 实现核心计算引擎这是最具技术挑战也最有趣的部分。我们需要一个函数能够根据输入的污染物类型和浓度值准确地找到对应的浓度区间并应用分段线性插值公式。3.1 计算单项污染物的IAQIdef calculate_iaqi(pollutant, concentration): 计算单一污染物的IAQI。 参数: pollutant (str): 污染物名称如 PM2.5_24h concentration (float): 污染物浓度实测值 返回: float: 计算得到的IAQI值。如果浓度超过最高限值返回None或进行特殊处理。 if pollutant not in AQI_BREAKPOINTS: raise ValueError(f不支持的污染物类型: {pollutant}) # 获取该污染物的所有限值区间 breakpoints AQI_BREAKPOINTS[pollutant] # 遍历所有区间找到浓度值所在的区间 for (iaqi_lo, iaqi_hi), (bp_lo, bp_hi) in breakpoints.items(): if bp_lo concentration bp_hi: # 应用分段线性插值公式 iaqi_value ((iaqi_hi - iaqi_lo) / (bp_hi - bp_lo)) * (concentration - bp_lo) iaqi_lo # 结果四舍五入取整AQI标准要求整数 return round(iaqi_value) # 如果浓度值超过了定义的最高区间例如PM2.5 500 # 处理方式可以返回最大值500或按最后一个区间外延计算需谨慎 # 这里我们返回None并给出警告 print(f警告{pollutant} 浓度值 {concentration} 超过标准定义范围。) return None3.2 批量计算与AQI判定有了单项计算函数我们就可以处理一整行一天的数据了。def calculate_aqi_for_row(row): 计算单行数据单日/单次监测的AQI及相关信息。 参数: row (pd.Series): 包含六项污染物浓度的一行数据 返回: dict: 包含AQI值、首要污染物、各污染物IAQI等信息的字典 # 映射关系数据框列名 - 计算用的污染物键名 pollutant_map { PM2.5: PM2.5_24h, PM10: PM10_24h, O3: O3_8h, # 注意这里假设是O3-8h数据实际需根据数据时间尺度调整 CO: CO_24h, NO2: NO2_24h, SO2: SO2_24h } iaqi_results {} for data_col, pollutant_key in pollutant_map.items(): conc row[data_col] # 检查浓度值是否有效非空且为数字 if pd.isna(conc): iaqi None else: iaqi calculate_iaqi(pollutant_key, conc) iaqi_results[data_col] iaqi # 找出有效的IAQI中的最大值即为AQI valid_iaqis {k: v for k, v in iaqi_results.items() if v is not None} if not valid_iaqis: return {AQI: None, Primary_Pollutant: None, IAQI: iaqi_results} aqi_value max(valid_iaqis.values()) # 确定首要污染物可能有多个 primary_pollutants [k for k, v in valid_iaqis.items() if v aqi_value] return { AQI: aqi_value, Primary_Pollutant: , .join(primary_pollutants) if primary_pollutants else None, IAQI: iaqi_results }3.3 为整个数据集进行计算现在我们将上述函数应用到加载的整个DataFrame上。def calculate_aqi_for_dataframe(df): 为整个DataFrame计算AQI。 参数: df (pd.DataFrame): 包含污染物浓度数据的DataFrame 返回: pd.DataFrame: 增加了AQI、首要污染物等列的DataFrame results df.apply(calculate_aqi_for_row, axis1, result_typeexpand) # 将结果合并到原数据框 df_result df.copy() df_result[AQI] results[AQI] df_result[Primary_Pollutant] results[Primary_Pollutant] # 可以选择性地将各污染物的IAQI也展开为单独的列 iaqi_df pd.json_normalize(results[IAQI]) iaqi_df.columns [fIAQI_{col} for col in iaqi_df.columns] df_result pd.concat([df_result, iaqi_df], axis1) return df_result4. 结果可视化与报告生成数字结果虽然精确但图表更能直观地揭示趋势和问题。我们使用matplotlib来创建两个核心图表。4.1 绘制AQI时间序列图这张图可以清晰地展示一段时间内空气质量的变化趋势。import matplotlib.pyplot as plt import matplotlib.dates as mdates def plot_aqi_trend(df_result, date_coldate): 绘制AQI随时间变化的趋势图。 fig, ax plt.subplots(figsize(12, 6)) # 确保日期列是datetime类型 dates pd.to_datetime(df_result[date_col]) aqi_values df_result[AQI] # 绘制折线 ax.plot(dates, aqi_values, markero, linestyle-, linewidth2, markersize6, colorsteelblue, labelAQI) # 添加AQI等级分界线例如100 150 200和背景色 ax.axhspan(0, 50, facecolorgreen, alpha0.1, label优) ax.axhspan(51, 100, facecoloryellow, alpha0.1, label良) ax.axhspan(101, 150, facecolororange, alpha0.1, label轻度污染) ax.axhspan(151, 200, facecolorred, alpha0.1, label中度污染) ax.axhspan(201, 300, facecolorpurple, alpha0.1, label重度污染) ax.axhspan(301, 500, facecolormaroon, alpha0.1, label严重污染) ax.set_xlabel(日期, fontsize12) ax.set_ylabel(AQI, fontsize12) ax.set_title(空气质量指数AQI变化趋势, fontsize14, fontweightbold) ax.legend(locupper left) ax.grid(True, linestyle--, alpha0.6) # 格式化x轴日期显示 ax.xaxis.set_major_formatter(mdates.DateFormatter(%m-%d)) ax.xaxis.set_major_locator(mdates.DayLocator()) fig.autofmt_xdate() # 旋转日期标签 plt.tight_layout() plt.savefig(aqi_trend.png, dpi300) plt.show()4.2 绘制污染物IAQI贡献雷达图这张图可以一次性展示某一天所有污染物对空气质量的“贡献”程度非常直观。import numpy as np def plot_iaqi_radar(df_result, target_date, date_coldate): 绘制指定日期各污染物IAQI的雷达图。 # 筛选目标日期的数据 row df_result[df_result[date_col] target_date].iloc[0] if row.empty: print(f未找到日期 {target_date} 的数据。) return # 准备雷达图数据 pollutants [PM2.5, PM10, O3, CO, NO2, SO2] iaqi_values [row[fIAQI_{p}] for p in pollutants] # 处理可能的None值 iaqi_values [v if v is not None else 0 for v in iaqi_values] # 雷达图需要闭合所以重复第一个值 angles np.linspace(0, 2 * np.pi, len(pollutants), endpointFalse).tolist() iaqi_values iaqi_values[:1] angles angles[:1] fig, ax plt.subplots(figsize(8, 8), subplot_kwdict(projectionpolar)) ax.plot(angles, iaqi_values, o-, linewidth2, labelf日期: {target_date}) ax.fill(angles, iaqi_values, alpha0.25) # 设置标签 ax.set_xticks(angles[:-1]) ax.set_xticklabels(pollutants, fontsize11) ax.set_ylim(0, max(iaqi_values) * 1.2 if max(iaqi_values) 0 else 100) ax.set_title(f污染物IAQI贡献雷达图 ({target_date}), fontsize13, fontweightbold, pad20) ax.grid(True) ax.legend(locupper right, bbox_to_anchor(1.3, 1.0)) plt.tight_layout() plt.savefig(fiaqi_radar_{target_date}.png, dpi300, bbox_inchestight) plt.show()5. 整合与实战运行完整的AQI计算器现在我们把所有模块像拼图一样组合起来形成一个完整的、可执行的脚本。# main.py import pandas as pd import numpy as np import matplotlib.pyplot as plt from datetime import datetime # 假设前面的函数load_concentration_data, calculate_iaqi, calculate_aqi_for_row, # calculate_aqi_for_dataframe, plot_aqi_trend, plot_iaqi_radar都定义在同一个文件或已导入 def main(): 主函数串联整个AQI计算流程。 print( Python AQI 计算器 \n) # 1. 加载数据 data_file sample_data.csv # 替换为你的数据文件路径 df_raw load_concentration_data(data_file) if df_raw is None: return # 2. 核心计算 print(\n正在进行AQI计算...) df_with_aqi calculate_aqi_for_dataframe(df_raw) # 3. 展示结果 print(\n计算完成结果预览) print(df_with_aqi[[date, PM2.5, PM10, O3, AQI, Primary_Pollutant]].to_string(indexFalse)) # 4. 输出详细结果到CSV output_file faqi_results_{datetime.now().strftime(%Y%m%d_%H%M%S)}.csv df_with_aqi.to_csv(output_file, indexFalse, encodingutf-8-sig) print(f\n详细结果已保存至: {output_file}) # 5. 可视化 print(\n生成可视化图表...) # 确保日期列名正确 date_column date # 根据你的数据列名调整 plot_aqi_trend(df_with_aqi, date_coldate_column) # 选择其中一天绘制雷达图 if len(df_with_aqi) 0: sample_date df_with_aqi.iloc[0][date_column] # 取第一天 plot_iaqi_radar(df_with_aqi, sample_date, date_coldate_column) print(\n程序执行完毕。) if __name__ __main__: main()运行这个main.py脚本你会看到控制台输出计算结果同时当前目录下会生成一个包含详细数据的CSV文件以及两张PNG格式的图表。整个过程从原始数据到分析报告一气呵成。提示在实际项目中你可能需要处理更复杂的情况例如数据缺失、异常值、不同时间尺度小时值vs日均值的计算差异。本文提供的代码是一个坚实的起点你可以在此基础上进行扩展比如添加数据清洗模块、连接实时API、构建Web界面或开发更复杂的预测模型。这个项目的代码包包括sample_data.csv、完整的aqi_calculator.py整合了所有函数和main.py我已经整理好。你可以通过它来复现整个过程并以此为模板处理你自己的空气质量数据。亲手运行一遍代码看着原始数据一步步变成图表上的点和线那种对数据“知其所以然”的掌控感是使用任何现成工具都无法替代的。