告别手动点击！用Python脚本+Wget批量下载NASA VIIRS夜间灯光数据（附完整代码）

自动化获取NASA VIIRS夜间灯光数据的Python实践指南夜间灯光数据已成为城市发展、能源消耗和灾害评估等领域的重要研究工具。NASA的VIIRSVisible Infrared Imaging Radiometer Suite传感器提供的DBNDay/Night Band数据特别是VNP46系列产品以其高时空分辨率成为学术界和工业界的热门选择。然而手动下载这些数据不仅耗时耗力还容易因网络波动导致前功尽弃。本文将展示如何用Python构建一个健壮的自动化下载系统彻底解放研究人员的双手。1. 理解VIIRS夜间灯光数据体系VIIRS传感器搭载在Suomi NPPSNPP卫星上每天提供全球覆盖的夜间灯光观测。VNP46产品系列包含多个层级的数据处理结果VNP46A1基础产品包含经过大气校正的原始辐射值VNP46A2月光校正版本可直接用于分析VNP46A3月度合成产品VNP46A4年度合成产品# 典型VNP46A1文件命名示例 filename VNP46A1.A2020356.h30v05.001.2020358000243.h5 # 各部分含义 # A2020356 - 采集日期2020年第356天 # h30v05 - MODIS网格编号 # 001 - 集合版本号 # 2020358000243 - 处理时间戳对于需要日尺度分析的研究VNP46A1虽然受月光影响较大但因其更新及时通常延迟仅1-2天仍是动态监测的理想选择。下表对比了主要产品的特性产品代码更新频率月光校正适用场景VNP46A1每日无实时监测、短期变化分析VNP46A2不定期有长期趋势研究目前只到2018年VNP46A3每月有月度经济活动评估VNP46A4每年有年度发展对比提示选择数据产品时需权衡时效性与数据质量。对于需要即时分析的应用VNP46A1配合后期月光校正可能是更优方案。2. 构建自动化下载系统的核心技术2.1 EarthData认证机制解析NASA EarthData使用OAuth 2.0进行身份验证需要处理授权令牌Bearer Token的获取与刷新。以下是关键步骤注册EarthData账号并获取APP_KEY通过API获取临时访问令牌在请求头中加入授权信息处理令牌过期情况import requests from getpass import getpass def get_earthdata_token(username, password): auth_url https://urs.earthdata.nasa.gov/api/users/token response requests.post( auth_url, auth(username, password), headers{Content-Type: application/x-www-form-urlencoded} ) if response.status_code 200: return response.json()[access_token] else: raise Exception(f认证失败: {response.text}) # 安全获取凭证 username input(EarthData用户名: ) password getpass(EarthData密码: ) token get_earthdata_token(username, password)2.2 订单管理与文件列表获取提交数据请求后系统会生成一个订单号通过API可以获取该订单包含的所有文件信息def get_order_files(order_id, token): order_url fhttps://ladsweb.modaps.eosdis.nasa.gov/archive/orders/{order_id}/ headers {Authorization: fBearer {token}} response requests.get(order_url, headersheaders) if response.status_code 200: # 解析HTML获取文件列表 from bs4 import BeautifulSoup soup BeautifulSoup(response.text, html.parser) return [a[href] for a in soup.select(a[href$.h5])] else: raise Exception(f获取订单文件失败: {response.status_code})3. 实现健壮的下载管理器3.1 基于Wget的Python封装虽然可以直接使用系统wget命令但通过Python的subprocess模块封装可以提供更好的控制和错误处理import subprocess import os from pathlib import Path def download_with_wget(url, target_dir, token): Path(target_dir).mkdir(parentsTrue, exist_okTrue) cmd [ wget, -e, robotsoff, -m, -np, -R, *.html,*.tmp, -nH, --cut-dirs3, --header, fAuthorization: Bearer {token}, -P, target_dir, url ] try: result subprocess.run( cmd, checkTrue, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) return True except subprocess.CalledProcessError as e: print(f下载失败: {e.stderr}) return False3.2 断点续传与错误恢复网络不稳定是批量下载大文件时的常见挑战。实现可靠的断点续传机制需要考虑记录已成功下载的文件检测部分下载的文件并恢复设置合理的重试机制import json from datetime import datetime class DownloadTracker: def __init__(self, state_filedownload_state.json): self.state_file state_file self.state self._load_state() def _load_state(self): try: with open(self.state_file, r) as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {completed: [], failed: {}} def save_state(self): with open(self.state_file, w) as f: json.dump(self.state, f, indent2) def add_completed(self, filename): if filename not in self.state[completed]: self.state[completed].append(filename) self.save_state() def record_failure(self, filename, error): self.state[failed][filename] { error: str(error), timestamp: datetime.now().isoformat() } self.save_state() def get_remaining_files(self, all_files): return [f for f in all_files if f not in self.state[completed]]4. 构建完整的自动化流水线4.1 主控制流程实现将各个组件整合为一个完整的自动化系统def automated_download_pipeline(order_id, target_dir, max_retries3): # 初始化组件 tracker DownloadTracker() username input(EarthData用户名: ) password getpass(EarthData密码: ) # 获取认证令牌 try: token get_earthdata_token(username, password) except Exception as e: print(f认证失败: {e}) return # 获取文件列表 try: all_files get_order_files(order_id, token) except Exception as e: print(f获取文件列表失败: {e}) return remaining_files tracker.get_remaining_files(all_files) print(f发现{len(remaining_files)}个待下载文件) # 下载循环 for file_url in remaining_files: retry_count 0 while retry_count max_retries: try: print(f正在下载 {file_url}...) success download_with_wget(file_url, target_dir, token) if success: tracker.add_completed(file_url) print(下载成功) break else: retry_count 1 print(f尝试 {retry_count}/{max_retries} 失败) except Exception as e: tracker.record_failure(file_url, e) retry_count 1 print(f发生异常: {e}) if retry_count max_retries: print(f文件 {file_url} 下载失败已达最大重试次数) print(下载任务完成) print(f成功: {len(tracker.state[completed])}) print(f失败: {len(tracker.state[failed])})4.2 进度监控与通知系统为长时间运行的下载任务添加可视化监控import time from tqdm import tqdm def monitor_download_progress(tracker, check_interval60): prev_count len(tracker.state[completed]) with tqdm(totallen(tracker.state[completed]) len(tracker.state[failed])) as pbar: while True: current_count len(tracker.state[completed]) new_completed current_count - prev_count pbar.update(new_completed) prev_count current_count if len(tracker.state[completed]) len(tracker.state[failed]) pbar.total: break time.sleep(check_interval)5. 高级技巧与优化策略5.1 并行下载加速使用多线程可以显著提高下载效率但需要注意EarthData的请求限制from concurrent.futures import ThreadPoolExecutor, as_completed def parallel_download(file_urls, target_dir, token, max_workers4): with ThreadPoolExecutor(max_workersmax_workers) as executor: futures { executor.submit(download_with_wget, url, target_dir, token): url for url in file_urls } for future in as_completed(futures): url futures[future] try: success future.result() if success: print(f{url} 下载完成) else: print(f{url} 下载失败) except Exception as e: print(f{url} 发生异常: {e})5.2 数据完整性验证NASA提供checksum文件用于验证下载完整性import hashlib def verify_checksum(file_path, expected_md5): hash_md5 hashlib.md5() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(4096), b): hash_md5.update(chunk) return hash_md5.hexdigest() expected_md5 def validate_downloads(target_dir, checksum_file): checksums {} with open(checksum_file, r) as f: for line in f: md5, filename line.strip().split() checksums[filename] md5 for filename, expected_md5 in checksums.items(): file_path os.path.join(target_dir, filename) if not os.path.exists(file_path): print(f{filename} 缺失) continue if verify_checksum(file_path, expected_md5): print(f{filename} 验证通过) else: print(f{filename} 校验失败)在实际项目中这套系统将NASA VIIRS数据的下载时间从数小时的手动操作缩短为完全自动化的过程同时通过完善的错误处理机制确保了数据完整性。一个典型的应用场景是监测东南亚地区夜间灯光变化系统可以设置为每天自动下载最新数据为区域发展研究提供实时数据支持。