用Python构建森林火灾实时监测系统的实战指南森林火灾是全球生态安全的重要威胁传统的人工巡查方式已无法满足大范围、高频次的监测需求。本文将手把手教你如何利用Python调用卫星遥感API搭建一套自动化火灾监测系统从环境配置到报警逻辑实现覆盖完整开发流程。1. 环境准备与API接入在开始编码前我们需要配置好开发环境并获取API访问权限。以下是具体步骤Python环境配置# 创建虚拟环境推荐 python -m venv fire_monitor source fire_monitor/bin/activate # Linux/Mac fire_monitor\Scripts\activate # Windows # 安装依赖库 pip install requests pandas geopy matplotlib scheduleAPI密钥申请访问星图云开发者平台注册账号进入我的应用创建新项目获取专属的API token通常以sk_开头的一串字符测试API连通性import requests API_URL https://api.open.geovisearth.com/v2/fire/api/dailyListPage TOKEN your_api_token_here # 替换为实际token def test_api(): params { current: 1, size: 1, query: {regionCode: 530000} # 云南省行政区划代码 } headers {token: TOKEN} try: response requests.post(API_URL, jsonparams, headersheaders) if response.status_code 200: print(API连接成功) return response.json() else: print(f连接失败状态码{response.status_code}) except Exception as e: print(f发生异常{str(e)}) test_api()注意实际开发中应将API token存储在环境变量中避免硬编码在脚本里。可以使用python-dotenv库管理敏感信息。2. 核心数据获取与解析成功连接API后我们需要设计高效的数据获取和解析方案。卫星遥感数据通常包含以下关键字段字段名数据类型说明latitudefloat火点纬度坐标longitudefloat火点经度坐标observeTimestring观测时间(UTC)reliabilitystring可信度评分(0-1)surfaceTypestring地表类型(林地/草地等)nearbyPlaceNamesstring最近的地理标志物完整的数据获取函数实现import pandas as pd from datetime import datetime def fetch_fire_data(region_code530000, hours24): 获取指定区域的最新火情数据 params { current: 1, size: 100, # 最大允许值 query: { regionCode: region_code, timeRange: [ (datetime.utcnow() - timedelta(hourshours)).strftime(%Y-%m-%d %H:%M:%S), datetime.utcnow().strftime(%Y-%m-%d %H:%M:%S) ] } } response requests.post(API_URL, jsonparams, headers{token: TOKEN}) if response.status_code ! 200: raise Exception(fAPI请求失败: {response.text}) raw_data response.json()[data][records] # 数据清洗与转换 df pd.DataFrame(raw_data) df[observeTime] pd.to_datetime(df[observeTime]) df[reliability] df[reliability].astype(float) # 解析地表类型比例 df[land_cover] df[surfaceTypeRatio].apply( lambda x: {k: float(v) for k, v in json.loads(x).items()} ) return df.sort_values(observeTime, ascendingFalse)数据质量优化技巧使用reliability字段过滤低可信度火点结合surfaceType优先关注林地火情对连续时间点的火点进行去重处理3. 实时监测系统架构设计要实现可靠的实时监测我们需要设计合理的系统架构。以下是推荐的技术方案数据采集层 → 数据处理层 → 报警判断层 → 通知输出层各模块具体实现定时任务调度import schedule import time def monitoring_job(): print(f{datetime.now()} - 开始执行监测任务...) try: new_fires detect_new_fires() if not new_fires.empty: send_alert(new_fires) except Exception as e: log_error(e) # 每30分钟执行一次 schedule.every(30).minutes.do(monitoring_job) while True: schedule.run_pending() time.sleep(1)新火情检测逻辑from geopy.distance import geodesic def detect_new_fires(radius_km5): 检测半径范围内新出现的火点 latest_df fetch_fire_data() if latest_df.empty: return pd.DataFrame() # 获取上次检查的火点位置 last_check load_last_positions() # 从数据库或文件读取 new_fires [] for _, row in latest_df.iterrows(): is_new True current_pos (row[latitude], row[longitude]) for last_pos in last_check: if geodesic(current_pos, last_pos).km radius_km: is_new False break if is_new and row[reliability] 0.7: # 可信度阈值 new_fires.append(row) save_current_positions(latest_df) # 保存当前状态 return pd.DataFrame(new_fires)报警规则配置火点面积超过阈值通过附近火点密度估算火点向保护区方向移动同一区域持续出现新火点4. 可视化与报警集成有效的可视化能帮助快速定位火情而及时的报警机制则是系统的核心价值所在。动态地图可视化import folium from folium.plugins import HeatMap def generate_fire_map(fire_data, save_pathfire_map.html): 生成交互式火情地图 if fire_data.empty: return None # 创建基础地图 center [fire_data[latitude].mean(), fire_data[longitude].mean()] fire_map folium.Map(locationcenter, zoom_start9) # 添加热力图层 heat_data [[row[latitude], row[longitude], row[reliability]*10] for _, row in fire_data.iterrows()] HeatMap(heat_data, radius15).add_to(fire_map) # 标记单个火点 for _, row in fire_data.iterrows(): popup_text f 时间: {row[observeTime]}br 可信度: {row[reliability]:.1%}br 地表类型: {row[surfaceType]}br 附近位置: {row[nearbyPlaceNames]} folium.CircleMarker( location[row[latitude], row[longitude]], radius5, colorred, fillTrue, popuppopup_text ).add_to(fire_map) fire_map.save(save_path) return save_path多通道报警集成邮件报警示例import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart def send_email_alert(fire_data, recipients): msg MIMEMultipart() msg[Subject] f森林火灾警报 - 发现{len(fire_data)}个新火点 msg[From] monitoryourdomain.com msg[To] , .join(recipients) html f h2最新火情通报/h2 p检测时间: {datetime.now().strftime(%Y-%m-%d %H:%M)}/p ol {.join(fli{row[nearbyPlaceNames]} (可信度: {row[reliability]:.0%})/li for _, row in fire_data.iterrows())} /ol p地图查看: a href{generate_fire_map(fire_data)}点击打开/a/p msg.attach(MIMEText(html, html)) with smtplib.SMTP(smtp.yourmailserver.com, 587) as server: server.starttls() server.login(user, password) server.send_message(msg)短信/钉钉机器人报警使用Twilio等短信API服务配置钉钉/企业微信机器人Webhook重要火情触发电话语音报警5. 系统优化与扩展基础功能实现后我们可以从以下几个方向提升系统能力性能优化技巧使用aiohttp替代requests实现异步请求对历史数据建立本地缓存减少API调用采用区域分块查询策略当监测范围较大时数据分析扩展def analyze_fire_trends(df): 火情时空分布分析 trends { hourly: df.groupby(df[observeTime].dt.hour).size(), weekly: df.groupby(df[observeTime].dt.dayofweek).size(), by_type: df[surfaceType].value_counts(), hot_zones: df.groupby( [round(df[latitude], 2), round(df[longitude], 2)] ).size().sort_values(ascendingFalse).head(10) } return trends与其他系统集成对接气象数据API获取风速风向结合地形数据预测火势蔓延方向接入无人机控制系统实现自动巡查6. 实战案例云南林区监测系统去年为云南某自然保护区部署的监测系统中我们实现了以下增强功能自定义区域关注def is_in_protected_area(lat, lng, geojson): 判断火点是否在保护区内 from shapely.geometry import shape, Point protected_area shape(geojson) return protected_area.contains(Point(lng, lat))多级报警策略一级报警保护区内任何火情二级报警保护区周边5公里内可靠火情三级报警其他区域大面积火情值班人员通讯录集成def get_on_duty_officer(): 根据排班表获取当前值班人员 weekday datetime.now().weekday() with open(schedule.json) as f: schedule json.load(f) return schedule[str(weekday)]这套系统上线后保护区的火情平均响应时间从原来的4小时缩短到35分钟成功避免了多起可能酿成大火的险情。
如何用Python调用星图云API实时监测森林火灾（附完整代码示例）
用Python构建森林火灾实时监测系统的实战指南森林火灾是全球生态安全的重要威胁传统的人工巡查方式已无法满足大范围、高频次的监测需求。本文将手把手教你如何利用Python调用卫星遥感API搭建一套自动化火灾监测系统从环境配置到报警逻辑实现覆盖完整开发流程。1. 环境准备与API接入在开始编码前我们需要配置好开发环境并获取API访问权限。以下是具体步骤Python环境配置# 创建虚拟环境推荐 python -m venv fire_monitor source fire_monitor/bin/activate # Linux/Mac fire_monitor\Scripts\activate # Windows # 安装依赖库 pip install requests pandas geopy matplotlib scheduleAPI密钥申请访问星图云开发者平台注册账号进入我的应用创建新项目获取专属的API token通常以sk_开头的一串字符测试API连通性import requests API_URL https://api.open.geovisearth.com/v2/fire/api/dailyListPage TOKEN your_api_token_here # 替换为实际token def test_api(): params { current: 1, size: 1, query: {regionCode: 530000} # 云南省行政区划代码 } headers {token: TOKEN} try: response requests.post(API_URL, jsonparams, headersheaders) if response.status_code 200: print(API连接成功) return response.json() else: print(f连接失败状态码{response.status_code}) except Exception as e: print(f发生异常{str(e)}) test_api()注意实际开发中应将API token存储在环境变量中避免硬编码在脚本里。可以使用python-dotenv库管理敏感信息。2. 核心数据获取与解析成功连接API后我们需要设计高效的数据获取和解析方案。卫星遥感数据通常包含以下关键字段字段名数据类型说明latitudefloat火点纬度坐标longitudefloat火点经度坐标observeTimestring观测时间(UTC)reliabilitystring可信度评分(0-1)surfaceTypestring地表类型(林地/草地等)nearbyPlaceNamesstring最近的地理标志物完整的数据获取函数实现import pandas as pd from datetime import datetime def fetch_fire_data(region_code530000, hours24): 获取指定区域的最新火情数据 params { current: 1, size: 100, # 最大允许值 query: { regionCode: region_code, timeRange: [ (datetime.utcnow() - timedelta(hourshours)).strftime(%Y-%m-%d %H:%M:%S), datetime.utcnow().strftime(%Y-%m-%d %H:%M:%S) ] } } response requests.post(API_URL, jsonparams, headers{token: TOKEN}) if response.status_code ! 200: raise Exception(fAPI请求失败: {response.text}) raw_data response.json()[data][records] # 数据清洗与转换 df pd.DataFrame(raw_data) df[observeTime] pd.to_datetime(df[observeTime]) df[reliability] df[reliability].astype(float) # 解析地表类型比例 df[land_cover] df[surfaceTypeRatio].apply( lambda x: {k: float(v) for k, v in json.loads(x).items()} ) return df.sort_values(observeTime, ascendingFalse)数据质量优化技巧使用reliability字段过滤低可信度火点结合surfaceType优先关注林地火情对连续时间点的火点进行去重处理3. 实时监测系统架构设计要实现可靠的实时监测我们需要设计合理的系统架构。以下是推荐的技术方案数据采集层 → 数据处理层 → 报警判断层 → 通知输出层各模块具体实现定时任务调度import schedule import time def monitoring_job(): print(f{datetime.now()} - 开始执行监测任务...) try: new_fires detect_new_fires() if not new_fires.empty: send_alert(new_fires) except Exception as e: log_error(e) # 每30分钟执行一次 schedule.every(30).minutes.do(monitoring_job) while True: schedule.run_pending() time.sleep(1)新火情检测逻辑from geopy.distance import geodesic def detect_new_fires(radius_km5): 检测半径范围内新出现的火点 latest_df fetch_fire_data() if latest_df.empty: return pd.DataFrame() # 获取上次检查的火点位置 last_check load_last_positions() # 从数据库或文件读取 new_fires [] for _, row in latest_df.iterrows(): is_new True current_pos (row[latitude], row[longitude]) for last_pos in last_check: if geodesic(current_pos, last_pos).km radius_km: is_new False break if is_new and row[reliability] 0.7: # 可信度阈值 new_fires.append(row) save_current_positions(latest_df) # 保存当前状态 return pd.DataFrame(new_fires)报警规则配置火点面积超过阈值通过附近火点密度估算火点向保护区方向移动同一区域持续出现新火点4. 可视化与报警集成有效的可视化能帮助快速定位火情而及时的报警机制则是系统的核心价值所在。动态地图可视化import folium from folium.plugins import HeatMap def generate_fire_map(fire_data, save_pathfire_map.html): 生成交互式火情地图 if fire_data.empty: return None # 创建基础地图 center [fire_data[latitude].mean(), fire_data[longitude].mean()] fire_map folium.Map(locationcenter, zoom_start9) # 添加热力图层 heat_data [[row[latitude], row[longitude], row[reliability]*10] for _, row in fire_data.iterrows()] HeatMap(heat_data, radius15).add_to(fire_map) # 标记单个火点 for _, row in fire_data.iterrows(): popup_text f 时间: {row[observeTime]}br 可信度: {row[reliability]:.1%}br 地表类型: {row[surfaceType]}br 附近位置: {row[nearbyPlaceNames]} folium.CircleMarker( location[row[latitude], row[longitude]], radius5, colorred, fillTrue, popuppopup_text ).add_to(fire_map) fire_map.save(save_path) return save_path多通道报警集成邮件报警示例import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart def send_email_alert(fire_data, recipients): msg MIMEMultipart() msg[Subject] f森林火灾警报 - 发现{len(fire_data)}个新火点 msg[From] monitoryourdomain.com msg[To] , .join(recipients) html f h2最新火情通报/h2 p检测时间: {datetime.now().strftime(%Y-%m-%d %H:%M)}/p ol {.join(fli{row[nearbyPlaceNames]} (可信度: {row[reliability]:.0%})/li for _, row in fire_data.iterrows())} /ol p地图查看: a href{generate_fire_map(fire_data)}点击打开/a/p msg.attach(MIMEText(html, html)) with smtplib.SMTP(smtp.yourmailserver.com, 587) as server: server.starttls() server.login(user, password) server.send_message(msg)短信/钉钉机器人报警使用Twilio等短信API服务配置钉钉/企业微信机器人Webhook重要火情触发电话语音报警5. 系统优化与扩展基础功能实现后我们可以从以下几个方向提升系统能力性能优化技巧使用aiohttp替代requests实现异步请求对历史数据建立本地缓存减少API调用采用区域分块查询策略当监测范围较大时数据分析扩展def analyze_fire_trends(df): 火情时空分布分析 trends { hourly: df.groupby(df[observeTime].dt.hour).size(), weekly: df.groupby(df[observeTime].dt.dayofweek).size(), by_type: df[surfaceType].value_counts(), hot_zones: df.groupby( [round(df[latitude], 2), round(df[longitude], 2)] ).size().sort_values(ascendingFalse).head(10) } return trends与其他系统集成对接气象数据API获取风速风向结合地形数据预测火势蔓延方向接入无人机控制系统实现自动巡查6. 实战案例云南林区监测系统去年为云南某自然保护区部署的监测系统中我们实现了以下增强功能自定义区域关注def is_in_protected_area(lat, lng, geojson): 判断火点是否在保护区内 from shapely.geometry import shape, Point protected_area shape(geojson) return protected_area.contains(Point(lng, lat))多级报警策略一级报警保护区内任何火情二级报警保护区周边5公里内可靠火情三级报警其他区域大面积火情值班人员通讯录集成def get_on_duty_officer(): 根据排班表获取当前值班人员 weekday datetime.now().weekday() with open(schedule.json) as f: schedule json.load(f) return schedule[str(weekday)]这套系统上线后保护区的火情平均响应时间从原来的4小时缩短到35分钟成功避免了多起可能酿成大火的险情。
