GEE 将本地 GeoJSON 文件上传到谷歌资产

在地理信息系统（GIS）领域，Google Earth Engine（GEE）是一个强大的平台，它允许用户处理和分析大规模地理空间数据。本文将介绍如何使用 Python 脚本批量上传本地 GeoJSON 文件到 GEE 资产存储，这对于需要将地理数据上传到 GEE 进行进一步分析的用户来说非常有用。

应用场景

数据集成：将本地 GeoJSON 数据集成到 GEE 中，以便进行更复杂的地理空间分析。
数据共享：与团队成员共享 GeoJSON 数据，以便协作分析和决策。
自动化处理：自动化数据上传流程，减少手动操作，提高效率。

使用方法

安装必要的库：确保你的 Python 环境中安装了 earthengine-api 和 json 库。
设置代理：如果你在中国大陆使用 GEE，可能需要设置代理来访问 GEE 服务。
授权 GEE 账户：使用 ee.Authenticate() 和 ee.Initialize() 函数进行授权和初始化。
运行脚本：执行脚本，脚本将自动处理指定文件夹中的所有 GeoJSON 文件，并上传到 GEE。

代码详解

导入库

# 导入必要的库
import ee
import os
import json
import time

这段代码导入了几个Python模块，用于后续的文件处理和与Google Earth Engine (GEE) 的交互。

ee: 用于与Google Earth Engine进行交互的专用库。
os: 用于处理操作系统功能，如文件路径操作。
json: 用于解析JSON格式的数据。
time: 提供各种和时间相关的功能。

设置代理和环境变量

# 设置代理和环境变量
os.environ['HTTP_PROXY'] = 'http://127.0.0.1:xxxx' 
os.environ['HTTPS_PROXY'] = 'http://127.0.0.1:xxxx' 
os.environ['CRYPTOGRAPHY_OPENSSL_NO_LEGACY'] = '1'

# 方法参见`https://mp.weixin.qq.com/s?__biz=Mzk0MTU1MjU5Mw==&mid=2247484766&idx=1&sn=40db6ec347539999af2332ba1e4996fb&chksm=c2d1e3e0f5a66af6d564bf19a1d6a260e5addaf6b7c42981c0e9917db7feb62f9bcb83fdaa9d#rd`

这段代码设置了两个环境变量，用于配置HTTP和HTTPS代理，以便程序可以通过指定的代理服务器进行网络请求。这对于访问某些需要通过特定网络路径访问的资源非常有用。CRYPTOGRAPHY_OPENSSL_NO_LEGACY 环境变量用于解决某些Python库与新版OpenSSL的兼容性问题。

授权和初始化 GEE

# 授权 GEE 账户并初始化 GEE API
ee.Authenticate()
ee.Initialize()

这段代码首先调用ee.Authenticate()函数进行用户认证，然后调用ee.Initialize()函数初始化GEE API。这是使用GEE进行任何操作前的必要步骤。

定义删除资产的函数

def delete_asset_if_exists(asset_path):
    """如果资产存在，则删除它。"""
    try:
        asset = ee.data.getAsset(asset_path)
        ee.data.deleteAsset(asset_path)
        print(f"Asset {asset_path} has been deleted.")
    except ee.EEException as e:
        if "not found" in str(e):
            print(f"Asset {asset_path} does not exist.")
        else:
            raise

这个函数尝试删除指定路径的GEE资产。如果资产存在，它将被删除并打印一条消息；如果资产不存在，将打印一条不存在的消息。如果发生其他类型的异常，则会抛出该异常。

获取特定扩展名的文件列表

def get_files_with_extension(directory, extension):
    """获取指定目录下所有具有特定扩展名的文件。"""
    directory = os.path.normpath(directory)
    all_files = os.listdir(directory)
    files_with_extension = [file for file in all_files if file.endswith(extension)]
    return files_with_extension

这个函数接收一个目录路径和文件扩展名作为参数，返回该目录下所有具有指定扩展名的文件列表。

主函数

def main():
    # 设置 JSON 文件路径
    json_folder_path = r'E:\Geojson'
    json_files = get_files_with_extension(directory=json_folder_path, extension='.geojson')

    # 遍历每个 GeoJSON 文件
    for json_file in json_files:
        json_name = json_file.split('.')[0]
        local_geojson_path = os.path.join(json_folder_path, json_file)

        # 打开并读取 GeoJSON 文件
        with open(local_geojson_path, 'r') as file:
            geojson_data = json.load(file)

        # 创建 Feature Collection
        feature_collection = ee.FeatureCollection([
            ee.Feature(
                ee.Geometry(geojson_feature['geometry']),
                {**geojson_feature['properties'], 'system:index': str(index)}
            )
            for index, geojson_feature in enumerate(geojson_data['features'])
        ])

        # 定义资产路径
        asset_path = fr'users/{json_name}'

        # 删除已存在的资产
        delete_asset_if_exists(asset_path)

        # 创建并启动上传任务
        task = ee.batch.Export.table.toAsset(
            collection=feature_collection,
            description=fr'Upload_geojson_to_asset_{json_name}',
            assetId=asset_path,
            fileFormat='GeoJSON'
        )

        task.start()
        print(f"Upload task started for {json_file}.")

        # 监控上传状态
        while True:
            if not task.active():
                print("Task completed.")
                break
            else:
                print("Task status:", task.status())
                time.sleep(10)

主函数执行以下步骤：

设置GeoJSON文件的存储路径。
使用get_files_with_extension函数获取所有GeoJSON文件。
遍历每个文件，读取其内容，并创建GEE的FeatureCollection。
定义每个文件对应的GEE资产路径。
如果资产已存在，则删除它。
创建一个上传任务，将FeatureCollection上传到GEE资产存储。
启动上传任务，并使用循环监控任务状态，直到任务完成。

这个脚本为批量上传GeoJSON文件到GEE提供了一个完整的解决方案，包括处理文件、创建特征集合、上传和监控任务状态。

完整案例

下列代码提供了一个完整的案例，展示了如何批量上传 GeoJSON 文件到 GEE 资产存储。

# -*- coding:utf-8 -*-
"""
此代码的主要用途是批量上传本地 GeoJSON 文件到 Google Earth Engine (GEE) 资产存储。
将多个本地 GeoJSON 文件上传到 GEE 资产存储，并确保每次上传前都清除了同名的已有资产。

"""

import ee
import os
import json
import time

# 构建网络代理
os.environ['HTTP_PROXY'] = 'http://127.0.0.1:xxxx'
os.environ['HTTPS_PROXY'] = 'http://127.0.0.1:xxxx'
# 设置环境变量以解决 OpenSSL 3.0 的兼容性问题
os.environ['CRYPTOGRAPHY_OPENSSL_NO_LEGACY'] = '1'
# 授权 Earth Engine 账户及初始化 Earth Engine API
ee.Authenticate()
ee.Initialize(project='ee-xxxx')


def delete_asset_if_exists(asset_path):
    """
    检查并删除指定的资产路径，如果存在的话。

    参数:
    asset_path (str): 资产路径。
    """
    try:
        asset = ee.data.getAsset(asset_path)
        ee.data.deleteAsset(asset_path)
        print(f"Asset {asset_path} has been deleted.")
    except ee.EEException as e:
        if "not found" in str(e):
            print(f"Asset {asset_path} does not exist.")
        else:
            raise


def get_files_with_extension(directory, extension):
    """
    获取指定文件夹中具有特定扩展名的所有文件名。

    参数:
    directory (str): 文件夹路径。
    extension (str): 文件扩展名（包括点，例如 '.shp'）。

    返回:
    list: 包含指定扩展名文件名的列表。
    """
    # 确保路径字符串是正确的格式
    directory = os.path.normpath(directory)

    # 获取文件夹中的所有文件
    all_files = os.listdir(directory)

    # 筛选出指定扩展名的文件
    files_with_extension = [file for file in all_files if file.endswith(extension)]

    # 返回结果
    return files_with_extension


def main():
    """
    主函数，用于批量上传 GeoJSON 文件到 Google Earth Engine (GEE) 资产存储。
    """
    # 确定裁剪的图像边界范围
    json_folder_path = r'E:\Geojson'
    json_files = get_files_with_extension(directory=json_folder_path, extension='.geojson')

    for json_file in json_files:
        json_name = json_file.split('.')[0]

        local_geojson_path = os.path.join(json_folder_path, json_file)

        # 读取 GeoJSON 文件
        with open(local_geojson_path, 'r') as file:
            geojson_data = json.load(file)

        # 将 GeoJSON 数据转换为 GEE 的 FeatureCollection
        feature_collection = ee.FeatureCollection([
            ee.Feature(
                ee.Geometry(geojson_feature['geometry']),
                {**geojson_feature['properties'], 'system:index': str(index)}
            )
            for index, geojson_feature in enumerate(geojson_data['features'])
        ])

        # 指定上传到 GEE 资产的路径
        asset_path = fr'ISID_{json_name}'

        # 检查并删除资产，如果存在的话
        delete_asset_if_exists(asset_path)

        # 创建上传任务
        task = ee.batch.Export.table.toAsset(
            collection=feature_collection,
            description=fr'Upload_geojson_to_asset_{json_name}',
            assetId=asset_path,
            fileFormat='GeoJSON'
        )

        # 开始上传任务
        task.start()
        print(f"Upload task started for {json_file}.")

        # 等待任务完成
        while True:
            if not task.active():
                print("Task completed.")
                break
            else:
                print("Task status:", task.status())
                time.sleep(10)


if __name__ == '__main__':
    main()

注意事项

术语/函数	解释
`ee.Authenticate()`	用户需要手动打开浏览器完成登录认证的过程。
`ee.Initialize()`	初始化 Earth Engine API，通常需要传入一个项目名。
`ee.FeatureCollection`	创建一个特征集合，用于存储地理特征数据。
`ee.batch.Export`	提供了导出数据的方法，如将数据导出到 Google Drive 或 Asset Manager。
`os.environ`	设置环境变量，这里用来配置代理服务器地址。
`os.listdir()`	列出指定目录下的所有文件和子目录。
`json.load()`	将文件中的 JSON 数据解析成 Python 对象。
`ee.Geometry`	表示地理几何对象，如点、线或多边形等。
`ee.Feature`	代表地理空间中的一个要素，包含几何信息和属性信息。
`task.active()`	检查任务是否还在活跃状态。