集成Cd模型

PROJECT_RULES.md

@@ -0,0 +1,471 @@
+# AcidMap 项目开发规范 (PROJECT RULES)
+
+## 0. 规则文件维护指导 (RULES MAINTENANCE GUIDE)
+
+### 0.1 规则更新原则
+**重要提示**: 当项目发生以下变化时，开发人员和AI助手必须同步更新本规则文件：
+
+#### 0.1.1 必须更新的情况
+1. **新功能模块集成** - 添加新的业务功能、API端点或服务组件
+2. **架构变更** - 修改目录结构、分层架构或技术栈
+3. **开发规范变更** - 修改编码规范、注释风格或命名约定
+4. **部署环境变更** - 更新依赖包、数据库版本或服务器配置
+5. **API接口变更** - 新增、修改或删除API路由和响应格式
+
+#### 0.1.2 更新流程
+1. **识别变更影响** - 确定变更影响的规范章节
+2. **更新相关章节** - 修改对应的规范内容和示例
+3. **验证一致性** - 确保规范与实际代码实现一致
+4. **文档同步** - 同时更新相关的集成文档和使用指南
+
+#### 0.1.3 AI助手责任
+- 在进行功能开发或架构调整时，主动识别需要更新的规范内容
+- 确保新增功能的开发遵循现有规范，或合理扩展规范
+- 提醒维护相关文档的同步更新
+
+## 1. 项目概述
+
+### 1.1 项目基本信息
+- **项目名称**: AcidMap (地图数据处理与预测分析系统)
+- **技术栈**: FastAPI + PostgreSQL + PostGIS + PyTorch
+- **主要功能**: 栅格和矢量地理数据的处理、存储、管理和Cd预测分析
+- **开发语言**: Python 3.8+
+- **架构模式**: 分层架构 (API层 -> Service层 -> Model层)
+
+### 1.2 核心业务域
+- 栅格数据处理 (Raster Data Processing)
+- 矢量数据处理 (Vector Data Processing)  
+- 空间数据查询和分析
+- GIS数据导入导出
+- **Cd预测模型分析 (Cd Prediction Analysis)** ✨新增
+  - 作物Cd含量预测
+  - 有效态Cd含量预测
+  - 预测结果可视化
+
+## 2. 项目架构
+
+### 2.1 目录结构规范
+```
+AcidMap/
+├── app/                    # 应用核心代码
+│   ├── api/               # API路由层
+│   │   ├── raster.py      # 栅格数据接口
+│   │   ├── vector.py      # 矢量数据接口
+│   │   └── cd_prediction.py # Cd预测模型接口 ✨新增
+│   ├── services/          # 业务逻辑层
+│   │   ├── raster_service.py
+│   │   ├── vector_service.py
+│   │   └── cd_prediction_service.py # Cd预测业务服务 ✨新增
+│   ├── models/            # 数据模型层
+│   │   ├── base.py        # 基础模型
+│   │   ├── orm_models.py  # ORM模型
+│   │   ├── raster.py      # 栅格数据模型
+│   │   ├── vector.py      # 矢量数据模型
+│   │   └── county.py      # 县域数据模型
+│   ├── utils/             # 工具函数
+│   │   ├── file_validators.py
+│   │   └── cd_prediction_wrapper.py # Cd预测系统包装器 ✨新增
+│   ├── config/            # 配置管理 ✨新增
+│   │   ├── __init__.py
+│   │   └── cd_prediction_config.py # Cd预测配置管理
+│   ├── static/            # 静态文件 ✨新增
+│   │   └── cd_predictions/ # Cd预测输出文件
+│   │       ├── figures/    # 地图和直方图
+│   │       ├── raster/     # 栅格文件
+│   │       ├── reports/    # 报告文件
+│   │       └── data/       # 数据文件
+│   ├── logs/              # 日志文件
+│   ├── scripts/           # 脚本文件
+│   ├── database.py        # 数据库配置
+│   └── main.py           # FastAPI应用入口
+├── Cd_Prediction_Integrated_System/ # Cd预测系统 ✨新增
+│   ├── models/            # 预测模型
+│   │   ├── crop_cd_model/
+│   │   └── effective_cd_model/
+│   ├── analysis/          # 数据分析模块
+│   ├── utils/             # 工具函数
+│   ├── data/              # 数据文件
+│   ├── output/            # 输出文件
+│   ├── main.py           # 主执行脚本
+│   └── config.py         # 配置文件
+├── data/                  # 数据文件
+│   └── counties.geojson   # 县域地理数据
+├── migrations/            # 数据库迁移文件
+├── scripts/               # 项目脚本
+├── ssl/                   # SSL证书文件
+├── config.env            # 环境配置
+├── environment.yml       # Conda环境依赖
+├── soilgd.sql           # 数据库备份文件
+├── db_migrate.py        # 数据库迁移脚本
+├── reset_db.py          # 数据库重置脚本
+├── test_cd_integration.py # Cd集成测试脚本 ✨新增
+├── INTEGRATION_GUIDE.md  # 集成使用指南 ✨新增
+├── CD_INTEGRATION_SUMMARY.md # 集成总结文档 ✨新增
+└── main.py              # 应用启动入口
+```
+
+### 2.2 分层架构说明
+1. **API层 (app/api/)**: 处理HTTP请求，参数验证，调用Service层
+2. **Service层 (app/services/)**: 业务逻辑处理，数据转换，调用Model层
+3. **Model层 (app/models/)**: 数据模型定义，数据库操作
+4. **Utils层 (app/utils/)**: 通用工具函数，验证器等
+5. **Config层 (app/config/)**: 配置管理，环境设置 ✨新增
+
+## 3. 开发规范
+
+### 3.1 代码规范
+
+#### 3.1.1 注释规范
+**必须使用JSDoc风格注释**:
+
+```python
+def process_raster_data(file_path: str, options: dict) -> dict:
+    """
+    处理栅格数据文件
+    
+    @param {str} file_path - 栅格文件路径
+    @param {dict} options - 处理选项配置
+    @returns {dict} 处理结果，包含状态和数据信息
+    @throws {ValueError} 当文件路径无效时抛出
+    @example
+    >>> result = process_raster_data('/path/to/file.tif', {'format': 'GeoTIFF'})
+    >>> print(result['status'])
+    """
+    pass
+```
+
+#### 3.1.2 命名规范
+- **文件名**: 使用下划线分隔 (snake_case)
+- **类名**: 使用帕斯卡命名 (PascalCase)
+- **函数名**: 使用下划线分隔 (snake_case)
+- **变量名**: 使用下划线分隔 (snake_case)
+- **常量名**: 全大写下划线分隔 (UPPER_SNAKE_CASE)
+
+#### 3.1.3 导入规范
+```python
+# 标准库导入
+import os
+import logging
+from typing import List, Dict, Optional
+
+# 第三方库导入
+from fastapi import FastAPI, HTTPException
+from sqlalchemy import create_engine
+from geoalchemy2 import Geometry
+
+# 本地模块导入
+from .models import Base
+from .services import RasterService
+```
+
+### 3.2 API设计规范
+
+#### 3.2.1 路由组织
+- 栅格数据: `/api/raster/*`
+- 矢量数据: `/api/vector/*`
+- **Cd预测分析: `/api/cd-prediction/*`** ✨新增
+- 使用RESTful设计原则
+
+#### 3.2.2 HTTP状态码
+- 200: 成功
+- 201: 创建成功
+- 400: 客户端错误
+- 404: 资源未找到
+- 500: 服务器错误
+
+#### 3.2.3 响应格式
+```python
+{
+    "success": true,
+    "message": "操作成功",
+    "data": {...},
+    "error": null
+}
+```
+
+### 3.3 数据库规范
+
+#### 3.3.1 数据库配置
+- 数据库: PostgreSQL 12+
+- 空间扩展: PostGIS 3.0+
+- 连接池: SQLAlchemy连接池管理
+- 环境配置: config.env文件
+
+#### 3.3.2 模型定义
+- 继承Base类
+- 使用GeoAlchemy2处理空间数据
+- 添加适当的索引和约束
+
+#### 3.3.3 迁移管理
+- 使用Alembic进行数据库迁移
+- 保持迁移文件版本控制
+- 提供数据库重置脚本
+
+## 4. 核心组件说明
+
+### 4.1 数据库层 (database.py)
+```python
+# 核心配置项
+- SQLALCHEMY_DATABASE_URL: 数据库连接字符串
+- SessionLocal: 会话工厂
+- get_db(): 依赖注入函数
+- execute_sql(): 原生SQL执行
+```
+
+### 4.2 API路由层
+- **栅格API (api/raster.py)**: 处理栅格数据的CRUD操作
+- **矢量API (api/vector.py)**: 处理矢量数据的CRUD操作
+- **Cd预测API (api/cd_prediction.py)**: 处理Cd预测模型的调用和结果获取 ✨新增
+
+### 4.3 业务服务层
+- **RasterService**: 栅格数据业务逻辑
+- **VectorService**: 矢量数据业务逻辑
+- **CdPredictionService**: Cd预测分析业务逻辑 ✨新增
+
+### 4.4 数据模型层
+- **ORM模型**: 数据库表映射
+- **Pydantic模型**: API输入输出验证
+
+### 4.5 Cd预测系统组件 ✨新增
+- **CdPredictionWrapper**: Cd预测系统包装器
+- **CdPredictionConfig**: Cd预测配置管理
+- **作物Cd预测模型**: 基于神经网络的作物Cd含量预测
+- **有效态Cd预测模型**: 基于神经网络的有效态Cd含量预测
+
+## 5. 开发流程
+
+### 5.1 新功能开发流程
+1. **需求分析**: 确定功能边界和技术方案
+2. **数据模型设计**: 设计相关的数据库表和ORM模型
+3. **Service层开发**: 实现业务逻辑
+4. **API层开发**: 实现HTTP接口
+5. **测试**: 单元测试和集成测试
+6. **文档更新**: 更新API文档和代码注释
+7. **规则文件更新**: 更新PROJECT_RULES.md中的相关规范 ✨新增
+
+### 5.2 数据库变更流程
+1. 修改ORM模型
+2. 生成迁移文件: `alembic revision --autogenerate -m "描述"`
+3. 执行迁移: `alembic upgrade head`
+4. 更新数据库备份文件
+
+### 5.3 代码提交规范
+- feat: 新功能
+- fix: 修复bug
+- docs: 文档更新
+- refactor: 代码重构
+- test: 测试相关
+- **integrate: 系统集成** ✨新增
+
+### 5.4 Cd预测功能开发流程 ✨新增
+1. **模型验证**: 确保Cd预测系统完整性
+2. **包装器开发**: 创建系统调用包装器
+3. **服务层集成**: 实现异步预测服务
+4. **API接口开发**: 创建预测和下载接口
+5. **文件管理**: 实现输出文件的管理和清理
+6. **集成测试**: 运行完整的集成测试脚本
+
+## 6. 环境配置
+
+### 6.1 开发环境要求
+- Python 3.8+
+- PostgreSQL 12+ (with PostGIS)
+- **Cd预测系统依赖**: ✨新增
+  - PyTorch >= 1.9.0
+  - scikit-learn >= 1.0.0
+  - geopandas >= 0.10.0
+  - rasterio >= 1.2.0
+  - matplotlib >= 3.4.0
+  - seaborn >= 0.11.0
+  - fiona (地理数据I/O)
+  - pyogrio (高性能地理数据I/O)
+
+### 6.2 环境配置文件
+- `config.env`: 数据库连接配置
+- `environment.yml`: Conda环境依赖
+
+### 6.3 启动命令
+```bash
+# 开发环境
+uvicorn app.main:app --reload
+
+# 生产环境  
+uvicorn app.main:app --host 0.0.0.0 --port 8000
+
+# Cd集成测试
+python test_cd_integration.py
+```
+
+## 7. 安全规范
+
+### 7.1 CORS配置
+- 限制允许的源域名
+- 配置适当的HTTP方法和头部
+
+### 7.2 数据库安全
+- 使用环境变量管理敏感信息
+- 实施连接池管理
+- SQL注入防护
+
+### 7.3 文件上传安全
+- 文件类型验证
+- 文件大小限制
+- 恶意文件检测
+
+### 7.4 Cd预测系统安全 ✨新增
+- 预测任务超时控制 (5分钟)
+- 输出文件访问权限管理
+- 敏感路径信息隐藏
+- 异常信息安全过滤
+
+## 8. 性能优化
+
+### 8.1 数据库优化
+- 适当的索引设计
+- 查询优化
+- 连接池配置
+
+### 8.2 API优化
+- 响应压缩
+- 缓存策略
+- 分页处理
+
+### 8.3 Cd预测性能优化 ✨新增
+- 异步任务处理 (asyncio)
+- CPU密集型任务线程池执行
+- 预测结果文件缓存
+- 自动文件清理 (最多保留10个)
+
+## 9. 错误处理
+
+### 9.1 异常分类
+- 业务异常: 4xx HTTP状态码
+- 系统异常: 5xx HTTP状态码
+- 数据库异常: 专门的错误处理
+
+### 9.2 日志记录
+- 使用Python logging模块
+- 分级别记录日志
+- 敏感信息脱敏
+
+### 9.3 Cd预测错误处理 ✨新增
+- 预测系统完整性检查
+- 依赖包缺失检测
+- 预测超时异常处理
+- 文件生成失败回退机制
+
+## 10. 测试规范
+
+### 10.1 测试覆盖
+- 单元测试: Service层和Utils层
+- 集成测试: API端点测试
+- 数据库测试: 模型和查询测试
+
+### 10.2 测试数据
+- 使用测试数据库
+- 测试数据隔离
+- 数据清理策略
+
+### 10.3 Cd预测测试规范 ✨新增
+- 集成测试脚本: `test_cd_integration.py`
+- 配置模块测试
+- 包装器功能测试
+- API端点注册验证
+- 文件系统完整性检查
+
+## 11. 部署规范
+
+### 11.1 部署环境
+- 支持Docker容器化部署
+- SSL证书配置
+- 环境变量管理
+
+### 11.2 监控和维护
+- 应用性能监控
+- 错误日志监控
+- 数据库性能监控
+
+### 11.3 Cd预测部署特殊要求 ✨新增
+- 确保Cd_Prediction_Integrated_System目录完整
+- 验证地理空间库安装 (fiona, pyogrio)
+- 配置预测输出目录权限
+- 监控预测任务执行时间
+
+## 12. API文档规范 ✨新增
+
+### 12.1 Cd预测API端点
+
+#### 一键接口 (生成并直接返回图片)
+```
+POST /api/cd-prediction/crop-cd/generate-and-get-map
+POST /api/cd-prediction/effective-cd/generate-and-get-map
+POST /api/cd-prediction/crop-cd/generate-and-get-histogram
+POST /api/cd-prediction/effective-cd/generate-and-get-histogram
+```
+
+#### 分步式接口 (先生成后下载)
+```
+POST /api/cd-prediction/crop-cd/generate-map
+POST /api/cd-prediction/effective-cd/generate-map
+GET  /api/cd-prediction/crop-cd/download-map
+GET  /api/cd-prediction/effective-cd/download-map
+GET  /api/cd-prediction/crop-cd/download-histogram
+GET  /api/cd-prediction/effective-cd/download-histogram
+```
+
+### 12.2 一键接口特点 ✨新增
+- **即时响应**: 生成完成后直接返回图片文件
+- **简化操作**: 无需分两步操作，一次调用完成
+- **适用场景**: 前端直接显示、浏览器预览、快速下载
+- **返回格式**: 直接返回`FileResponse`图片文件
+
+### 12.3 分步式接口特点
+- **灵活控制**: 可以先生成后选择性下载
+- **状态查询**: 生成接口返回详细的统计信息
+- **批量操作**: 生成一次可多次下载不同格式
+- **返回格式**: JSON响应包含文件路径和统计信息
+
+### 12.4 响应格式规范
+
+#### 一键接口响应
+```
+Content-Type: image/jpeg
+Content-Disposition: attachment; filename="crop_cd_prediction_map.jpg"
+[图片文件二进制数据]
+```
+
+#### 分步式接口响应
+```json
+{
+    "success": true,
+    "message": "作物Cd预测地图生成成功",
+    "data": {
+        "map_path": "string",
+        "histogram_path": "string", 
+        "raster_path": "string",
+        "prediction_stats": {}
+    },
+    "error": null
+}
+```
+
+### 12.5 文档维护
+- 使用FastAPI自动生成文档 (`/docs`)
+- 维护详细的集成指南 (`INTEGRATION_GUIDE.md`)
+- 更新API变更日志
+
+---
+
+## 总结
+
+本规范文档旨在为AcidMap项目的开发提供统一的标准和指导。所有开发人员和AI助手在进行功能开发、代码维护和系统扩展时，都应严格遵循本规范，以确保代码质量、系统稳定性和开发效率。
+
+**重要提醒**: 本规范文件应与项目功能同步更新。任何新功能集成、架构调整或开发流程变更都必须及时反映到相应的规范章节中。
+
+遇到规范未覆盖的情况时，应优先考虑：
+1. 代码可读性和可维护性
+2. 系统安全性和稳定性  
+3. 性能优化和用户体验
+4. 与现有架构的一致性
+
+**最后更新**: 2025-06-01 (Cd预测功能集成) 

app/api/cd_prediction.py

@@ -0,0 +1,389 @@
+"""
+Cd预测模型API接口
+@description: 提供作物Cd和有效态Cd的预测与可视化功能
+"""
+
+from fastapi import APIRouter, HTTPException, BackgroundTasks
+from fastapi.responses import FileResponse
+from typing import Dict, Any
+import os
+import logging
+from ..services.cd_prediction_service import CdPredictionService
+
+router = APIRouter()
+
+# 设置日志
+logger = logging.getLogger(__name__)
+
+@router.post("/crop-cd/generate-and-get-map", 
+            summary="一键生成并获取作物Cd预测地图", 
+            description="执行作物Cd模型预测，生成可视化地图后直接返回图片文件")
+async def generate_and_get_crop_cd_map():
+    """
+    一键生成并获取作物Cd预测地图
+    
+    @returns {FileResponse} 作物Cd预测地图文件
+    @throws {HTTPException} 当预测过程发生错误时抛出500错误
+    @example
+    >>> response = await generate_and_get_crop_cd_map()
+    >>> # 直接返回图片文件，可在浏览器中查看或下载
+    """
+    try:
+        logger.info("开始一键生成作物Cd预测地图")
+        
+        service = CdPredictionService()
+        result = await service.generate_crop_cd_prediction()
+        
+        if not result['map_path'] or not os.path.exists(result['map_path']):
+            raise HTTPException(
+                status_code=500,
+                detail="地图文件生成失败"
+            )
+        
+        logger.info(f"作物Cd预测地图生成成功，直接返回文件: {result['map_path']}")
+        
+        return FileResponse(
+            path=result['map_path'],
+            filename="crop_cd_prediction_map.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"一键生成作物Cd预测地图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"一键生成作物Cd预测地图失败: {str(e)}"
+        )
+
+@router.post("/effective-cd/generate-and-get-map", 
+            summary="一键生成并获取有效态Cd预测地图", 
+            description="执行有效态Cd模型预测，生成可视化地图后直接返回图片文件")
+async def generate_and_get_effective_cd_map():
+    """
+    一键生成并获取有效态Cd预测地图
+    
+    @returns {FileResponse} 有效态Cd预测地图文件
+    @throws {HTTPException} 当预测过程发生错误时抛出500错误
+    @example
+    >>> response = await generate_and_get_effective_cd_map()
+    >>> # 直接返回图片文件，可在浏览器中查看或下载
+    """
+    try:
+        logger.info("开始一键生成有效态Cd预测地图")
+        
+        service = CdPredictionService()
+        result = await service.generate_effective_cd_prediction()
+        
+        if not result['map_path'] or not os.path.exists(result['map_path']):
+            raise HTTPException(
+                status_code=500,
+                detail="地图文件生成失败"
+            )
+        
+        logger.info(f"有效态Cd预测地图生成成功，直接返回文件: {result['map_path']}")
+        
+        return FileResponse(
+            path=result['map_path'],
+            filename="effective_cd_prediction_map.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"一键生成有效态Cd预测地图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"一键生成有效态Cd预测地图失败: {str(e)}"
+        )
+
+@router.post("/crop-cd/generate-and-get-histogram", 
+            summary="一键生成并获取作物Cd预测直方图", 
+            description="执行作物Cd模型预测，生成预测值分布直方图后直接返回图片文件")
+async def generate_and_get_crop_cd_histogram():
+    """
+    一键生成并获取作物Cd预测直方图
+    
+    @returns {FileResponse} 作物Cd预测直方图文件
+    @throws {HTTPException} 当预测过程发生错误时抛出500错误
+    @example
+    >>> response = await generate_and_get_crop_cd_histogram()
+    >>> # 直接返回直方图文件，可在浏览器中查看或下载
+    """
+    try:
+        logger.info("开始一键生成作物Cd预测直方图")
+        
+        service = CdPredictionService()
+        result = await service.generate_crop_cd_prediction()
+        
+        if not result['histogram_path'] or not os.path.exists(result['histogram_path']):
+            raise HTTPException(
+                status_code=500,
+                detail="直方图文件生成失败"
+            )
+        
+        logger.info(f"作物Cd预测直方图生成成功，直接返回文件: {result['histogram_path']}")
+        
+        return FileResponse(
+            path=result['histogram_path'],
+            filename="crop_cd_prediction_histogram.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"一键生成作物Cd预测直方图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"一键生成作物Cd预测直方图失败: {str(e)}"
+        )
+
+@router.post("/effective-cd/generate-and-get-histogram", 
+            summary="一键生成并获取有效态Cd预测直方图", 
+            description="执行有效态Cd模型预测，生成预测值分布直方图后直接返回图片文件")
+async def generate_and_get_effective_cd_histogram():
+    """
+    一键生成并获取有效态Cd预测直方图
+    
+    @returns {FileResponse} 有效态Cd预测直方图文件
+    @throws {HTTPException} 当预测过程发生错误时抛出500错误
+    @example
+    >>> response = await generate_and_get_effective_cd_histogram()
+    >>> # 直接返回直方图文件，可在浏览器中查看或下载
+    """
+    try:
+        logger.info("开始一键生成有效态Cd预测直方图")
+        
+        service = CdPredictionService()
+        result = await service.generate_effective_cd_prediction()
+        
+        if not result['histogram_path'] or not os.path.exists(result['histogram_path']):
+            raise HTTPException(
+                status_code=500,
+                detail="直方图文件生成失败"
+            )
+        
+        logger.info(f"有效态Cd预测直方图生成成功，直接返回文件: {result['histogram_path']}")
+        
+        return FileResponse(
+            path=result['histogram_path'],
+            filename="effective_cd_prediction_histogram.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"一键生成有效态Cd预测直方图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"一键生成有效态Cd预测直方图失败: {str(e)}"
+        )
+
+# 分步式接口 - 先生成后下载
+@router.post("/crop-cd/generate-map", 
+            summary="生成作物Cd预测地图", 
+            description="执行作物Cd模型预测并生成可视化地图")
+async def generate_crop_cd_map() -> Dict[str, Any]:
+    """
+    生成作物Cd预测地图
+    
+    @returns {Dict[str, Any]} 包含地图文件路径和预测统计信息的响应
+    @throws {HTTPException} 当预测过程发生错误时抛出500错误
+    @example
+    >>> response = await generate_crop_cd_map()
+    >>> print(response['data']['map_path'])
+    """
+    try:
+        logger.info("开始生成作物Cd预测地图")
+        
+        service = CdPredictionService()
+        result = await service.generate_crop_cd_prediction()
+        
+        logger.info(f"作物Cd预测地图生成成功: {result['map_path']}")
+        
+        return {
+            "success": True,
+            "message": "作物Cd预测地图生成成功",
+            "data": {
+                "map_path": result['map_path'],
+                "histogram_path": result['histogram_path'],
+                "raster_path": result['raster_path'],
+                "prediction_stats": result.get('stats', {})
+            },
+            "error": None
+        }
+        
+    except Exception as e:
+        logger.error(f"生成作物Cd预测地图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"生成作物Cd预测地图失败: {str(e)}"
+        )
+
+@router.post("/effective-cd/generate-map", 
+            summary="生成有效态Cd预测地图", 
+            description="执行有效态Cd模型预测并生成可视化地图")
+async def generate_effective_cd_map() -> Dict[str, Any]:
+    """
+    生成有效态Cd预测地图
+    
+    @returns {Dict[str, Any]} 包含地图文件路径和预测统计信息的响应
+    @throws {HTTPException} 当预测过程发生错误时抛出500错误
+    @example
+    >>> response = await generate_effective_cd_map()
+    >>> print(response['data']['map_path'])
+    """
+    try:
+        logger.info("开始生成有效态Cd预测地图")
+        
+        service = CdPredictionService()
+        result = await service.generate_effective_cd_prediction()
+        
+        logger.info(f"有效态Cd预测地图生成成功: {result['map_path']}")
+        
+        return {
+            "success": True,
+            "message": "有效态Cd预测地图生成成功",
+            "data": {
+                "map_path": result['map_path'],
+                "histogram_path": result['histogram_path'],
+                "raster_path": result['raster_path'],
+                "prediction_stats": result.get('stats', {})
+            },
+            "error": None
+        }
+        
+    except Exception as e:
+        logger.error(f"生成有效态Cd预测地图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"生成有效态Cd预测地图失败: {str(e)}"
+        )
+
+@router.get("/crop-cd/download-map", 
+           summary="下载作物Cd预测地图", 
+           description="下载最新生成的作物Cd预测地图文件")
+async def download_crop_cd_map():
+    """
+    下载作物Cd预测地图文件
+    
+    @returns {FileResponse} 作物Cd预测地图文件
+    @throws {HTTPException} 当地图文件不存在时抛出404错误
+    """
+    try:
+        service = CdPredictionService()
+        map_path = service.get_latest_crop_cd_map()
+        
+        if not map_path or not os.path.exists(map_path):
+            raise HTTPException(
+                status_code=404, 
+                detail="作物Cd预测地图文件不存在，请先生成地图"
+            )
+        
+        return FileResponse(
+            path=map_path,
+            filename="crop_cd_prediction_map.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"下载作物Cd预测地图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"下载作物Cd预测地图失败: {str(e)}"
+        )
+
+@router.get("/effective-cd/download-map", 
+           summary="下载有效态Cd预测地图", 
+           description="下载最新生成的有效态Cd预测地图文件")
+async def download_effective_cd_map():
+    """
+    下载有效态Cd预测地图文件
+    
+    @returns {FileResponse} 有效态Cd预测地图文件
+    @throws {HTTPException} 当地图文件不存在时抛出404错误
+    """
+    try:
+        service = CdPredictionService()
+        map_path = service.get_latest_effective_cd_map()
+        
+        if not map_path or not os.path.exists(map_path):
+            raise HTTPException(
+                status_code=404, 
+                detail="有效态Cd预测地图文件不存在，请先生成地图"
+            )
+        
+        return FileResponse(
+            path=map_path,
+            filename="effective_cd_prediction_map.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"下载有效态Cd预测地图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"下载有效态Cd预测地图失败: {str(e)}"
+        )
+
+@router.get("/crop-cd/download-histogram", 
+           summary="下载作物Cd预测直方图", 
+           description="下载最新生成的作物Cd预测值分布直方图")
+async def download_crop_cd_histogram():
+    """
+    下载作物Cd预测直方图文件
+    
+    @returns {FileResponse} 作物Cd预测直方图文件
+    @throws {HTTPException} 当直方图文件不存在时抛出404错误
+    """
+    try:
+        service = CdPredictionService()
+        histogram_path = service.get_latest_crop_cd_histogram()
+        
+        if not histogram_path or not os.path.exists(histogram_path):
+            raise HTTPException(
+                status_code=404, 
+                detail="作物Cd预测直方图文件不存在，请先生成图表"
+            )
+        
+        return FileResponse(
+            path=histogram_path,
+            filename="crop_cd_prediction_histogram.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"下载作物Cd预测直方图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"下载作物Cd预测直方图失败: {str(e)}"
+        )
+
+@router.get("/effective-cd/download-histogram", 
+           summary="下载有效态Cd预测直方图", 
+           description="下载最新生成的有效态Cd预测值分布直方图")
+async def download_effective_cd_histogram():
+    """
+    下载有效态Cd预测直方图文件
+    
+    @returns {FileResponse} 有效态Cd预测直方图文件
+    @throws {HTTPException} 当直方图文件不存在时抛出404错误
+    """
+    try:
+        service = CdPredictionService()
+        histogram_path = service.get_latest_effective_cd_histogram()
+        
+        if not histogram_path or not os.path.exists(histogram_path):
+            raise HTTPException(
+                status_code=404, 
+                detail="有效态Cd预测直方图文件不存在，请先生成图表"
+            )
+        
+        return FileResponse(
+            path=histogram_path,
+            filename="effective_cd_prediction_histogram.jpg",
+            media_type="image/jpeg"
+        )
+        
+    except Exception as e:
+        logger.error(f"下载有效态Cd预测直方图失败: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"下载有效态Cd预测直方图失败: {str(e)}"
+        ) 

app/config/__init__.py

@@ -0,0 +1,9 @@
+"""
+配置模块
+@description: AcidMap 系统配置相关模块
+@author: AcidMap Team
+"""
+
+from .cd_prediction_config import cd_config, CdPredictionConfig
+
+__all__ = ['cd_config', 'CdPredictionConfig'] 

app/config/cd_prediction_config.py

@@ -0,0 +1,115 @@
+"""
+Cd预测集成配置文件
+@description: 管理Cd预测功能集成到AcidMap系统的相关配置
+@author: AcidMap Team
+@version: 1.0.0
+"""
+
+import os
+from typing import Dict, Any
+
+class CdPredictionConfig:
+    """
+    Cd预测配置类
+    
+    @description: 管理Cd预测系统在AcidMap中的配置信息
+    @example
+    >>> config = CdPredictionConfig()
+    >>> print(config.get_cd_system_path())
+    """
+    
+    def __init__(self):
+        """
+        初始化配置
+        
+        @description: 设置基本路径和配置信息
+        """
+        # 获取项目根目录
+        self.project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+        
+        # Cd预测系统路径
+        self.cd_system_path = os.path.join(self.project_root, "Cd_Prediction_Integrated_System")
+        
+        # 输出目录配置
+        self.output_base_dir = os.path.join(self.project_root, "app", "static", "cd_predictions")
+        
+        # 子目录配置
+        self.directories = {
+            "figures": os.path.join(self.output_base_dir, "figures"),
+            "raster": os.path.join(self.output_base_dir, "raster"),
+            "reports": os.path.join(self.output_base_dir, "reports"),
+            "data": os.path.join(self.output_base_dir, "data")
+        }
+        
+        # API响应配置
+        self.api_config = {
+            "max_prediction_files": 10,  # 最多保留的预测文件数量
+            "file_cleanup_days": 7,      # 文件清理天数
+            "default_image_format": "jpg",
+            "supported_formats": ["jpg", "png", "tiff"]
+        }
+        
+        # 确保目录存在
+        self._ensure_directories()
+    
+    def _ensure_directories(self):
+        """
+        确保所有必要的目录存在
+        
+        @description: 创建输出目录结构
+        """
+        # 创建基础输出目录
+        os.makedirs(self.output_base_dir, exist_ok=True)
+        
+        # 创建子目录
+        for directory in self.directories.values():
+            os.makedirs(directory, exist_ok=True)
+    
+    def get_cd_system_path(self) -> str:
+        """
+        获取Cd预测系统路径
+        
+        @returns {str} Cd预测系统的绝对路径
+        """
+        return self.cd_system_path
+    
+    def get_output_dir(self, dir_type: str = "base") -> str:
+        """
+        获取输出目录路径
+        
+        @param {str} dir_type - 目录类型 (base/figures/raster/reports/data)
+        @returns {str} 指定类型的输出目录路径
+        @throws {KeyError} 当目录类型不存在时抛出
+        """
+        if dir_type == "base":
+            return self.output_base_dir
+        
+        if dir_type not in self.directories:
+            raise KeyError(f"不支持的目录类型: {dir_type}")
+        
+        return self.directories[dir_type]
+    
+    def get_api_config(self) -> Dict[str, Any]:
+        """
+        获取API配置信息
+        
+        @returns {Dict[str, Any]} API相关配置
+        """
+        return self.api_config.copy()
+    
+    def get_full_config(self) -> Dict[str, Any]:
+        """
+        获取完整配置信息
+        
+        @returns {Dict[str, Any]} 所有配置信息
+        """
+        return {
+            "project_root": self.project_root,
+            "cd_system_path": self.cd_system_path,
+            "output_base_dir": self.output_base_dir,
+            "directories": self.directories.copy(),
+            "api_config": self.api_config.copy()
+        }
+
+# 全局配置实例
+cd_config = CdPredictionConfig() 

app/main.py

@@ -1,5 +1,5 @@
 from fastapi import FastAPI
-from .api import vector, raster
+from .api import vector, raster, cd_prediction
 from .database import engine, Base
 from fastapi.middleware.cors import CORSMiddleware
 
@@ -18,6 +18,10 @@ app = FastAPI(
         {
             "name": "raster",
             "description": "栅格数据相关接口",
+        },
+        {
+            "name": "cd-prediction",
+            "description": "Cd预测模型相关接口",
         }
     ]
 )
@@ -36,6 +40,7 @@ app.add_middleware(
 # 注册路由
 app.include_router(vector.router, prefix="/api/vector", tags=["vector"])
 app.include_router(raster.router, prefix="/api/raster", tags=["raster"])
+app.include_router(cd_prediction.router, prefix="/api/cd-prediction", tags=["cd-prediction"])
 
 @app.get("/")
 async def root():

app/services/cd_prediction_service.py

@@ -0,0 +1,346 @@
+"""
+Cd预测服务类
+@description: 封装Cd预测模型的业务逻辑，提供作物Cd和有效态Cd的预测功能
+@author: AcidMap Team
+@version: 1.0.0
+"""
+
+import os
+import logging
+import asyncio
+from datetime import datetime
+from typing import Dict, Any, Optional
+import glob
+import shutil
+
+from ..config.cd_prediction_config import cd_config
+from ..utils.cd_prediction_wrapper import CdPredictionWrapper
+
+class CdPredictionService:
+    """
+    Cd预测服务类
+    
+    @description: 提供作物Cd和有效态Cd模型的预测与可视化功能
+    @example
+    >>> service = CdPredictionService()
+    >>> result = await service.generate_crop_cd_prediction()
+    """
+    
+    def __init__(self):
+        """
+        初始化Cd预测服务
+        
+        @description: 设置Cd预测系统的路径和配置
+        """
+        # 设置日志
+        self.logger = logging.getLogger(__name__)
+        
+        # 获取配置
+        self.config = cd_config
+        
+        # 初始化包装器
+        cd_system_path = self.config.get_cd_system_path()
+        self.wrapper = CdPredictionWrapper(cd_system_path)
+        
+        # 输出目录
+        self.output_figures_dir = self.config.get_output_dir("figures")
+        self.output_raster_dir = self.config.get_output_dir("raster")
+        
+        self.logger.info("Cd预测服务初始化完成")
+    
+    async def generate_crop_cd_prediction(self) -> Dict[str, Any]:
+        """
+        生成作物Cd预测结果和可视化
+        
+        @returns {Dict[str, Any]} 包含地图文件路径、直方图路径等信息
+        @throws {Exception} 当预测过程发生错误时抛出
+        @example
+        >>> service = CdPredictionService()
+        >>> result = await service.generate_crop_cd_prediction()
+        >>> print(result['map_path'])
+        """
+        try:
+            self.logger.info("开始作物Cd模型预测流程")
+            
+            # 在线程池中运行CPU密集型任务
+            loop = asyncio.get_event_loop()
+            result = await loop.run_in_executor(
+                None, 
+                self._run_crop_cd_prediction
+            )
+            
+            return result
+            
+        except Exception as e:
+            self.logger.error(f"作物Cd预测流程失败: {str(e)}")
+            raise
+    
+    async def generate_effective_cd_prediction(self) -> Dict[str, Any]:
+        """
+        生成有效态Cd预测结果和可视化
+        
+        @returns {Dict[str, Any]} 包含地图文件路径、直方图路径等信息
+        @throws {Exception} 当预测过程发生错误时抛出
+        @example
+        >>> service = CdPredictionService()
+        >>> result = await service.generate_effective_cd_prediction()
+        >>> print(result['map_path'])
+        """
+        try:
+            self.logger.info("开始有效态Cd模型预测流程")
+            
+            # 在线程池中运行CPU密集型任务
+            loop = asyncio.get_event_loop()
+            result = await loop.run_in_executor(
+                None, 
+                self._run_effective_cd_prediction
+            )
+            
+            return result
+            
+        except Exception as e:
+            self.logger.error(f"有效态Cd预测流程失败: {str(e)}")
+            raise
+    
+    def _run_crop_cd_prediction(self) -> Dict[str, Any]:
+        """
+        执行作物Cd预测的同步逻辑
+        
+        @returns {Dict[str, Any]} 预测结果信息
+        @throws {Exception} 当预测过程发生错误时抛出
+        """
+        try:
+            # 运行作物Cd预测
+            self.logger.info("执行作物Cd预测")
+            prediction_result = self.wrapper.run_prediction_script("crop")
+            
+            # 获取输出文件
+            latest_outputs = self.wrapper.get_latest_outputs("all")
+            
+            # 复制文件到API输出目录
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            copied_files = self._copy_output_files(latest_outputs, "crop_cd", timestamp)
+            
+            # 清理旧文件
+            self._cleanup_old_files("crop_cd")
+            
+            return {
+                'map_path': copied_files.get('map_path'),
+                'histogram_path': copied_files.get('histogram_path'),
+                'raster_path': copied_files.get('raster_path'),
+                'model_type': 'crop_cd',
+                'timestamp': timestamp,
+                'stats': self._get_file_stats(copied_files.get('map_path'))
+            }
+            
+        except Exception as e:
+            self.logger.error(f"作物Cd预测执行失败: {str(e)}")
+            raise
+    
+    def _run_effective_cd_prediction(self) -> Dict[str, Any]:
+        """
+        执行有效态Cd预测的同步逻辑
+        
+        @returns {Dict[str, Any]} 预测结果信息
+        @throws {Exception} 当预测过程发生错误时抛出
+        """
+        try:
+            # 运行有效态Cd预测
+            self.logger.info("执行有效态Cd预测")
+            prediction_result = self.wrapper.run_prediction_script("effective")
+            
+            # 获取输出文件
+            latest_outputs = self.wrapper.get_latest_outputs("all")
+            
+            # 复制文件到API输出目录
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            copied_files = self._copy_output_files(latest_outputs, "effective_cd", timestamp)
+            
+            # 清理旧文件
+            self._cleanup_old_files("effective_cd")
+            
+            return {
+                'map_path': copied_files.get('map_path'),
+                'histogram_path': copied_files.get('histogram_path'),
+                'raster_path': copied_files.get('raster_path'),
+                'model_type': 'effective_cd',
+                'timestamp': timestamp,
+                'stats': self._get_file_stats(copied_files.get('map_path'))
+            }
+            
+        except Exception as e:
+            self.logger.error(f"有效态Cd预测执行失败: {str(e)}")
+            raise
+    
+    def _copy_output_files(self, latest_outputs: Dict[str, Optional[str]], 
+                          model_type: str, timestamp: str) -> Dict[str, Optional[str]]:
+        """
+        复制输出文件到API目录
+        
+        @param {Dict[str, Optional[str]]} latest_outputs - 最新输出文件路径
+        @param {str} model_type - 模型类型
+        @param {str} timestamp - 时间戳
+        @returns {Dict[str, Optional[str]]} 复制后的文件路径
+        """
+        copied_files = {}
+        
+        try:
+            # 复制地图文件
+            if latest_outputs.get('latest_map'):
+                src_map = latest_outputs['latest_map']
+                dst_map = os.path.join(
+                    self.output_figures_dir, 
+                    f"{model_type}_prediction_map_{timestamp}.jpg"
+                )
+                shutil.copy2(src_map, dst_map)
+                copied_files['map_path'] = dst_map
+                self.logger.info(f"地图文件已复制到: {dst_map}")
+            
+            # 复制直方图文件
+            if latest_outputs.get('latest_histogram'):
+                src_histogram = latest_outputs['latest_histogram']
+                dst_histogram = os.path.join(
+                    self.output_figures_dir, 
+                    f"{model_type}_prediction_histogram_{timestamp}.jpg"
+                )
+                shutil.copy2(src_histogram, dst_histogram)
+                copied_files['histogram_path'] = dst_histogram
+                self.logger.info(f"直方图文件已复制到: {dst_histogram}")
+            
+            # 复制栅格文件
+            if latest_outputs.get('latest_raster'):
+                src_raster = latest_outputs['latest_raster']
+                dst_raster = os.path.join(
+                    self.output_raster_dir, 
+                    f"{model_type}_prediction_raster_{timestamp}.tif"
+                )
+                shutil.copy2(src_raster, dst_raster)
+                copied_files['raster_path'] = dst_raster
+                self.logger.info(f"栅格文件已复制到: {dst_raster}")
+            
+        except Exception as e:
+            self.logger.error(f"复制输出文件失败: {str(e)}")
+            
+        return copied_files
+    
+    def _cleanup_old_files(self, model_type: str):
+        """
+        清理旧的预测文件
+        
+        @param {str} model_type - 模型类型
+        """
+        try:
+            max_files = self.config.get_api_config().get("max_prediction_files", 10)
+            
+            # 清理地图文件
+            map_pattern = os.path.join(self.output_figures_dir, f"{model_type}_prediction_map_*.jpg")
+            self._cleanup_files_by_pattern(map_pattern, max_files)
+            
+            # 清理直方图文件
+            histogram_pattern = os.path.join(self.output_figures_dir, f"{model_type}_prediction_histogram_*.jpg")
+            self._cleanup_files_by_pattern(histogram_pattern, max_files)
+            
+            # 清理栅格文件
+            raster_pattern = os.path.join(self.output_raster_dir, f"{model_type}_prediction_raster_*.tif")
+            self._cleanup_files_by_pattern(raster_pattern, max_files)
+            
+        except Exception as e:
+            self.logger.warning(f"清理旧文件失败: {str(e)}")
+    
+    def _cleanup_files_by_pattern(self, pattern: str, max_files: int):
+        """
+        按模式清理文件
+        
+        @param {str} pattern - 文件模式
+        @param {int} max_files - 最大保留文件数
+        """
+        try:
+            files = glob.glob(pattern)
+            if len(files) > max_files:
+                # 按修改时间排序，删除最旧的文件
+                files.sort(key=os.path.getmtime)
+                for file_to_delete in files[:-max_files]:
+                    os.remove(file_to_delete)
+                    self.logger.info(f"已删除旧文件: {file_to_delete}")
+        except Exception as e:
+            self.logger.warning(f"清理文件失败 {pattern}: {str(e)}")
+    
+    def _get_file_stats(self, file_path: Optional[str]) -> Dict[str, Any]:
+        """
+        获取文件统计信息
+        
+        @param {Optional[str]} file_path - 文件路径
+        @returns {Dict[str, Any]} 文件统计信息
+        """
+        if not file_path or not os.path.exists(file_path):
+            return {}
+        
+        try:
+            stat = os.stat(file_path)
+            return {
+                'file_size': stat.st_size,
+                'created_time': datetime.fromtimestamp(stat.st_ctime).isoformat(),
+                'modified_time': datetime.fromtimestamp(stat.st_mtime).isoformat()
+            }
+        except Exception:
+            return {}
+    
+    def get_latest_crop_cd_map(self) -> Optional[str]:
+        """
+        获取最新的作物Cd预测地图文件路径
+        
+        @returns {Optional[str]} 最新地图文件路径，如果不存在则返回None
+        """
+        try:
+            pattern = os.path.join(self.output_figures_dir, "crop_cd_prediction_map_*.jpg")
+            files = glob.glob(pattern)
+            if files:
+                return max(files, key=os.path.getctime)
+            return None
+        except Exception:
+            return None
+    
+    def get_latest_effective_cd_map(self) -> Optional[str]:
+        """
+        获取最新的有效态Cd预测地图文件路径
+        
+        @returns {Optional[str]} 最新地图文件路径，如果不存在则返回None
+        """
+        try:
+            pattern = os.path.join(self.output_figures_dir, "effective_cd_prediction_map_*.jpg")
+            files = glob.glob(pattern)
+            if files:
+                return max(files, key=os.path.getctime)
+            return None
+        except Exception:
+            return None
+    
+    def get_latest_crop_cd_histogram(self) -> Optional[str]:
+        """
+        获取最新的作物Cd预测直方图文件路径
+        
+        @returns {Optional[str]} 最新直方图文件路径，如果不存在则返回None
+        """
+        try:
+            pattern = os.path.join(self.output_figures_dir, "crop_cd_prediction_histogram_*.jpg")
+            files = glob.glob(pattern)
+            if files:
+                return max(files, key=os.path.getctime)
+            return None
+        except Exception:
+            return None
+    
+    def get_latest_effective_cd_histogram(self) -> Optional[str]:
+        """
+        获取最新的有效态Cd预测直方图文件路径
+        
+        @returns {Optional[str]} 最新直方图文件路径，如果不存在则返回None
+        """
+        try:
+            pattern = os.path.join(self.output_figures_dir, "effective_cd_prediction_histogram_*.jpg")
+            files = glob.glob(pattern)
+            if files:
+                return max(files, key=os.path.getctime)
+            return None
+        except Exception:
+            return None 

app/utils/cd_prediction_wrapper.py

@@ -0,0 +1,254 @@
+"""
+Cd预测系统包装器
+@description: 封装对原始Cd预测系统的调用，简化集成过程
+@author: AcidMap Team
+@version: 1.0.0
+"""
+
+import os
+import sys
+import logging
+import subprocess
+from typing import Dict, Any, Optional
+from datetime import datetime
+
+class CdPredictionWrapper:
+    """
+    Cd预测系统包装器类
+    
+    @description: 简化对原始Cd预测系统的调用和集成
+    @example
+    >>> wrapper = CdPredictionWrapper()
+    >>> result = wrapper.run_crop_cd_prediction()
+    """
+    
+    def __init__(self, cd_system_path: str):
+        """
+        初始化包装器
+        
+        @param {str} cd_system_path - Cd预测系统的路径
+        """
+        self.cd_system_path = cd_system_path
+        self.logger = logging.getLogger(__name__)
+        
+        # 验证Cd预测系统是否存在
+        if not os.path.exists(cd_system_path):
+            raise FileNotFoundError(f"Cd预测系统路径不存在: {cd_system_path}")
+        
+        # 检查关键文件是否存在
+        self._validate_cd_system()
+    
+    def _validate_cd_system(self):
+        """
+        验证Cd预测系统的完整性
+        
+        @throws {FileNotFoundError} 当关键文件缺失时抛出
+        """
+        required_files = [
+            "main.py",
+            "config.py",
+            "models",
+            "analysis"
+        ]
+        
+        for file_path in required_files:
+            full_path = os.path.join(self.cd_system_path, file_path)
+            if not os.path.exists(full_path):
+                raise FileNotFoundError(f"Cd预测系统缺少必要文件: {file_path}")
+        
+        self.logger.info("Cd预测系统验证通过")
+    
+    def run_prediction_script(self, model_type: str = "both") -> Dict[str, Any]:
+        """
+        运行Cd预测脚本
+        
+        @param {str} model_type - 模型类型 ("crop", "effective", "both")
+        @returns {Dict[str, Any]} 预测结果信息
+        @throws {Exception} 当预测过程失败时抛出
+        """
+        try:
+            self.logger.info(f"开始运行Cd预测脚本，模型类型: {model_type}")
+            
+            # 切换到Cd预测系统目录
+            original_cwd = os.getcwd()
+            os.chdir(self.cd_system_path)
+            
+            try:
+                # 修改配置文件以只运行指定模型
+                self._modify_workflow_config(model_type)
+                
+                # 运行主脚本
+                result = subprocess.run(
+                    [sys.executable, "main.py"],
+                    capture_output=True,
+                    text=True,
+                    timeout=300  # 5分钟超时
+                )
+                
+                if result.returncode != 0:
+                    self.logger.error(f"Cd预测脚本执行失败: {result.stderr}")
+                    raise Exception(f"预测脚本执行失败: {result.stderr}")
+                
+                self.logger.info("Cd预测脚本执行成功")
+                
+                # 获取输出文件信息
+                output_info = self._get_output_files(model_type)
+                
+                return {
+                    "success": True,
+                    "model_type": model_type,
+                    "output_files": output_info,
+                    "stdout": result.stdout,
+                    "timestamp": datetime.now().isoformat()
+                }
+                
+            finally:
+                # 恢复原始工作目录
+                os.chdir(original_cwd)
+                
+        except subprocess.TimeoutExpired:
+            self.logger.error("Cd预测脚本执行超时")
+            raise Exception("预测脚本执行超时")
+        except Exception as e:
+            self.logger.error(f"运行Cd预测脚本失败: {str(e)}")
+            raise
+    
+    def _modify_workflow_config(self, model_type: str):
+        """
+        修改工作流配置
+        
+        @param {str} model_type - 模型类型
+        """
+        config_file = os.path.join(self.cd_system_path, "config.py")
+        
+        # 根据模型类型设置配置
+        if model_type == "crop":
+            workflow_config = {
+                "run_crop_model": True,
+                "run_effective_model": False,
+                "combine_predictions": True,
+                "generate_raster": True,
+                "create_visualization": True,
+                "create_histogram": True
+            }
+        elif model_type == "effective":
+            workflow_config = {
+                "run_crop_model": False,
+                "run_effective_model": True,
+                "combine_predictions": True,
+                "generate_raster": True,
+                "create_visualization": True,
+                "create_histogram": True
+            }
+        else:  # both
+            workflow_config = {
+                "run_crop_model": True,
+                "run_effective_model": True,
+                "combine_predictions": True,
+                "generate_raster": True,
+                "create_visualization": True,
+                "create_histogram": True
+            }
+        
+        # 读取当前配置文件
+        with open(config_file, 'r', encoding='utf-8') as f:
+            config_content = f.read()
+        
+        # 替换 WORKFLOW_CONFIG
+        import re
+        pattern = r'WORKFLOW_CONFIG\s*=\s*\{[^}]*\}'
+        replacement = f"WORKFLOW_CONFIG = {workflow_config}"
+        
+        new_content = re.sub(pattern, replacement, config_content, flags=re.MULTILINE | re.DOTALL)
+        
+        # 写回文件
+        with open(config_file, 'w', encoding='utf-8') as f:
+            f.write(new_content)
+        
+        self.logger.info(f"已更新工作流配置为模型类型: {model_type}")
+    
+    def _get_output_files(self, model_type: str) -> Dict[str, Any]:
+        """
+        获取输出文件信息
+        
+        @param {str} model_type - 模型类型
+        @returns {Dict[str, Any]} 输出文件信息
+        """
+        output_dir = os.path.join(self.cd_system_path, "output")
+        figures_dir = os.path.join(output_dir, "figures")
+        raster_dir = os.path.join(output_dir, "raster")
+        
+        output_files = {
+            "figures_dir": figures_dir,
+            "raster_dir": raster_dir,
+            "maps": [],
+            "histograms": [],
+            "rasters": []
+        }
+        
+        # 查找相关文件
+        if os.path.exists(figures_dir):
+            for file in os.listdir(figures_dir):
+                if file.endswith(('.jpg', '.png')):
+                    file_path = os.path.join(figures_dir, file)
+                    if "Prediction" in file and "results" in file:
+                        output_files["maps"].append(file_path)
+                    elif "frequency" in file.lower() or "histogram" in file.lower():
+                        output_files["histograms"].append(file_path)
+        
+        if os.path.exists(raster_dir):
+            for file in os.listdir(raster_dir):
+                if file.endswith('.tif') and file.startswith('output'):
+                    output_files["rasters"].append(os.path.join(raster_dir, file))
+        
+        return output_files
+    
+    def get_latest_outputs(self, output_type: str = "all") -> Dict[str, Optional[str]]:
+        """
+        获取最新的输出文件
+        
+        @param {str} output_type - 输出类型 ("maps", "histograms", "rasters", "all")
+        @returns {Dict[str, Optional[str]]} 最新输出文件路径
+        """
+        try:
+            output_dir = os.path.join(self.cd_system_path, "output")
+            figures_dir = os.path.join(output_dir, "figures")
+            raster_dir = os.path.join(output_dir, "raster")
+            
+            latest_files = {}
+            
+            # 获取最新地图文件
+            if output_type in ["maps", "all"]:
+                map_files = []
+                if os.path.exists(figures_dir):
+                    for file in os.listdir(figures_dir):
+                        if "Prediction" in file and "results" in file and file.endswith(('.jpg', '.png')):
+                            map_files.append(os.path.join(figures_dir, file))
+                
+                latest_files["latest_map"] = max(map_files, key=os.path.getctime) if map_files else None
+            
+            # 获取最新直方图文件
+            if output_type in ["histograms", "all"]:
+                histogram_files = []
+                if os.path.exists(figures_dir):
+                    for file in os.listdir(figures_dir):
+                        if ("frequency" in file.lower() or "histogram" in file.lower()) and file.endswith(('.jpg', '.png')):
+                            histogram_files.append(os.path.join(figures_dir, file))
+                
+                latest_files["latest_histogram"] = max(histogram_files, key=os.path.getctime) if histogram_files else None
+            
+            # 获取最新栅格文件
+            if output_type in ["rasters", "all"]:
+                raster_files = []
+                if os.path.exists(raster_dir):
+                    for file in os.listdir(raster_dir):
+                        if file.startswith('output') and file.endswith('.tif'):
+                            raster_files.append(os.path.join(raster_dir, file))
+                
+                latest_files["latest_raster"] = max(raster_files, key=os.path.getctime) if raster_files else None
+            
+            return latest_files
+            
+        except Exception as e:
+            self.logger.error(f"获取最新输出文件失败: {str(e)}")
+            return {} 

test_cd_integration.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""
+Cd预测集成测试脚本
+@description: 测试Cd预测功能是否正确集成到AcidMap系统中
+@author: AcidMap Team
+@version: 1.0.0
+"""
+
+import os
+import sys
+import asyncio
+import logging
+from datetime import datetime
+
+# 添加app目录到Python路径
+sys.path.append(os.path.join(os.path.dirname(__file__), 'app'))
+
+# 设置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+async def test_cd_prediction_integration():
+    """
+    测试Cd预测功能集成
+    
+    @description: 验证配置、服务和包装器是否正常工作
+    """
+    logger = logging.getLogger(__name__)
+    logger.info("开始测试Cd预测功能集成")
+    
+    try:
+        # 测试1: 验证配置模块
+        logger.info("测试1: 验证配置模块")
+        from app.config.cd_prediction_config import cd_config
+        
+        logger.info(f"项目根目录: {cd_config.project_root}")
+        logger.info(f"Cd系统路径: {cd_config.get_cd_system_path()}")
+        logger.info(f"输出目录: {cd_config.get_output_dir()}")
+        
+        # 检查Cd系统是否存在
+        cd_system_path = cd_config.get_cd_system_path()
+        if not os.path.exists(cd_system_path):
+            logger.error(f"Cd预测系统路径不存在: {cd_system_path}")
+            return False
+        
+        logger.info("✅ 配置模块测试通过")
+        
+        # 测试2: 验证包装器
+        logger.info("测试2: 验证包装器模块")
+        from app.utils.cd_prediction_wrapper import CdPredictionWrapper
+        
+        wrapper = CdPredictionWrapper(cd_system_path)
+        logger.info("✅ 包装器初始化成功")
+        
+        # 测试3: 验证服务类
+        logger.info("测试3: 验证服务类")
+        from app.services.cd_prediction_service import CdPredictionService
+        
+        service = CdPredictionService()
+        logger.info("✅ 服务类初始化成功")
+        
+        # 测试4: 检查输出目录
+        logger.info("测试4: 检查输出目录")
+        
+        figures_dir = cd_config.get_output_dir("figures")
+        raster_dir = cd_config.get_output_dir("raster")
+        
+        logger.info(f"图片输出目录: {figures_dir}")
+        logger.info(f"栅格输出目录: {raster_dir}")
+        
+        if os.path.exists(figures_dir) and os.path.exists(raster_dir):
+            logger.info("✅ 输出目录创建成功")
+        else:
+            logger.warning("⚠️ 输出目录可能有问题")
+        
+        # 测试5: 检查Cd系统的关键文件
+        logger.info("测试5: 检查Cd系统的关键文件")
+        
+        required_files = [
+            "main.py",
+            "config.py", 
+            "models",
+            "analysis"
+        ]
+        
+        missing_files = []
+        for file_path in required_files:
+            full_path = os.path.join(cd_system_path, file_path)
+            if not os.path.exists(full_path):
+                missing_files.append(file_path)
+        
+        if missing_files:
+            logger.error(f"❌ Cd系统缺少关键文件: {missing_files}")
+            return False
+        else:
+            logger.info("✅ Cd系统文件完整性检查通过")
+        
+        logger.info("🎉 所有测试通过！Cd预测功能集成成功！")
+        return True
+        
+    except ImportError as e:
+        logger.error(f"❌ 导入模块失败: {str(e)}")
+        return False
+    except Exception as e:
+        logger.error(f"❌ 测试过程中发生错误: {str(e)}")
+        return False
+
+def test_api_endpoints():
+    """
+    测试API端点是否正确注册
+    
+    @description: 验证FastAPI应用是否正确包含Cd预测路由
+    """
+    logger = logging.getLogger(__name__)
+    logger.info("测试API端点注册")
+    
+    try:
+        from app.main import app
+        
+        # 获取所有路由
+        routes = []
+        for route in app.routes:
+            if hasattr(route, 'path'):
+                routes.append(route.path)
+        
+        # 检查Cd预测相关路由
+        cd_routes = [route for route in routes if '/cd-prediction/' in route]
+        
+        expected_routes = [
+            '/api/cd-prediction/crop-cd/generate-map',
+            '/api/cd-prediction/effective-cd/generate-map',
+            '/api/cd-prediction/crop-cd/download-map',
+            '/api/cd-prediction/effective-cd/download-map',
+            '/api/cd-prediction/crop-cd/download-histogram',
+            '/api/cd-prediction/effective-cd/download-histogram'
+        ]
+        
+        logger.info(f"发现的Cd预测路由: {cd_routes}")
+        
+        # 检查是否所有预期路由都存在
+        found_routes = 0
+        for expected_route in expected_routes:
+            if any(expected_route in route for route in cd_routes):
+                found_routes += 1
+                logger.info(f"✅ 找到路由: {expected_route}")
+            else:
+                logger.warning(f"⚠️ 未找到路由: {expected_route}")
+        
+        if found_routes == len(expected_routes):
+            logger.info("✅ 所有API端点注册成功")
+            return True
+        else:
+            logger.warning(f"⚠️ 只找到 {found_routes}/{len(expected_routes)} 个预期路由")
+            return False
+            
+    except Exception as e:
+        logger.error(f"❌ API端点测试失败: {str(e)}")
+        return False
+
+def main():
+    """
+    主测试函数
+    
+    @description: 运行所有集成测试
+    """
+    print("=" * 60)
+    print("🔬 Cd预测功能集成测试")
+    print("=" * 60)
+    
+    # 运行异步测试
+    integration_success = asyncio.run(test_cd_prediction_integration())
+    
+    # 运行API测试
+    api_success = test_api_endpoints()
+    
+    print("\n" + "=" * 60)
+    print("📊 测试结果汇总")
+    print("=" * 60)
+    
+    if integration_success:
+        print("✅ 集成功能测试: 通过")
+    else:
+        print("❌ 集成功能测试: 失败")
+    
+    if api_success:
+        print("✅ API端点测试: 通过")
+    else:
+        print("❌ API端点测试: 失败")
+    
+    overall_success = integration_success and api_success
+    
+    if overall_success:
+        print("\n🎉 恭喜！Cd预测功能已成功集成到AcidMap系统中！")
+        print("\n📝 接下来您可以:")
+        print("   1. 启动FastAPI服务器: uvicorn app.main:app --reload")
+        print("   2. 访问API文档: http://localhost:8000/docs")
+        print("   3. 测试Cd预测接口")
+        print("   4. 参考 INTEGRATION_GUIDE.md 了解详细使用方法")
+    else:
+        print("\n❌ 集成存在问题，请检查错误信息并修复")
+        print("\n🔧 常见问题排查:")
+        print("   1. 确保 Cd_Prediction_Integrated_System 文件夹存在且完整")
+        print("   2. 检查Python依赖是否安装完整")
+        print("   3. 验证文件路径和权限")
+    
+    print("=" * 60)
+    
+    return overall_success
+
+if __name__ == "__main__":
+    success = main()
+    sys.exit(0 if success else 1)