夜雨聆风项目背景
之前下载了一块谷歌19级影像图，然后下了osm全国的建筑数据，想那这个和数据，做一个样本示例，升入了解一下样本训练的过程和生成的记过。写好要求交给Agents，在没有约束所使用的库时，代码使用了一下库。
import osimport jsonimport zipfileimport requestsimport numpy as npfrom pathlib import Pathfrom typing importDict, Tuple, Optional, Any# 地理数据处理库import geopandas as gpdimport rasteriofrom rasterio.crs import CRSfrom rasterio.mask import maskfrom rasterio.features import rasterizefrom rasterio.transform import from_boundsfrom shapely.geometry import box, mappingimport matplotlib.pyplot as plt约束使用geoai-py库后有了才有了今天这篇文章。
数据来源
本项目基于 STAC（SpatioTemporal Asset Catalog）标准数据管理平台进行地理空间数据管理。前段时间撸好了一个STAC时空数据资产管理平台（下个专题可以介绍一下STAC）然后将tif影像和OSM的shape上传到数管平台中，数据来源包括：
• 高分辨率影像数据：Google 19级影像，覆盖西安市部分地区，通过 STAC item 访问 • 建筑矢量数据：OpenStreetMap 建筑 shp 数据，覆盖全国范围，以 ZIP 格式存储于 STAC 平台
数据特征
系统环境
• Python 环境：conda geoai • 核心依赖库：geoai-py、geopandas、rasterio、requests • 处理工具：GDAL/OGR 地理空间库
处理流程
init_dirs() | |||
get_asset_href()download_file() | |||
unzip_vector() | |||
align_vector_to_raster() | |||
export_geotiff_tiles() |
核心代码解析
STAC 资产获取
defget_asset_href(item_url): r = requests.get(item_url) r.raise_for_status() item = r.json() assets = item.get("assets", {})for _, asset in assets.items(): href = asset.get("href")if href:return hrefraise Exception("未找到资产地址")技术要点：
• 通过 REST API 获取 STAC Item 元数据 • 遍历 assets 获取第一个有效资源地址 • 支持标准 STAC API 规范
矢量与影像对齐
defalign_vector_to_raster(raster_path, shp_path):with rasterio.open(raster_path) as src: raster_crs = src.crs raster_bounds = src.bounds raster_geom = box(*raster_bounds) gdf = gpd.read_file(shp_path)if gdf.crs != raster_crs: gdf = gdf.to_crs(raster_crs) gdf = gdf[gdf.is_valid] clip_gdf = gdf.clip(raster_geom) clip_gdf.to_file(aligned_shp, driver="GeoJSON")return aligned_shp处理流程：
1. 读取影像获取 CRS 和边界范围 2. 读取矢量数据并检查有效性 3. 坐标转换（如需要） 4. 按影像边界裁剪矢量 5. 输出 GeoJSON 格式
训练样本生成
geoai.export_geotiff_tiles( in_raster=str(raster_path), out_folder=str(OUTPUT_DIR), in_class_data=str(vector_path), tile_size=512, stride=256, skip_empty_tiles=True, buffer_radius=0, all_touched=True, create_overview=True, apply_augmentation=True, augmentation_count=2, quiet=False,)参数说明：
tile_size | ||
stride | ||
skip_empty_tiles | ||
all_touched | ||
apply_augmentation | ||
augmentation_count |
数据增强策略
增强类型
geoai-py 库支持多种数据增强方式：
• 几何变换：随机翻转、旋转、平移 • 颜色变换：亮度、对比度、饱和度调整 • 噪声注入：高斯噪声、椒盐噪声 • 模糊处理：高斯模糊
增强效果
数据增强使原始样本数量扩展为原来的 3 倍（原始 + 2 次增强），有效提升模型泛化能力。
处理结果统计
数据规模
根据 metadata.json 统计：
样本分布
• 有效切片数：约 490+ 个（去除空白区域） • 标注文件数：约 200+ 个（仅包含目标的切片） • 数据增强倍数：3x
技术亮点
STAC 标准集成
项目采用 STAC 标准进行数据管理，具有以下优势：
• 统一的元数据规范 • 灵活的数据源接入 • 便于数据版本管理
自动化处理流程
从数据下载到样本生成实现全流程自动化：
• 自动创建目录结构 • 自动坐标系统一 • 自动空间范围裁剪 • 智能跳过无效区域
高效数据处理
• 并行处理：geoai-py 内部支持多线程 • 内存优化：分块读取大文件 • 增量处理：跳过已存在文件
运行方式
cd GIS/geoai/makelablepython geoai-process.py运行日志示例
训练样本比较慢
注意事项
1. 网络依赖：需确保 STAC 服务正常运行 2. 存储空间：处理大型影像需预留足够磁盘空间 3. 内存配置：建议 16GB+ 内存处理高分辨率影像 4. 坐标系统：确保输入数据坐标系统一或支持自动转换
完整源码
import shutilimport zipfilefrom pathlib import Pathimport geoaiimport geopandas as gpdimport rasterioimport requestsfrom shapely.geometry import box# =========================================================# STAC 配置# =========================================================RASTER_ITEM_URL = ("http://localhost:8010/collections/raster-default/items/""0dcc269fd72a4c4e8a3c388564122fd1")VECTOR_ITEM_URL = ("http://localhost:8010/collections/vector-default/items/""90dbe2fc21b84d83891b87fd61ed24e6")BASE_DIR = Path("./data")RAW_RASTER_DIR = BASE_DIR / "raw/raster"RAW_VECTOR_DIR = BASE_DIR / "raw/vector"ALIGN_RASTER_DIR = BASE_DIR / "aligned/raster"ALIGN_VECTOR_DIR = BASE_DIR / "aligned/vector"OUTPUT_DIR = BASE_DIR / "output"# =========================================================# 初始化目录# =========================================================definit_dirs(): dirs = [ RAW_RASTER_DIR, RAW_VECTOR_DIR, ALIGN_RASTER_DIR, ALIGN_VECTOR_DIR, OUTPUT_DIR, ]for d in dirs: d.mkdir(parents=True, exist_ok=True)print("目录初始化完成")# =========================================================# 获取 STAC Asset# =========================================================defget_asset_href(item_url): r = requests.get(item_url) r.raise_for_status() item = r.json() assets = item.get("assets", {})for _, asset in assets.items(): href = asset.get("href")if href:return hrefraise Exception("未找到资产地址")# =========================================================# 下载文件# =========================================================defdownload_file(url, save_path):if save_path.exists():print(f"文件已存在: {save_path}")returnprint(f"下载: {url}") r = requests.get(url, stream=True) r.raise_for_status()withopen(save_path, "wb") as f:for chunk in r.iter_content(chunk_size=8192): f.write(chunk)print(f"保存: {save_path}")# =========================================================# 解压 shp zip# =========================================================defunzip_vector(zip_path, extract_dir):with zipfile.ZipFile(zip_path, "r") as zip_ref: zip_ref.extractall(extract_dir) shp_files = list(extract_dir.glob("*.shp"))ifnot shp_files:raise Exception("未找到 shp 文件")return shp_files[0]# =========================================================# 矢量与影像对齐# =========================================================defalign_vector_to_raster(raster_path, shp_path):print("开始对齐矢量与影像")with rasterio.open(raster_path) as src: raster_crs = src.crs raster_bounds = src.bounds raster_geom = box(*raster_bounds)print(f"影像 CRS: {raster_crs}") gdf = gpd.read_file(shp_path)print(f"矢量 CRS: {gdf.crs}")# 坐标统一if gdf.crs != raster_crs:print("执行坐标转换") gdf = gdf.to_crs(raster_crs)# 修复 geometry gdf = gdf[gdf.is_valid]# 裁剪 clip_gdf = gdf.clip(raster_geom)print(f"裁剪后数量: {len(clip_gdf)}") aligned_shp = ALIGN_VECTOR_DIR / "buildings_clip.geojson" clip_gdf.to_file( aligned_shp, driver="GeoJSON" )print(f"保存裁剪结果: {aligned_shp}")return aligned_shp# =========================================================# 准备影像# =========================================================defprepare_raster(raster_path): out_raster = ALIGN_RASTER_DIR / "image_align.tif"ifnot out_raster.exists(): shutil.copy(raster_path, out_raster)return out_raster# =========================================================# 使用 geoai-py 生成训练样本# =========================================================defgenerate_training_data( raster_path, vector_path):print("=" * 60)print("开始生成训练样本")print("=" * 60) geoai.export_geotiff_tiles( in_raster=str(raster_path), out_folder=str(OUTPUT_DIR), in_class_data=str(vector_path),# 切片大小 tile_size=512,# 滑窗步长 stride=256,# 只保留有目标的切片 skip_empty_tiles=True,# polygon buffer buffer_radius=0,# 包含接触边界像素 all_touched=True,# 创建 overview create_overview=True,# 数据增强 apply_augmentation=True,# 每个 tile 增强数量 augmentation_count=2, quiet=False, )print("训练样本生成完成")# =========================================================# 主函数# =========================================================defmain():print("=" * 60)print("GeoAI 样本生产")print("=" * 60)# 初始化 init_dirs()# 获取资产地址 raster_href = get_asset_href(RASTER_ITEM_URL) vector_href = get_asset_href(VECTOR_ITEM_URL)# 下载 raster_path = RAW_RASTER_DIR / "image.tif" vector_zip = RAW_VECTOR_DIR / "buildings.zip" download_file(raster_href, raster_path) download_file(vector_href, vector_zip)# 解压 vector_extract_dir = RAW_VECTOR_DIR / "buildings" shp_path = unzip_vector( vector_zip, vector_extract_dir )# 对齐 aligned_vector = align_vector_to_raster( raster_path, shp_path ) aligned_raster = prepare_raster( raster_path )# 生成训练数据 generate_training_data( aligned_raster, aligned_vector )print("\n")print("=" * 60)print("全部完成")print("=" * 60)print(f"输出目录:\n{OUTPUT_DIR}")# =========================================================# 入口# =========================================================if __name__ == "__main__": main()项目地址会在地下留言中添加
