集成耕地质量分类接口

README.md

@@ -6,6 +6,7 @@
 
 - 支持栅格数据的导入和导出
 - 支持矢量数据的导入和导出
+- **单元分类功能**: 基于空间插值算法的土壤环境质量分类
 - 使用 PostgreSQL + PostGIS 存储空间数据
 - 提供 RESTful API 接口
 - 支持空间数据查询和分析
@@ -16,14 +17,30 @@
 app/
 ├── api/            # API 路由层
 │   ├── raster.py   # 栅格数据接口
-│   └── vector.py   # 矢量数据接口
+│   ├── vector.py   # 矢量数据接口
+│   └── unit_grouping.py  # 单元分类接口
 ├── services/       # 业务逻辑层
 │   ├── raster_service.py
-│   └── vector_service.py
+│   ├── vector_service.py
+│   └── unit_grouping_service.py  # 单元分类服务
 ├── models/         # 数据模型
 ├── utils/          # 工具函数
 ├── database.py     # 数据库配置
 └── main.py         # 主程序入口
+
+docs/
+├── features/       # 功能文档
+│   └── unit-grouping/
+│       └── README.md  # 单元分类功能说明
+
+tests/
+├── integration/    # 集成测试
+│   └── test_unit_grouping.py
+
+scripts/
+├── demos/          # 演示脚本
+│   └── unit_grouping_demo.py
+└── ...
 ```
 
 ## 安装依赖
@@ -80,7 +97,17 @@ pg_restore -U postgres -d data_db soilgd.sql
 uvicorn app.main:app --reload
 ```
 
-2. 访问 API 文档：
+2. 快速演示单元分类功能：
+```bash
+python scripts/demos/unit_grouping_demo.py
+```
+
+3. 运行单元分类集成测试：
+```bash
+python tests/integration/test_unit_grouping.py
+```
+
+4. 访问 API 文档：
 - Swagger UI: http://localhost:8000/docs
 - ReDoc: http://localhost:8000/redoc
 
@@ -96,6 +123,16 @@ uvicorn app.main:app --reload
 - GET /export: 导出矢量数据
 - GET /query: 查询矢量数据
 
+### 单元分类接口 (/api/unit-grouping)
+- GET /h_xtfx: 获取所有单元的h_xtfx分类结果
+- GET /statistics: 获取统计信息
+- GET /unit/{unit_id}: 获取特定单元的h_xtfx值
+- GET /points/statistics: 获取点位统计信息
+- GET /database/summary: 获取数据库摘要信息
+- GET /units/batch: 批量获取单元信息
+
+详细说明请参考: [单元分类功能文档](docs/features/unit-grouping/README.md)
+
 ## 开发环境
 
 - Python 3.8+

app/api/unit_grouping.py

@@ -0,0 +1,402 @@
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from typing import Dict, Any, List
+import logging
+from ..database import get_db
+from ..services.unit_grouping_service import UnitGroupingService
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+@router.get("/h_xtfx", 
+           summary="获取单元h_xtfx分类结果", 
+           description="基于点位数据计算各单元的土壤环境质量类别(h_xtfx)，使用反距离加权插值算法")
+async def get_unit_h_xtfx_result(db: Session = Depends(get_db)) -> Dict[str, Any]:
+    """
+    获取单元h_xtfx分类结果
+    
+    该接口实现以下功能：
+    1. 获取单元几何数据和点位数据
+    2. 判断每个单元内包含的点位
+    3. 根据点位的h_xtfx值计算单元的h_xtfx值
+    4. 使用反距离加权插值算法进行计算
+    
+    算法逻辑：
+    - 如果单元内无严格管控类点位：
+      - 优先判断某类别占比是否≥80%，是则直接采用该类别
+      - 否则对优先保护类和安全利用类点位进行插值
+    - 如果单元内有严格管控类点位：
+      - 对所有点位（包括严格管控类）进行插值
+    
+    Returns:
+        Dict[str, Any]: 包含以下字段的响应：
+        - success: 是否成功
+        - data: 单元ID到h_xtfx值的映射
+        - statistics: 统计信息
+        - error: 错误信息（如果失败）
+    
+    Raises:
+        HTTPException: 当数据库查询失败或处理异常时抛出
+    """
+    try:
+        logger.info("开始获取单元h_xtfx分类结果")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 获取结果
+        result = service.get_unit_h_xtfx_result()
+        
+        if not result["success"]:
+            logger.error(f"服务层处理失败: {result.get('error', 'Unknown error')}")
+            raise HTTPException(
+                status_code=500, 
+                detail=f"计算单元h_xtfx值失败: {result.get('error', 'Unknown error')}"
+            )
+        
+        logger.info(f"成功获取 {result['statistics']['total_units']} 个单元的h_xtfx结果")
+        return result
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"获取单元h_xtfx结果时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/statistics", 
+           summary="获取单元分组统计信息", 
+           description="获取单元分组的统计摘要信息")
+async def get_unit_grouping_statistics(db: Session = Depends(get_db)) -> Dict[str, Any]:
+    """
+    获取单元分组统计信息
+    
+    提供单元分组的统计摘要，包括：
+    - 总单元数
+    - 有数据的单元数
+    - 无数据的单元数
+    - 各类别的分布情况
+    
+    Returns:
+        Dict[str, Any]: 统计信息
+    """
+    try:
+        logger.info("开始获取单元分组统计信息")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 获取完整结果以提取统计信息
+        result = service.get_unit_h_xtfx_result()
+        
+        if not result["success"]:
+            raise HTTPException(
+                status_code=500, 
+                detail=f"获取统计信息失败: {result.get('error', 'Unknown error')}"
+            )
+        
+        # 返回统计信息
+        return {
+            "success": True,
+            "statistics": result["statistics"]
+        }
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"获取统计信息时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/unit/{unit_id}", 
+           summary="获取特定单元的h_xtfx值", 
+           description="获取指定单元ID的h_xtfx分类结果")
+async def get_unit_h_xtfx_by_id(unit_id: int, db: Session = Depends(get_db)) -> Dict[str, Any]:
+    """
+    获取特定单元的h_xtfx值
+    
+    Args:
+        unit_id: 单元ID
+        
+    Returns:
+        Dict[str, Any]: 包含单元h_xtfx值的响应
+    """
+    try:
+        logger.info(f"开始获取单元 {unit_id} 的h_xtfx值")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 获取所有结果
+        result = service.get_unit_h_xtfx_result()
+        
+        if not result["success"]:
+            raise HTTPException(
+                status_code=500, 
+                detail=f"获取单元数据失败: {result.get('error', 'Unknown error')}"
+            )
+        
+        # 检查指定单元是否存在
+        if unit_id not in result["data"]:
+            raise HTTPException(
+                status_code=404, 
+                detail=f"未找到单元 {unit_id}"
+            )
+        
+        return {
+            "success": True,
+            "unit_id": unit_id,
+            "h_xtfx": result["data"][unit_id]
+        }
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"获取单元 {unit_id} 的h_xtfx值时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/units/batch",
+           summary="批量获取单元信息",
+           description="使用ORM方式批量获取指定单元的基本信息")
+async def get_units_batch(
+    unit_ids: List[int] = Query(..., description="单元ID列表"),
+    db: Session = Depends(get_db)
+) -> Dict[str, Any]:
+    """
+    批量获取单元信息
+    
+    Args:
+        unit_ids: 单元ID列表
+        
+    Returns:
+        Dict[str, Any]: 包含单元信息的响应
+    """
+    try:
+        logger.info(f"开始批量获取 {len(unit_ids)} 个单元的信息")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 使用ORM方式批量获取单元信息
+        units = service.get_units_by_ids(unit_ids)
+        
+        # 转换为字典格式
+        unit_data = []
+        for unit in units:
+            unit_data.append({
+                "gid": unit.gid,
+                "BSM": unit.BSM,
+                "PXZQMC": unit.PXZQMC,
+                "CXZQMC": unit.CXZQMC,
+                "SUM_NYDTBM": float(unit.SUM_NYDTBM) if unit.SUM_NYDTBM else None,
+                "Shape_Area": float(unit.Shape_Area) if unit.Shape_Area else None
+            })
+        
+        return {
+            "success": True,
+            "total_requested": len(unit_ids),
+            "total_found": len(units),
+            "data": unit_data
+        }
+        
+    except Exception as e:
+        logger.error(f"批量获取单元信息时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/points/statistics",
+           summary="获取点位统计信息",
+           description="使用ORM方式获取点位按h_xtfx分类的统计信息")
+async def get_points_statistics(db: Session = Depends(get_db)) -> Dict[str, Any]:
+    """
+    获取点位统计信息
+    
+    返回按h_xtfx分类的点位数量统计
+    
+    Returns:
+        Dict[str, Any]: 点位统计信息
+    """
+    try:
+        logger.info("开始获取点位统计信息")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 使用ORM方式获取统计信息
+        point_stats = service.get_point_count_by_h_xtfx()
+        
+        # 计算总数
+        total_points = sum(point_stats.values())
+        
+        # 计算百分比
+        percentage_stats = {}
+        for category, count in point_stats.items():
+            percentage_stats[category] = {
+                "count": count,
+                "percentage": round(count / total_points * 100, 2) if total_points > 0 else 0
+            }
+        
+        return {
+            "success": True,
+            "total_points": total_points,
+            "distribution": percentage_stats
+        }
+        
+    except Exception as e:
+        logger.error(f"获取点位统计信息时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/points/by-area",
+           summary="按区域获取点位数据",
+           description="使用ORM方式获取特定区域的点位数据")
+async def get_points_by_area(
+    area_name: str = Query(None, description="区域名称（县名称）"),
+    db: Session = Depends(get_db)
+) -> Dict[str, Any]:
+    """
+    按区域获取点位数据
+    
+    Args:
+        area_name: 区域名称（县名称），如果为空则获取所有区域
+        
+    Returns:
+        Dict[str, Any]: 点位数据
+    """
+    try:
+        logger.info(f"开始获取区域 '{area_name}' 的点位数据")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 使用ORM方式获取点位数据
+        points = service.get_points_in_area(area_name)
+        
+        # 转换为字典格式
+        point_data = []
+        for point in points:
+            point_data.append({
+                "id": point.id,
+                "dwmc": point.dwmc,
+                "xmc": point.xmc,
+                "zmc": point.zmc,
+                "cmc": point.cmc,
+                "h_xtfx": point.h_xtfx,
+                "lat": float(point.lat) if point.lat else None,
+                "lon": float(point.lon) if point.lon else None,
+                "ph": float(point.ph) if point.ph else None
+            })
+        
+        # 统计该区域的h_xtfx分布
+        h_xtfx_stats = {}
+        for point in points:
+            if point.h_xtfx:
+                h_xtfx_stats[point.h_xtfx] = h_xtfx_stats.get(point.h_xtfx, 0) + 1
+        
+        return {
+            "success": True,
+            "area_name": area_name or "全部区域",
+            "total_points": len(points),
+            "h_xtfx_distribution": h_xtfx_stats,
+            "data": point_data
+        }
+        
+    except Exception as e:
+        logger.error(f"获取区域点位数据时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/database/summary",
+           summary="获取数据库摘要信息",
+           description="获取数据库中单元和点位的摘要统计")
+async def get_database_summary(db: Session = Depends(get_db)) -> Dict[str, Any]:
+    """
+    获取数据库摘要信息
+    
+    Returns:
+        Dict[str, Any]: 数据库摘要信息
+    """
+    try:
+        logger.info("开始获取数据库摘要信息")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 使用ORM方式获取各种统计信息
+        total_units = service.get_unit_count()
+        point_stats = service.get_point_count_by_h_xtfx()
+        total_points = sum(point_stats.values())
+        
+        return {
+            "success": True,
+            "summary": {
+                "total_units": total_units,
+                "total_points": total_points,
+                "points_with_h_xtfx": total_points,
+                "h_xtfx_categories": len(point_stats),
+                "point_distribution": point_stats
+            }
+        }
+        
+    except Exception as e:
+        logger.error(f"获取数据库摘要信息时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        )
+
+@router.get("/areas/statistics",
+           summary="获取各区域统计信息",
+           description="使用ORM聚合查询获取各区域的h_xtfx分布统计")
+async def get_areas_statistics(db: Session = Depends(get_db)) -> Dict[str, Any]:
+    """
+    获取各区域统计信息
+    
+    使用ORM的高级聚合查询功能，按区域分组统计h_xtfx分布
+    
+    Returns:
+        Dict[str, Any]: 各区域的统计信息
+    """
+    try:
+        logger.info("开始获取各区域统计信息")
+        
+        # 创建服务实例
+        service = UnitGroupingService(db)
+        
+        # 使用ORM方式获取区域统计信息
+        area_stats = service.get_area_statistics()
+        
+        # 计算总体统计
+        total_areas = len(area_stats)
+        total_points_by_area = {}
+        
+        for area_name, stats in area_stats.items():
+            total_points_by_area[area_name] = sum(stats.values())
+        
+        return {
+            "success": True,
+            "total_areas": total_areas,
+            "area_statistics": area_stats,
+            "area_totals": total_points_by_area
+        }
+        
+    except Exception as e:
+        logger.error(f"获取各区域统计信息时发生异常: {str(e)}")
+        raise HTTPException(
+            status_code=500, 
+            detail=f"内部服务器错误: {str(e)}"
+        ) 

app/main.py

@@ -1,5 +1,5 @@
 from fastapi import FastAPI
-from .api import vector, raster, cd_prediction
+from .api import vector, raster, cd_prediction, unit_grouping
 from .database import engine, Base
 from fastapi.middleware.cors import CORSMiddleware
 import logging
@@ -91,6 +91,10 @@ app = FastAPI(
         {
             "name": "cd-prediction",
             "description": "Cd预测模型相关接口",
+        },
+        {
+            "name": "unit-grouping",
+            "description": "单元分组相关接口",
         }
     ]
 )
@@ -110,6 +114,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.include_router(unit_grouping.router, prefix="/api/unit-grouping", tags=["unit-grouping"])
 
 @app.get("/")
 async def root():

app/services/unit_grouping_service.py

@@ -0,0 +1,423 @@
+from sqlalchemy.orm import Session
+from sqlalchemy import func, text
+from collections import Counter
+from typing import Dict, List, Tuple, Optional
+import logging
+from ..models.orm_models import FiftyThousandSurveyDatum, UnitCeil
+
+logger = logging.getLogger(__name__)
+
+class UnitGroupingService:
+    """
+    单元分组服务
+    
+    提供基于点位数据的单元h_xtfx值计算功能
+    """
+    
+    # 定义 h_xtfx 值的映射关系（数值用于插值计算）
+    H_XTFX_MAPPING = {
+        "优先保护类": 1,
+        "安全利用类": 2,
+        "严格管控类": 3
+    }
+    
+    REVERSE_H_XTFX_MAPPING = {v: k for k, v in H_XTFX_MAPPING.items()}
+    
+    def __init__(self, db_session: Session):
+        """
+        初始化服务
+        
+        Args:
+            db_session: 数据库会话
+        """
+        self.db_session = db_session
+    
+    def calculate_unit_h_xtfx_values(self) -> Dict[int, Optional[str]]:
+        """
+        核心逻辑：使用数据库端空间查询计算单元的 h_xtfx 值
+        
+        Returns:
+            Dict[int, Optional[str]]: 单元ID到h_xtfx值的映射
+        """
+        try:
+            # 直接在数据库中进行空间查询，获取每个单元包含的点位及其h_xtfx值
+            spatial_query = text("""
+                SELECT 
+                    u.gid as unit_id,
+                    p.h_xtfx,
+                    ST_X(ST_Centroid(u.geom)) as unit_center_x,
+                    ST_Y(ST_Centroid(u.geom)) as unit_center_y,
+                    ST_X(ST_Transform(ST_SetSRID(p.geom, 4490), 4490)) as point_x,
+                    ST_Y(ST_Transform(ST_SetSRID(p.geom, 4490), 4490)) as point_y
+                FROM unit_ceil u
+                JOIN fifty_thousand_survey_data p ON ST_Contains(
+                    u.geom, 
+                    ST_Transform(ST_SetSRID(p.geom, 4490), 4490)
+                )
+                WHERE p.h_xtfx IS NOT NULL
+                ORDER BY u.gid
+            """)
+            
+            spatial_results = self.db_session.execute(spatial_query).fetchall()
+            
+            # 组织数据：单元ID -> 包含的点位列表
+            unit_points = {}
+            for result in spatial_results:
+                unit_id = result.unit_id
+                h_xtfx = result.h_xtfx
+                point_x = result.point_x
+                point_y = result.point_y
+                
+                if unit_id not in unit_points:
+                    unit_points[unit_id] = []
+                
+                unit_points[unit_id].append({
+                    'h_xtfx': h_xtfx,
+                    'x': point_x,
+                    'y': point_y
+                })
+            
+            # 计算每个单元的h_xtfx值
+            result = {}
+            
+            # 获取所有单元的ID
+            all_units = self.db_session.execute(text("SELECT gid FROM unit_ceil ORDER BY gid")).fetchall()
+            
+            for unit_row in all_units:
+                unit_id = unit_row.gid
+                points = unit_points.get(unit_id, [])
+                
+                if not points:
+                    result[unit_id] = None
+                    continue
+                
+                try:
+                    result[unit_id] = self._calculate_single_unit_h_xtfx_from_points(unit_id, points)
+                except Exception as e:
+                    logger.error(f"计算单元 {unit_id} 的h_xtfx值失败: {e}")
+                    result[unit_id] = None
+            
+            logger.info(f"计算完成，共处理 {len(result)} 个单元，其中 {sum(1 for v in result.values() if v is not None)} 个有有效结果")
+            return result
+            
+        except Exception as e:
+            logger.error(f"计算单元h_xtfx值失败: {e}")
+            return {}
+    
+    def _calculate_single_unit_h_xtfx_from_points(self, unit_id: int, points: List[Dict]) -> Optional[str]:
+        """
+        基于点位列表计算单个单元的h_xtfx值
+        
+        Args:
+            unit_id: 单元ID
+            points: 单元内的点位列表，每个点位包含 {'h_xtfx': str, 'x': float, 'y': float}
+            
+        Returns:
+            str: h_xtfx值
+        """
+        if not points:
+            return None
+        
+        h_xtfx_list = [point['h_xtfx'] for point in points]
+        has_strict_control = any(h_xtfx == "严格管控类" for h_xtfx in h_xtfx_list)
+        
+        if not has_strict_control:
+            # 无严格管控类：先判断比例是否 ≥80%
+            counter = Counter(h_xtfx_list)
+            most_common, count = counter.most_common(1)[0]
+            if count / len(points) >= 0.8:
+                return most_common
+            
+            # 比例不达标：对优先保护类和安全利用类进行插值
+            valid_points = [
+                (point['x'], point['y'], self.H_XTFX_MAPPING[point['h_xtfx']]) 
+                for point in points 
+                if point['h_xtfx'] in ["优先保护类", "安全利用类"]
+            ]
+            
+            if len(valid_points) < 2:
+                # 有效点位不足，取最常见值
+                return most_common
+            else:
+                # 获取单元中心点坐标
+                unit_center = self._get_unit_center(unit_id)
+                if unit_center is None:
+                    return most_common
+                
+                interpolated = self._idw_interpolation_simple(valid_points, unit_center)
+                if interpolated is None:
+                    return most_common
+                return self._interpolated_value_to_category(interpolated)
+        else:
+            # 存在严格管控类：对所有点位进行插值
+            all_points = [
+                (point['x'], point['y'], self.H_XTFX_MAPPING[point['h_xtfx']]) 
+                for point in points
+                if point['h_xtfx'] in self.H_XTFX_MAPPING
+            ]
+            
+            if not all_points:
+                return None
+            
+            # 获取单元中心点坐标
+            unit_center = self._get_unit_center(unit_id)
+            if unit_center is None:
+                return None
+            
+            interpolated = self._idw_interpolation_simple(all_points, unit_center)
+            if interpolated is None:
+                return None
+            return self._interpolated_value_to_category(interpolated)
+    
+    def _get_unit_center(self, unit_id: int) -> Optional[Tuple[float, float]]:
+        """
+        获取单元中心点坐标
+        
+        Args:
+            unit_id: 单元ID
+            
+        Returns:
+            Tuple[float, float]: (x, y) 坐标
+        """
+        try:
+            center_query = text("""
+                SELECT 
+                    ST_X(ST_Centroid(geom)) as center_x,
+                    ST_Y(ST_Centroid(geom)) as center_y
+                FROM unit_ceil
+                WHERE gid = :unit_id
+            """)
+            
+            result = self.db_session.execute(center_query, {"unit_id": unit_id}).fetchone()
+            if result:
+                return (result.center_x, result.center_y)
+            return None
+            
+        except Exception as e:
+            logger.error(f"获取单元 {unit_id} 中心点失败: {e}")
+            return None
+    
+    def _idw_interpolation_simple(self, points: List[Tuple], target_point: Tuple[float, float]) -> Optional[float]:
+        """
+        简单的反距离加权插值函数
+        
+        Args:
+            points: 点位坐标和值的列表 [(x, y, value), ...]
+            target_point: 目标点位坐标 (x, y)
+            
+        Returns:
+            float: 插值结果
+        """
+        if not points:
+            return None
+            
+        total_weight = 0
+        weighted_sum = 0
+        power = 2  # 距离权重的幂
+        
+        target_x, target_y = target_point
+        
+        for point_x, point_y, value in points:
+            # 计算欧几里得距离
+            distance = ((point_x - target_x) ** 2 + (point_y - target_y) ** 2) ** 0.5
+            
+            if distance == 0:
+                return value  # 距离为0时直接返回该点值
+            
+            weight = 1 / (distance ** power)
+            total_weight += weight
+            weighted_sum += weight * value
+            
+        return weighted_sum / total_weight if total_weight != 0 else None
+    
+    def _interpolated_value_to_category(self, interpolated_value: float) -> str:
+        """
+        将插值结果转换为类别
+        
+        Args:
+            interpolated_value: 插值结果
+            
+        Returns:
+            str: 类别名称
+        """
+        if interpolated_value <= 1.5:
+            return "优先保护类"
+        elif interpolated_value <= 2.5:
+            return "安全利用类"
+        else:
+            return "严格管控类"
+    
+    def get_unit_count(self) -> int:
+        """
+        获取单元总数
+        
+        Returns:
+            int: 单元总数
+        """
+        return self.db_session.query(UnitCeil).count()
+    
+    def get_point_count_by_h_xtfx(self) -> Dict[str, int]:
+        """
+        获取不同h_xtfx类别的点位数量统计
+        
+        Returns:
+            Dict[str, int]: 各类别点位数量
+        """
+        try:
+            result = self.db_session.query(
+                FiftyThousandSurveyDatum.h_xtfx,
+                func.count(FiftyThousandSurveyDatum.h_xtfx).label('count')
+            ).filter(
+                FiftyThousandSurveyDatum.h_xtfx.isnot(None)
+            ).group_by(
+                FiftyThousandSurveyDatum.h_xtfx
+            ).all()
+            
+            return {row.h_xtfx: row.count for row in result}
+            
+        except Exception as e:
+            logger.error(f"获取点位统计失败: {e}")
+            return {}
+    
+    def get_units_by_ids(self, unit_ids: List[int]) -> List[UnitCeil]:
+        """
+        批量获取单元信息
+        
+        Args:
+            unit_ids: 单元ID列表
+            
+        Returns:
+            List[UnitCeil]: 单元对象列表
+        """
+        return self.db_session.query(UnitCeil).filter(
+            UnitCeil.gid.in_(unit_ids)
+        ).all()
+    
+    def get_points_in_area(self, area_name: str = None) -> List[FiftyThousandSurveyDatum]:
+        """
+        获取特定区域的点位数据
+        
+        Args:
+            area_name: 区域名称（县名称）
+            
+        Returns:
+            List[FiftyThousandSurveyDatum]: 点位数据列表
+        """
+        query = self.db_session.query(FiftyThousandSurveyDatum).filter(
+            FiftyThousandSurveyDatum.h_xtfx.isnot(None)
+        )
+        
+        if area_name:
+            query = query.filter(FiftyThousandSurveyDatum.xmc == area_name)
+        
+        return query.all()
+    
+    def get_units_containing_points_optimized(self, point_ids: List[int]) -> List[Tuple[int, int]]:
+        """
+        优化的空间查询：找出包含指定点位的单元
+        
+        Args:
+            point_ids: 点位ID列表
+            
+        Returns:
+            List[Tuple[int, int]]: (unit_id, point_id) 对的列表
+        """
+        try:
+            # 使用数据库端的空间查询
+            spatial_query = text("""
+                SELECT 
+                    u.gid as unit_id,
+                    p.id as point_id
+                FROM unit_ceil u
+                JOIN fifty_thousand_survey_data p ON ST_Contains(
+                    u.geom, 
+                    ST_Transform(ST_SetSRID(p.geom, 4490), 4490)
+                )
+                WHERE p.id = ANY(:point_ids)
+            """)
+            
+            result = self.db_session.execute(spatial_query, {"point_ids": point_ids}).fetchall()
+            return [(row.unit_id, row.point_id) for row in result]
+            
+        except Exception as e:
+            logger.error(f"优化空间查询失败: {e}")
+            return []
+    
+    def get_area_statistics(self) -> Dict[str, Dict[str, int]]:
+        """
+        获取各区域的统计信息
+        
+        Returns:
+            Dict[str, Dict[str, int]]: 区域名称到统计信息的映射
+        """
+        try:
+            result = self.db_session.query(
+                FiftyThousandSurveyDatum.xmc,
+                FiftyThousandSurveyDatum.h_xtfx,
+                func.count(FiftyThousandSurveyDatum.id).label('count')
+            ).filter(
+                FiftyThousandSurveyDatum.h_xtfx.isnot(None),
+                FiftyThousandSurveyDatum.xmc.isnot(None)
+            ).group_by(
+                FiftyThousandSurveyDatum.xmc,
+                FiftyThousandSurveyDatum.h_xtfx
+            ).order_by(
+                FiftyThousandSurveyDatum.xmc
+            ).all()
+            
+            area_stats = {}
+            for row in result:
+                area_name = row.xmc
+                h_xtfx = row.h_xtfx
+                count = row.count
+                
+                if area_name not in area_stats:
+                    area_stats[area_name] = {}
+                
+                area_stats[area_name][h_xtfx] = count
+            
+            return area_stats
+            
+        except Exception as e:
+            logger.error(f"获取区域统计信息失败: {e}")
+            return {}
+    
+    def get_unit_h_xtfx_result(self) -> Dict[str, any]:
+        """
+        获取单元h_xtfx结果的API方法
+        
+        Returns:
+            Dict[str, any]: API响应数据
+        """
+        try:
+            result = self.calculate_unit_h_xtfx_values()
+            
+            total_units = self.get_unit_count()
+            units_with_data = sum(1 for v in result.values() if v is not None)
+            
+            # 按类别统计
+            category_stats = {}
+            for category in self.H_XTFX_MAPPING.keys():
+                category_stats[category] = sum(1 for v in result.values() if v == category)
+            
+            point_stats = self.get_point_count_by_h_xtfx()
+            
+            return {
+                "success": True,
+                "data": result,
+                "statistics": {
+                    "total_units": total_units,
+                    "units_with_data": units_with_data,
+                    "units_without_data": total_units - units_with_data,
+                    "category_distribution": category_stats,
+                    "point_distribution": point_stats
+                }
+            }
+            
+        except Exception as e:
+            logger.error(f"获取单元h_xtfx结果失败: {e}")
+            return {
+                "success": False,
+                "error": str(e),
+                "data": None
+            } 

docs/PROJECT_STRUCTURE.md

@@ -0,0 +1,204 @@
+# 项目结构说明
+
+本文档详细说明了项目的目录结构和组织方式。
+
+## 整体结构
+
+```
+AcidMap/
+├── app/                    # 主应用代码
+├── docs/                   # 项目文档
+├── tests/                  # 测试代码
+├── scripts/                # 脚本工具
+├── data/                   # 数据文件
+├── migrations/             # 数据库迁移
+├── ssl/                    # SSL证书
+├── config.env              # 配置文件
+├── environment.yml         # Conda环境配置
+├── main.py                 # 应用入口
+└── README.md               # 项目说明
+```
+
+## 应用代码结构 (app/)
+
+```
+app/
+├── api/                    # API路由层
+│   ├── raster.py          # 栅格数据API
+│   ├── vector.py          # 矢量数据API
+│   └── unit_grouping.py   # 单元分类API
+├── services/              # 业务逻辑层
+│   ├── raster_service.py  # 栅格数据服务
+│   ├── vector_service.py  # 矢量数据服务
+│   └── unit_grouping_service.py  # 单元分类服务
+├── models/                # 数据模型
+│   └── orm_models.py      # ORM模型定义
+├── utils/                 # 工具函数
+├── config/                # 配置管理
+├── static/                # 静态文件
+├── scripts/               # 应用内脚本
+├── logs/                  # 日志文件
+├── database.py            # 数据库连接
+├── main.py                # FastAPI主应用
+└── __init__.py
+```
+
+## 文档结构 (docs/)
+
+```
+docs/
+├── features/              # 功能文档
+│   ├── unit-grouping/     # 单元分类功能
+│   │   └── README.md      # 详细说明文档
+│   ├── raster-processing/ # 栅格处理功能 (待添加)
+│   └── vector-processing/ # 矢量处理功能 (待添加)
+├── api/                   # API文档 (待添加)
+├── deployment/            # 部署文档 (待添加)
+└── PROJECT_STRUCTURE.md   # 本文档
+```
+
+## 测试结构 (tests/)
+
+```
+tests/
+├── unit/                  # 单元测试
+├── integration/           # 集成测试
+│   ├── test_unit_grouping.py  # 单元分类集成测试
+│   └── test_cd_integration.py # 镉预测集成测试
+└── fixtures/              # 测试数据
+```
+
+## 脚本工具结构 (scripts/)
+
+```
+scripts/
+├── demos/                 # 演示脚本
+│   └── unit_grouping_demo.py  # 单元分类演示
+├── import_farmland_data.py    # 农田数据导入
+├── db_health_check.py         # 数据库健康检查
+└── import_counties.py         # 县区数据导入
+```
+
+## 目录组织原则
+
+### 1. 按功能模块组织
+- 每个主要功能模块都有对应的API、服务和文档
+- 相关文件集中在一起，便于维护
+
+### 2. 分层架构
+- **API层** (app/api/): 处理HTTP请求和响应
+- **服务层** (app/services/): 业务逻辑实现
+- **模型层** (app/models/): 数据模型和ORM
+
+### 3. 文档优先
+- 每个功能模块都有详细的文档说明
+- 文档与代码同步维护
+- 使用统一的文档格式
+
+### 4. 测试分类
+- **单元测试**: 测试单个函数或类
+- **集成测试**: 测试完整的功能流程
+- **演示脚本**: 提供快速体验功能的方式
+
+## 新功能集成指南
+
+当添加新功能时，请按照以下结构组织文件：
+
+### 1. API实现
+```
+app/api/your_feature.py
+```
+
+### 2. 业务逻辑
+```
+app/services/your_feature_service.py
+```
+
+### 3. 功能文档
+```
+docs/features/your-feature/
+├── README.md              # 功能说明
+├── api-reference.md       # API参考 (可选)
+└── examples.md            # 使用示例 (可选)
+```
+
+### 4. 测试代码
+```
+tests/integration/test_your_feature.py
+```
+
+### 5. 演示脚本 (可选)
+```
+scripts/demos/your_feature_demo.py
+```
+
+### 6. 更新主文档
+- 在 `README.md` 中添加功能描述
+- 在 `README.md` 中添加API接口说明
+- 在本文档中更新目录结构
+
+## 文件命名规范
+
+### 1. 文件名
+- 使用小写字母和下划线
+- API文件: `{feature}.py`
+- 服务文件: `{feature}_service.py`
+- 测试文件: `test_{feature}.py`
+- 演示脚本: `{feature}_demo.py`
+
+### 2. 目录名
+- 使用小写字母和连字符
+- 功能目录: `{feature-name}/`
+- 文档目录: `docs/features/{feature-name}/`
+
+### 3. 文档名
+- 主文档: `README.md`
+- 其他文档: `{purpose}.md` (如 `api-reference.md`)
+
+## 最佳实践
+
+### 1. 代码组织
+- 保持各层职责清晰
+- 避免循环依赖
+- 使用依赖注入
+
+### 2. 文档维护
+- 代码变更时同步更新文档
+- 提供完整的使用示例
+- 包含错误处理说明
+
+### 3. 测试覆盖
+- 每个功能都要有集成测试
+- 重要功能提供演示脚本
+- 测试要能独立运行
+
+### 4. 版本控制
+- 功能相关的文件一起提交
+- 使用清晰的提交信息
+- 避免在主目录下堆积临时文件
+
+## 示例：单元分类功能
+
+单元分类功能是按照上述结构组织的一个很好的例子：
+
+- **API**: `app/api/unit_grouping.py`
+- **服务**: `app/services/unit_grouping_service.py`
+- **文档**: `docs/features/unit-grouping/README.md`
+- **测试**: `tests/integration/test_unit_grouping.py`
+- **演示**: `scripts/demos/unit_grouping_demo.py`
+
+这种组织方式确保了：
+- 功能完整性
+- 文档完备性
+- 测试覆盖率
+- 易于维护
+
+## 总结
+
+良好的项目结构能够：
+- 提高开发效率
+- 降低维护成本
+- 方便新人上手
+- 支持项目扩展
+
+请在添加新功能时严格遵循这些组织原则和命名规范。 

docs/features/unit-grouping/README.md

@@ -0,0 +1,327 @@
+# 单元分类功能集成说明
+
+## 功能概述
+
+单元分类功能已成功集成到项目中，提供基于点位数据的单元h_xtfx值计算功能。该功能使用反距离加权插值算法，根据五万亩调查数据中的点位信息来计算各单元的土壤环境质量类别(h_xtfx)。
+
+## 核心算法
+
+### 分类逻辑
+- **优先保护类**: 数值映射为 1
+- **安全利用类**: 数值映射为 2  
+- **严格管控类**: 数值映射为 3
+
+### 计算规则
+1. **无严格管控类点位时**:
+   - 优先判断某类别占比是否≥80%，是则直接采用该类别
+   - 否则对优先保护类和安全利用类点位进行反距离加权插值
+   
+2. **存在严格管控类点位时**:
+   - 对所有点位（包括严格管控类）进行反距离加权插值
+
+### 插值结果分类
+- ≤ 1.5: 优先保护类
+- 1.5 < x ≤ 2.5: 安全利用类
+- > 2.5: 严格管控类
+
+## API接口
+
+### 核心功能接口
+
+#### 1. 获取所有单元的h_xtfx分类结果
+```http
+GET /api/unit-grouping/h_xtfx
+```
+
+**响应示例:**
+```json
+{
+  "success": true,
+  "data": {
+    "1": "优先保护类",
+    "2": "安全利用类",
+    "3": null,
+    "4": "严格管控类"
+  },
+  "statistics": {
+    "total_units": 1000,
+    "units_with_data": 856,
+    "units_without_data": 144,
+    "category_distribution": {
+      "优先保护类": 324,
+      "安全利用类": 432,
+      "严格管控类": 100
+    },
+    "point_distribution": {
+      "优先保护类": 1200,
+      "安全利用类": 800,
+      "严格管控类": 150
+    }
+  }
+}
+```
+
+#### 2. 获取统计信息
+```http
+GET /api/unit-grouping/statistics
+```
+
+#### 3. 获取特定单元的h_xtfx值
+```http
+GET /api/unit-grouping/unit/{unit_id}
+```
+
+### ORM功能接口
+
+#### 4. 获取点位统计信息（ORM）
+```http
+GET /api/unit-grouping/points/statistics
+```
+
+**响应示例:**
+```json
+{
+  "success": true,
+  "total_points": 2150,
+  "distribution": {
+    "优先保护类": {
+      "count": 1200,
+      "percentage": 55.81
+    },
+    "安全利用类": {
+      "count": 800,
+      "percentage": 37.21
+    },
+    "严格管控类": {
+      "count": 150,
+      "percentage": 6.98
+    }
+  }
+}
+```
+
+#### 5. 批量获取单元信息（ORM）
+```http
+GET /api/unit-grouping/units/batch?unit_ids=1&unit_ids=2&unit_ids=3
+```
+
+**响应示例:**
+```json
+{
+  "success": true,
+  "total_requested": 3,
+  "total_found": 2,
+  "data": [
+    {
+      "gid": 1,
+      "BSM": "440200001",
+      "PXZQMC": "广东省",
+      "CXZQMC": "韶关市",
+      "SUM_NYDTBM": 1234.56,
+      "Shape_Area": 987654.32
+    }
+  ]
+}
+```
+
+#### 6. 按区域获取点位数据（ORM）
+```http
+GET /api/unit-grouping/points/by-area?area_name=乐昌市
+```
+
+**响应示例:**
+```json
+{
+  "success": true,
+  "area_name": "乐昌市",
+  "total_points": 125,
+  "h_xtfx_distribution": {
+    "优先保护类": 80,
+    "安全利用类": 35,
+    "严格管控类": 10
+  },
+  "data": [
+    {
+      "id": 1,
+      "dwmc": "LC001",
+      "xmc": "乐昌市",
+      "zmc": "乐城街道",
+      "cmc": "某村",
+      "h_xtfx": "优先保护类",
+      "lat": 25.123456,
+      "lon": 113.654321,
+      "ph": 6.5
+    }
+  ]
+}
+```
+
+#### 7. 获取数据库摘要信息（ORM）
+```http
+GET /api/unit-grouping/database/summary
+```
+
+**响应示例:**
+```json
+{
+  "success": true,
+  "summary": {
+    "total_units": 1000,
+    "total_points": 2150,
+    "points_with_h_xtfx": 2150,
+    "h_xtfx_categories": 3,
+    "point_distribution": {
+      "优先保护类": 1200,
+      "安全利用类": 800,
+      "严格管控类": 150
+    }
+  }
+}
+```
+
+## 项目文件结构
+
+```
+app/
+├── api/
+│   └── unit_grouping.py          # API接口定义
+├── services/
+│   └── unit_grouping_service.py  # 业务逻辑服务
+├── models/
+│   └── orm_models.py             # 数据模型(已存在)
+└── main.py                       # 主应用(已更新)
+```
+
+## 使用步骤
+
+### 1. 安装依赖
+确保已安装所需的Python包：
+```bash
+# 使用conda环境
+conda env create -f environment.yml
+conda activate your_env_name
+
+# 或使用pip（如果有requirements.txt）
+pip install shapely
+```
+
+### 2. 启动服务
+```bash
+# 在项目根目录下运行
+uvicorn main:app --reload
+
+# 或使用演示脚本
+python scripts/demos/unit_grouping_demo.py
+```
+
+### 3. 访问API文档
+打开浏览器访问: `http://localhost:8000/docs`
+
+### 4. 测试功能
+```bash
+# 运行集成测试
+python tests/integration/test_unit_grouping.py
+```
+
+## 数据模型
+
+### 输入数据
+- **unit_ceil表**: 单元几何数据
+  - `gid`: 单元ID
+  - `geom`: 几何信息
+  
+- **fifty_thousand_survey_data表**: 五万亩调查数据
+  - `id`: 点位ID
+  - `geom`: 点位几何信息
+  - `h_xtfx`: 土壤环境质量类别
+
+### 输出数据
+- 单元ID到h_xtfx分类的映射
+- 统计信息（总数、分布等）
+
+## ORM vs SQL 对比
+
+### 使用ORM的优势
+
+1. **类型安全**: 编译时检查，减少运行时错误
+2. **代码可维护性**: 更加Pythonic，易于理解和维护
+3. **自动化功能**: 
+   - 自动处理连接管理
+   - 自动类型转换
+   - 自动防SQL注入
+4. **查询优化**: SQLAlchemy自动优化查询
+5. **关系映射**: 自动处理表间关系
+
+### 代码对比示例
+
+**传统SQL方式:**
+```python
+# 原始SQL查询
+unit_query = text("SELECT gid, ST_AsBinary(geom) as geom FROM unit_ceil")
+unit_result = self.db_session.execute(unit_query)
+unit_rows = unit_result.fetchall()
+```
+
+**ORM方式:**
+```python
+# 使用ORM查询
+unit_query = self.db_session.query(
+    UnitCeil.gid,
+    func.ST_AsBinary(UnitCeil.geom).label('geom')
+)
+unit_rows = unit_query.all()
+```
+
+### 新增的ORM功能
+
+1. **统计查询**: 使用`func.count()`和`group_by()`
+2. **批量查询**: 使用`filter(Model.id.in_(ids))`
+3. **条件过滤**: 使用`filter()`和`isnot(None)`
+4. **结果计数**: 使用`query().count()`
+
+## 性能考虑
+
+1. **空间查询优化**: 使用PostGIS的空间索引来加速点位包含查询
+2. **数据过滤**: 只查询有效的h_xtfx值，避免处理空值
+3. **异常处理**: 对几何数据解析失败的情况进行graceful处理
+4. **日志记录**: 详细的操作日志便于调试和监控
+5. **ORM优化**: 
+   - 延迟加载减少内存占用
+   - 批量查询减少数据库往返
+   - 查询缓存提高重复查询性能
+
+## 错误处理
+
+- 数据库连接失败
+- 几何数据解析异常
+- 插值计算错误
+- 单元不存在等情况
+
+所有错误都会返回适当的HTTP状态码和错误信息。
+
+## 扩展功能
+
+未来可以考虑添加：
+1. 缓存机制以提高查询性能
+2. 批量查询单元功能
+3. 结果导出功能（GeoJSON、CSV等）
+4. 可视化界面
+5. 算法参数配置（如插值权重、阈值等）
+
+## 注意事项
+
+1. 确保数据库中有足够的单元和点位数据
+2. 几何数据的坐标系统需要一致
+3. 服务启动前确保数据库连接正常
+4. 大数据量时可能需要考虑分页处理
+
+## 相关文件
+
+- **API实现**: `app/api/unit_grouping.py`
+- **业务逻辑**: `app/services/unit_grouping_service.py`
+- **集成测试**: `tests/integration/test_unit_grouping.py`
+- **演示脚本**: `scripts/demos/unit_grouping_demo.py`
+
+## 联系方式
+
+如有问题或建议，请联系开发团队。 

scripts/demos/unit_grouping_demo.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+单元分类功能演示脚本
+
+该脚本帮助用户快速启动和测试单元分类功能
+"""
+import os
+import sys
+import subprocess
+import time
+import threading
+from pathlib import Path
+
+def print_banner():
+    """打印横幅"""
+    print("=" * 70)
+    print("单元分类功能演示")
+    print("=" * 70)
+    print()
+
+def check_dependencies():
+    """检查依赖项"""
+    print("🔍 检查依赖项...")
+    
+    required_packages = [
+        'fastapi',
+        'uvicorn',
+        'sqlalchemy',
+        'shapely',
+        'psycopg2',
+        'geoalchemy2'
+    ]
+    
+    missing_packages = []
+    
+    for package in required_packages:
+        try:
+            __import__(package)
+            print(f"✓ {package}")
+        except ImportError:
+            missing_packages.append(package)
+            print(f"✗ {package} (缺失)")
+    
+    if missing_packages:
+        print("\n❌ 缺少依赖项，请先安装:")
+        print("conda env create -f environment.yml")
+        print("或手动安装: pip install " + " ".join(missing_packages))
+        return False
+    
+    print("✓ 所有依赖项已安装")
+    return True
+
+def check_database_config():
+    """检查数据库配置"""
+    print("\n🔍 检查数据库配置...")
+    
+    # 回到项目根目录查找配置文件
+    project_root = Path(__file__).parent.parent.parent
+    config_file = project_root / "config.env"
+    
+    if not config_file.exists():
+        print("❌ 配置文件 config.env 不存在")
+        return False
+    
+    # 检查基本配置项
+    required_vars = ['DB_HOST', 'DB_PORT', 'DB_NAME', 'DB_USER', 'DB_PASSWORD']
+    
+    try:
+        from dotenv import load_dotenv
+        load_dotenv(str(config_file))
+        
+        for var in required_vars:
+            if not os.getenv(var):
+                print(f"❌ 环境变量 {var} 未设置")
+                return False
+        
+        print("✓ 数据库配置正确")
+        return True
+        
+    except Exception as e:
+        print(f"❌ 检查数据库配置时出错: {e}")
+        return False
+
+def start_server():
+    """启动服务器"""
+    print("\n🚀 启动FastAPI服务器...")
+    print("服务器将在 http://localhost:8000 启动")
+    print("API文档地址: http://localhost:8000/docs")
+    print("按 Ctrl+C 停止服务器")
+    print("-" * 50)
+    
+    try:
+        # 切换到项目根目录
+        project_root = Path(__file__).parent.parent.parent
+        os.chdir(project_root)
+        
+        # 使用subprocess启动uvicorn
+        process = subprocess.Popen([
+            sys.executable, '-m', 'uvicorn', 
+            'main:app', 
+            '--reload',
+            '--host', '0.0.0.0',
+            '--port', '8000'
+        ])
+        
+        # 等待服务器启动
+        time.sleep(3)
+        
+        return process
+        
+    except Exception as e:
+        print(f"❌ 启动服务器失败: {e}")
+        return None
+
+def run_tests():
+    """运行测试"""
+    print("\n🧪 运行集成测试...")
+    print("-" * 50)
+    
+    try:
+        # 等待服务器完全启动
+        time.sleep(2)
+        
+        # 运行测试脚本
+        project_root = Path(__file__).parent.parent.parent
+        test_file = project_root / "tests" / "integration" / "test_unit_grouping.py"
+        
+        if test_file.exists():
+            subprocess.run([sys.executable, str(test_file)])
+        else:
+            print(f"❌ 测试文件不存在: {test_file}")
+        
+    except Exception as e:
+        print(f"❌ 运行测试失败: {e}")
+
+def show_usage_info():
+    """显示使用信息"""
+    print("\n📖 使用说明:")
+    print("-" * 50)
+    print("1. 主要API端点:")
+    print("   GET /api/unit-grouping/h_xtfx          - 获取所有单元的h_xtfx分类")
+    print("   GET /api/unit-grouping/statistics      - 获取统计信息")
+    print("   GET /api/unit-grouping/unit/{unit_id}  - 获取特定单元的h_xtfx值")
+    print()
+    print("2. 测试示例:")
+    print("   curl http://localhost:8000/api/unit-grouping/h_xtfx")
+    print("   curl http://localhost:8000/api/unit-grouping/statistics")
+    print()
+    print("3. 查看完整API文档:")
+    print("   http://localhost:8000/docs")
+    print()
+    print("4. 重新运行测试:")
+    print("   python tests/integration/test_unit_grouping.py")
+    print()
+    print("5. 查看功能文档:")
+    print("   docs/features/unit-grouping/README.md")
+
+def main():
+    """主函数"""
+    print_banner()
+    
+    # 检查依赖项
+    if not check_dependencies():
+        return
+    
+    # 检查数据库配置
+    if not check_database_config():
+        return
+    
+    # 启动服务器
+    server_process = start_server()
+    if not server_process:
+        return
+    
+    try:
+        # 运行测试
+        test_thread = threading.Thread(target=run_tests)
+        test_thread.daemon = True
+        test_thread.start()
+        
+        # 等待测试完成
+        test_thread.join(timeout=30)
+        
+        # 显示使用信息
+        show_usage_info()
+        
+        print("\n🎉 服务器正在运行中...")
+        print("按 Ctrl+C 退出")
+        
+        # 等待用户中断
+        server_process.wait()
+        
+    except KeyboardInterrupt:
+        print("\n\n👋 正在关闭服务器...")
+        server_process.terminate()
+        server_process.wait()
+        print("服务器已关闭")
+    
+    except Exception as e:
+        print(f"\n❌ 运行时出错: {e}")
+        if server_process:
+            server_process.terminate()
+
+if __name__ == "__main__":
+    main() 

tests/integration/test_unit_grouping.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+单元分类功能集成测试
+
+该测试文件用于验证单元分类功能是否正确集成到项目中
+"""
+import requests
+import json
+from typing import Dict, Any
+
+def test_unit_grouping_api():
+    """
+    测试单元分类API接口
+    """
+    base_url = "http://localhost:8000"
+    
+    print("=" * 60)
+    print("单元分类功能集成测试")
+    print("=" * 60)
+    
+    # 测试1: 获取单元h_xtfx分类结果
+    print("\n1. 测试获取单元h_xtfx分类结果")
+    print("-" * 40)
+    
+    try:
+        response = requests.get(f"{base_url}/api/unit-grouping/h_xtfx")
+        
+        if response.status_code == 200:
+            data = response.json()
+            print(f"✓ 请求成功")
+            print(f"✓ 成功状态: {data.get('success', False)}")
+            
+            if data.get('success', False):
+                statistics = data.get('statistics', {})
+                print(f"✓ 总单元数: {statistics.get('total_units', 0)}")
+                print(f"✓ 有数据的单元数: {statistics.get('units_with_data', 0)}")
+                print(f"✓ 无数据的单元数: {statistics.get('units_without_data', 0)}")
+                
+                category_dist = statistics.get('category_distribution', {})
+                print(f"✓ 类别分布:")
+                for category, count in category_dist.items():
+                    print(f"  - {category}: {count}")
+                    
+                # 显示前5个单元的结果
+                unit_data = data.get('data', {})
+                if unit_data:
+                    print(f"✓ 前5个单元结果样例:")
+                    for i, (unit_id, h_xtfx) in enumerate(list(unit_data.items())[:5]):
+                        print(f"  - 单元 {unit_id}: {h_xtfx}")
+            else:
+                print(f"✗ 服务返回失败: {data.get('error', 'Unknown error')}")
+        else:
+            print(f"✗ 请求失败，状态码: {response.status_code}")
+            print(f"✗ 错误信息: {response.text}")
+            
+    except requests.exceptions.ConnectionError:
+        print("✗ 连接失败，请确保服务正在运行 (uvicorn main:app --reload)")
+    except Exception as e:
+        print(f"✗ 测试异常: {str(e)}")
+    
+    # 测试2: 获取统计信息
+    print("\n2. 测试获取统计信息")
+    print("-" * 40)
+    
+    try:
+        response = requests.get(f"{base_url}/api/unit-grouping/statistics")
+        
+        if response.status_code == 200:
+            data = response.json()
+            print(f"✓ 请求成功")
+            print(f"✓ 成功状态: {data.get('success', False)}")
+            
+            if data.get('success', False):
+                statistics = data.get('statistics', {})
+                print(f"✓ 统计信息获取成功")
+                print(f"  - 总单元数: {statistics.get('total_units', 0)}")
+                print(f"  - 有数据的单元数: {statistics.get('units_with_data', 0)}")
+                print(f"  - 无数据的单元数: {statistics.get('units_without_data', 0)}")
+        else:
+            print(f"✗ 请求失败，状态码: {response.status_code}")
+            print(f"✗ 错误信息: {response.text}")
+            
+    except requests.exceptions.ConnectionError:
+        print("✗ 连接失败，请确保服务正在运行")
+    except Exception as e:
+        print(f"✗ 测试异常: {str(e)}")
+    
+    # 测试3: 获取特定单元的h_xtfx值
+    print("\n3. 测试获取特定单元的h_xtfx值")
+    print("-" * 40)
+    
+    try:
+        # 首先获取一个存在的单元ID
+        response = requests.get(f"{base_url}/api/unit-grouping/h_xtfx")
+        if response.status_code == 200:
+            data = response.json()
+            if data.get('success', False):
+                unit_data = data.get('data', {})
+                if unit_data:
+                    # 取第一个单元进行测试
+                    test_unit_id = list(unit_data.keys())[0]
+                    
+                    # 测试获取特定单元
+                    response = requests.get(f"{base_url}/api/unit-grouping/unit/{test_unit_id}")
+                    
+                    if response.status_code == 200:
+                        unit_result = response.json()
+                        print(f"✓ 请求成功")
+                        print(f"✓ 单元 {test_unit_id} 的h_xtfx值: {unit_result.get('h_xtfx')}")
+                    else:
+                        print(f"✗ 请求失败，状态码: {response.status_code}")
+                else:
+                    print("✗ 没有可用的单元数据进行测试")
+            else:
+                print("✗ 无法获取单元数据进行测试")
+        else:
+            print("✗ 无法获取单元数据进行测试")
+            
+    except requests.exceptions.ConnectionError:
+        print("✗ 连接失败，请确保服务正在运行")
+    except Exception as e:
+        print(f"✗ 测试异常: {str(e)}")
+    
+    print("\n" + "=" * 60)
+    print("测试完成")
+    print("=" * 60)
+    
+    # 测试4: 测试新的ORM接口
+    print("\n4. 测试新的ORM接口功能")
+    print("-" * 40)
+    
+    # 测试点位统计信息
+    try:
+        response = requests.get(f"{base_url}/api/unit-grouping/points/statistics")
+        
+        if response.status_code == 200:
+            data = response.json()
+            print(f"✓ 点位统计信息获取成功")
+            distribution = data.get('distribution', {})
+            print(f"✓ 总点位数: {data.get('total_points', 0)}")
+            for category, stats in distribution.items():
+                print(f"  - {category}: {stats['count']} ({stats['percentage']}%)")
+        else:
+            print(f"✗ 获取点位统计信息失败，状态码: {response.status_code}")
+    except Exception as e:
+        print(f"✗ 测试点位统计信息异常: {str(e)}")
+    
+    # 测试数据库摘要信息
+    try:
+        response = requests.get(f"{base_url}/api/unit-grouping/database/summary")
+        
+        if response.status_code == 200:
+            data = response.json()
+            print(f"✓ 数据库摘要信息获取成功")
+            summary = data.get('summary', {})
+            print(f"  - 总单元数: {summary.get('total_units', 0)}")
+            print(f"  - 总点位数: {summary.get('total_points', 0)}")
+            print(f"  - h_xtfx分类数: {summary.get('h_xtfx_categories', 0)}")
+        else:
+            print(f"✗ 获取数据库摘要信息失败，状态码: {response.status_code}")
+    except Exception as e:
+        print(f"✗ 测试数据库摘要信息异常: {str(e)}")
+    
+    # 测试批量获取单元信息
+    try:
+        response = requests.get(f"{base_url}/api/unit-grouping/units/batch?unit_ids=1&unit_ids=2&unit_ids=3")
+        
+        if response.status_code == 200:
+            data = response.json()
+            print(f"✓ 批量获取单元信息成功")
+            print(f"  - 请求数量: {data.get('total_requested', 0)}")
+            print(f"  - 找到数量: {data.get('total_found', 0)}")
+        else:
+            print(f"✗ 批量获取单元信息失败，状态码: {response.status_code}")
+    except Exception as e:
+        print(f"✗ 测试批量获取单元信息异常: {str(e)}")
+    
+    # 输出使用说明
+    print("\n使用说明:")
+    print("1. 启动服务: uvicorn main:app --reload")
+    print("2. 访问API文档: http://localhost:8000/docs")
+    print("3. 主要接口:")
+    print("   - GET /api/unit-grouping/h_xtfx - 获取所有单元的h_xtfx分类结果")
+    print("   - GET /api/unit-grouping/statistics - 获取统计信息")
+    print("   - GET /api/unit-grouping/unit/{unit_id} - 获取特定单元的h_xtfx值")
+    print("   - GET /api/unit-grouping/points/statistics - 获取点位统计信息（ORM）")
+    print("   - GET /api/unit-grouping/database/summary - 获取数据库摘要信息（ORM）")
+    print("   - GET /api/unit-grouping/units/batch - 批量获取单元信息（ORM）")
+    print("   - GET /api/unit-grouping/points/by-area - 按区域获取点位数据（ORM）")
+
+if __name__ == "__main__":
+    test_unit_grouping_api()