基于python实现Modis数据的检索与下载
本文介绍一个用于从 NASA 的 MODIS 数据平台批量下载文件的 Python 脚本。该脚本支持通过 HTTP 请求搜索数据文件，并通过多线程机制进行高效下载。它可以处理代理、断点续传，并提供日志记录功能，方便调试和监控下载过程。
一、代码功能概述
-
MODISDownloader 类
- 初始化：设置 NASA LADSWEB MODIS 数据 API 令牌、代理（可选）、最大线程数，以及下载的基础 URL。
- download_file_by_name(filename, output_path)：通过指定的文件名直接下载文件到指定路径，使用
curl命令进行断点续传。 - extract_file_urls(fileInfo)：从 API 返回的文件信息中提取文件的下载 URL 列表。
- search_files_and_get_url(search_params)：根据指定的查询参数（如产品类型、日期范围、区域等）进行数据文件的检索，返回符合条件的文件信息，并提取下载链接。
- download_search_files(output_path)：使用多线程下载所有搜索到的文件，线程池的大小由初始化参数
max_workers决定。 - _download_with_curl(url, output_path, proxy)：通过
curl命令下载文件，支持断点续传和代理配置。 - _execute_command(command)：执行系统命令并处理潜在的错误，记录执行过程。
-
日志记录
该脚本使用 Python 的logging库来记录程序的执行过程，包括下载开始、成功、失败等信息，方便追踪进度和排查错误。
二、使用方法
-
安装依赖 在运行该脚本之前，请确保安装以下 Python 库： pip install requests
-
脚本参数配置
1）token：NASA LADSWEB 平台的授权令牌，需自行申请并填写（注册 NASA Earthdata 账户，登录成功后，在个人中心找密钥相关的就能生成密钥）。 2）proxy：可选的代理服务器地址，用于在特定网络环境下进行访问。 3）max_workers：多线程下载时的线程数，默认设置为4。 -
实例化 MODISDownloader 类
token = "your_token_here" proxy = "http://127.0.0.1:7000" downloader = MODISDownloader(token=token, proxy=proxy) -
通过文件名下载文件
downloader.download_file_by_name("/archive/allData/61/MOD021KM/2024/223/MOD021KM.A2024223.0240.061.2024223142954.hdf", "F:/lv/test.hdf") -
通过查询参数检索并下载文件
- 设置查询条件，如产品类型、时间范围等。
search_params = { "product": "MOD021KM", "collection": "61", "dateRanges": "2024-08-12..2024-08-13", "areaOfInterest": "x69.4y52,x106.5y36", "dayCoverage": "true", "dnboundCoverage": "true" } downloader.search_files_and_get_url(search_params) downloader.download_search_files(r"F:\lv\test")
总结
该脚本适用于需要批量下载 NASA MODIS 数据的场景，尤其是需要通过检索条件筛选文件的情况。它使用 requests 进行数据搜索，curl 进行高效下载，并结合多线程机制显著提升下载效率。
import logging
import subprocess
import os
from concurrent.futures import ThreadPoolExecutorimport requestsclass MODISDownloader:def __init__(self, token, proxy=None, max_workers=4):self.token = token # 存储令牌self.base_url = "https://ladsweb.modaps.eosdis.nasa.gov/" # 基础 URL，用于访问 MODIS 数据self.base_serch_url = "https://ladsweb.modaps.eosdis.nasa.gov/api/v1/files/" # 基础 URL，用于检索 MODIS 数据self.headers = ["X-Requested-With: XMLHttpRequest",f"Authorization: Bearer {self.token}" # 在请求中包含令牌]self.search_files = []self.max_workers = max_workersself.proxy = proxylogging.basicConfig(level=logging.INFO) # 设置日志记录级别为 INFOdef download_file_by_name(self, filename, output_path):"""通过文件名下载文件"""url = f"{self.base_url}{filename}" # 构建文件的 URLlogging.info(f"即将下载: {url}") # 记录即将执行的命令self._download_with_curl(url, output_path, proxy=self.proxy) # 使用 curl 下载def extract_file_urls(self, fileInfo):"""从 MODIS 数据字典中提取完整的文件下载 URL 列表"""for file_id, file_info in fileInfo.items():file_url = f"{file_info['fileURL']}"self.search_files.append(file_url)def search_files_and_get_url(self, search_params):"""搜索并下载符合条件的文件"""query_string = "&".join([f"{key}={value}" for key, value in search_params.items()])# 生成完整的 URLfull_url = f"{self.base_serch_url}{query_string}"# 设置代理配置proxies = {"http": self.proxy,"https": self.proxy} if self.proxy else Noneresponse = requests.get(full_url, proxies=proxies, verify=False)# 检查响应状态if response.status_code == 200:fileInfo = response.json() # 返回 JSON 格式的数据self.extract_file_urls(fileInfo)logging.info(f"已检索到{len(self.search_files)}个文件！") # 记录即将执行的命令else:logging.error(f"搜索失败，状态码: {response.status_code}")def download_search_files(self, output_path):"""多线程下载搜索到的文件"""with ThreadPoolExecutor(max_workers=self.max_workers) as executor:futures = []for file_url in self.search_files:output_file_path = os.path.join(output_path, os.path.basename(file_url))futures.append(executor.submit(self.download_file_by_name, file_url, output_file_path))for future in futures:try:future.result() # 获取线程的结果，处理异常except Exception as e:logging.error(f"下载过程中出现错误: {str(e)}")def _download_with_curl(self, url, output_path, proxy=None):"""使用 curl 进行下载"""command = ["curl","--header", self.headers[0],"--header", self.headers[1],"-o", output_path,"--create-dirs","-C", "-", # 断点续传"-#", # 显示进度]# 如果指定了代理，则添加代理参数if proxy:command.extend(["--proxy", proxy])command.append(url)self._execute_command(command)def _execute_command(self, command):"""执行 shell 命令并进行错误处理"""try:logging.info(f"执行: {' '.join(command)}") # 记录即将执行的命令result = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)logging.info(f"{command}\n执行成功!") # 记录命令输出except subprocess.CalledProcessError as e:logging.error(f"执行失败，错误原因为: {e.stderr.decode()}") # 记录命令错误except Exception as e:logging.error(f"未知错误: {str(e)}") # 处理其他异常# 使用示例
if __name__ == "__main__":# # 代替为您的实际令牌# token = ""# downloader = MODISDownloader(token)# # 通过文件名下载# downloader.download_file_by_name("/archive/allData/61/MOD021KM/2024/223/MOD021KM.A2024223.0240.061.2024223142954.hdf",# "F:/lv/test.hdf")###################################################################################################################### 代替为您的实际令牌token = ""proxy = "http://127.0.0.1:7000"max_workers=4downloader = MODISDownloader(token=token, proxy=proxy, max_workers=max_workers)# 搜索并下载search_params = {"product": "MOD021KM","collection": "61","dateRanges": "2024-08-12..2024-08-13","areaOfInterest": "x69.4y52,x106.5y36","dayCoverage": "true","dnboundCoverage": "true"}downloader.search_files_and_get_url(search_params)downloader.download_search_files(r"F:\lv\test")