from geopipe_agent import step, StepContext, StepResult

@step(
    id="custom.my_step",        # 步骤唯一标识
    name="My Custom Step",      # 显示名称
    description="步骤的详细描述", # 描述信息
    category="custom",          # 类别
    params={...},               # 参数定义
    outputs={...},              # 输出定义
    backends=["native_python"], # 支持的后端
)
def my_custom_step(ctx: StepContext) -> StepResult:
    # 步骤实现
    ...

class StepContext:
    """步骤执行上下文"""

    def input(self, name: str):
        """获取输入数据（通过引用 $step_id 传入的数据）"""
        ...

    def param(self, name: str, default=None):
        """获取参数值"""
        ...

    def get_variable(self, name: str):
        """获取流水线变量值"""
        ...

    @property
    def logger(self):
        """获取日志记录器"""
        ...

class StepResult:
    """步骤执行结果"""

    def __init__(self, output=None, stats=None):
        """
        Args:
            output: 主输出数据（GeoDataFrame、栅格数据或文件路径等）
            stats: 统计信息字典，可在后续步骤的 when 条件中引用
        """
        self.output = output
        self.stats = stats or {}

params={
    "input": {
        "type": "layer",
        "required": True,
        "description": "输入矢量图层",
    },
    "distance": {
        "type": "number",
        "required": True,
        "description": "缓冲距离（地图单位）",
    },
    "resolution": {
        "type": "integer",
        "required": False,
        "default": 16,
        "description": "缓冲区圆弧分辨率",
    },
    "cap_style": {
        "type": "string",
        "required": False,
        "default": "round",
        "description": "端点样式",
        "enum": ["round", "flat", "square"],
    },
}

"join_style": {
    "type": "string",
    "required": False,
    "default": "round",
    "description": "连接样式",
    "enum": ["round", "mitre", "bevel"],
}

outputs={
    "output": {
        "type": "layer",
        "description": "处理后的矢量图层",
    },
}

outputs={
    "output": {
        "type": "layer",
        "description": "主输出图层",
    },
    "report": {
        "type": "object",
        "description": "质检报告",
    },
}

# src/geopipe_agent/steps/vector/calculate_area.py

from geopipe_agent import step, StepContext, StepResult

@step(
    id="vector.calculate_area",
    name="Calculate Area",
    description="计算多边形要素的面积并添加为新列",
    category="vector",
    params={
        "input": {
            "type": "layer",
            "required": True,
            "description": "输入多边形矢量图层",
        },
        "unit": {
            "type": "string",
            "required": False,
            "default": "sqm",
            "description": "面积单位",
            "enum": ["sqm", "sqkm", "hectare"],
        },
        "column_name": {
            "type": "string",
            "required": False,
            "default": "area",
            "description": "面积列的列名",
        },
    },
    outputs={
        "output": {
            "type": "layer",
            "description": "添加了面积列的矢量图层",
        },
    },
    backends=["native_python"],
)
def calculate_area(ctx: StepContext) -> StepResult:
    """计算多边形面积"""
    # 获取输入和参数
    gdf = ctx.input("input")
    unit = ctx.param("unit", "sqm")
    column_name = ctx.param("column_name", "area")

    ctx.logger.info(f"计算面积，单位: {unit}，输出列: {column_name}")

    # 复制以避免修改原始数据
    result = gdf.copy()

    # 计算面积（以地图单位的平方为基础）
    area = result.geometry.area

    # 单位转换
    UNIT_FACTORS = {
        "sqm": 1.0,           # 平方米
        "sqkm": 1e-6,         # 平方千米
        "hectare": 1e-4,      # 公顷
    }
    factor = UNIT_FACTORS.get(unit, 1.0)
    result[column_name] = area * factor

    # 统计信息
    stats = {
        "feature_count": len(result),
        "min_area": float(result[column_name].min()),
        "max_area": float(result[column_name].max()),
        "mean_area": float(result[column_name].mean()),
        "total_area": float(result[column_name].sum()),
        "unit": unit,
    }

    ctx.logger.info(
        f"面积计算完成: {stats['feature_count']} 个要素, "
        f"总面积: {stats['total_area']:.2f} {unit}"
    )

    return StepResult(output=result, stats=stats)

# src/geopipe_agent/steps/raster/resample.py

import numpy as np
from rasterio.enums import Resampling
from rasterio.transform import from_bounds

from geopipe_agent import step, StepContext, StepResult

@step(
    id="raster.resample",
    name="Raster Resample",
    description="对栅格数据进行空间分辨率重采样",
    category="raster",
    params={
        "input": {
            "type": "layer",
            "required": True,
            "description": "输入栅格数据",
        },
        "resolution": {
            "type": "number",
            "required": True,
            "description": "目标分辨率（地图单位）",
        },
        "method": {
            "type": "string",
            "required": False,
            "default": "bilinear",
            "description": "重采样方法",
            "enum": ["nearest", "bilinear", "cubic", "average", "mode"],
        },
    },
    outputs={
        "output": {
            "type": "layer",
            "description": "重采样后的栅格数据",
        },
    },
    backends=["native_python"],
)
def resample(ctx: StepContext) -> StepResult:
    """栅格重采样"""
    dataset = ctx.input("input")
    target_res = ctx.param("resolution")
    method_name = ctx.param("method", "bilinear")

    ctx.logger.info(
        f"栅格重采样: {dataset.res} -> ({target_res}, {target_res}), "
        f"方法: {method_name}"
    )

    # 选择重采样方法
    METHODS = {
        "nearest": Resampling.nearest,
        "bilinear": Resampling.bilinear,
        "cubic": Resampling.cubic,
        "average": Resampling.average,
        "mode": Resampling.mode,
    }
    resampling = METHODS.get(method_name, Resampling.bilinear)

    # 计算新的尺寸
    bounds = dataset.bounds
    new_width = int((bounds.right - bounds.left) / target_res)
    new_height = int((bounds.top - bounds.bottom) / target_res)

    # 计算新的变换矩阵
    new_transform = from_bounds(
        bounds.left, bounds.bottom, bounds.right, bounds.top,
        new_width, new_height
    )

    # 读取并重采样数据
    data = dataset.read(
        out_shape=(dataset.count, new_height, new_width),
        resampling=resampling,
    )

    # 构建结果元数据
    result_meta = dataset.meta.copy()
    result_meta.update({
        "width": new_width,
        "height": new_height,
        "transform": new_transform,
    })

    stats = {
        "original_resolution": list(dataset.res),
        "target_resolution": [target_res, target_res],
        "original_size": [dataset.width, dataset.height],
        "new_size": [new_width, new_height],
        "method": method_name,
        "bands": dataset.count,
    }

    ctx.logger.info(
        f"重采样完成: {dataset.width}x{dataset.height} -> "
        f"{new_width}x{new_height}"
    )

    return StepResult(
        output={"data": data, "meta": result_meta},
        stats=stats,
    )

from dataclasses import dataclass

@dataclass
class QcIssue:
    """质检问题记录"""
    feature_id: int       # 要素 ID
    issue_type: str       # 问题类型
    severity: str         # 严重程度: error, warning, info
    message: str          # 问题描述
    field: str = None     # 相关字段名
    value: str = None     # 问题值

# src/geopipe_agent/steps/qc/code_standard.py

from geopipe_agent import step, StepContext, StepResult

@step(
    id="qc.code_standard",
    name="Code Standard Check",
    description="检查属性值是否符合预定义的编码规范",
    category="qc",
    params={
        "input": {
            "type": "layer",
            "required": True,
            "description": "输入矢量图层",
        },
        "column": {
            "type": "string",
            "required": True,
            "description": "要检查的列名",
        },
        "valid_codes": {
            "type": "array",
            "required": True,
            "description": "合法的编码值列表",
        },
        "severity": {
            "type": "string",
            "required": False,
            "default": "error",
            "description": "问题严重程度",
            "enum": ["error", "warning", "info"],
        },
    },
    outputs={
        "output": {
            "type": "layer",
            "description": "原始输入图层（未修改）",
        },
    },
    backends=["native_python"],
)
def code_standard_check(ctx: StepContext) -> StepResult:
    """编码规范检查"""
    gdf = ctx.input("input")
    column = ctx.param("column")
    valid_codes = ctx.param("valid_codes")
    severity = ctx.param("severity", "error")

    ctx.logger.info(f"编码规范检查: 列={column}, 合法编码数={len(valid_codes)}")

    # 检查列是否存在
    if column not in gdf.columns:
        raise ValueError(f"列 '{column}' 不存在于数据中，可用列: {list(gdf.columns)}")

    # 找出不合法的编码
    valid_set = set(valid_codes)
    issues = []
    invalid_count = 0
    null_count = 0

    for idx, row in gdf.iterrows():
        value = row[column]

        # 检查空值
        if value is None or (isinstance(value, float) and str(value) == "nan"):
            null_count += 1
            issues.append({
                "feature_id": int(idx),
                "issue_type": "null_code",
                "severity": severity,
                "message": f"编码为空值",
                "field": column,
                "value": None,
            })
            continue

        # 检查值是否合法
        if str(value) not in valid_set:
            invalid_count += 1
            issues.append({
                "feature_id": int(idx),
                "issue_type": "invalid_code",
                "severity": severity,
                "message": f"编码 '{value}' 不在合法范围内",
                "field": column,
                "value": str(value),
            })

    total = len(gdf)
    pass_count = total - invalid_count - null_count
    pass_rate = pass_count / total if total > 0 else 1.0

    stats = {
        "total_features": total,
        "pass_count": pass_count,
        "invalid_count": invalid_count,
        "null_count": null_count,
        "pass_rate": round(pass_rate, 4),
        "issues": issues,
        "column": column,
    }

    ctx.logger.info(
        f"检查完成: 通过率 {pass_rate:.1%} "
        f"({pass_count}/{total}), "
        f"无效编码: {invalid_count}, 空值: {null_count}"
    )

    return StepResult(output=gdf, stats=stats)

from geopipe_agent.backends import run_backend_op

@step(
    id="custom.multi_backend_step",
    name="Multi Backend Step",
    description="支持多后端的步骤",
    category="custom",
    params={...},
    outputs={...},
    backends=["native_python", "geopandas", "qgis"],
)
def multi_backend_step(ctx: StepContext) -> StepResult:
    gdf = ctx.input("input")

    # 委托给后端执行
    result = run_backend_op(
        operation="buffer",
        backend=ctx.backend,       # 引擎选择的后端
        input_data=gdf,
        distance=ctx.param("distance"),
    )

    return StepResult(output=result)

import pkgutil
import importlib

def discover_steps():
    """自动发现并加载所有步骤模块"""
    steps_package = importlib.import_module("geopipe_agent.steps")

    for importer, modname, ispkg in pkgutil.walk_packages(
        steps_package.__path__,
        prefix=steps_package.__name__ + ".",
    ):
        try:
            importlib.import_module(modname)
        except ImportError as e:
            logger.warning(f"无法加载步骤模块 {modname}: {e}")

# tests/conftest.py

import pytest
import geopandas as gpd
from shapely.geometry import Point, Polygon, box

@pytest.fixture
def sample_point_gdf():
    """创建示例点数据"""
    points = [Point(116.3 + i * 0.01, 39.9 + i * 0.01) for i in range(10)]
    return gpd.GeoDataFrame(
        {"name": [f"point_{i}" for i in range(10)]},
        geometry=points,
        crs="EPSG:4326",
    )

@pytest.fixture
def sample_polygon_gdf():
    """创建示例多边形数据"""
    polygons = [
        box(116.0 + i * 0.1, 39.0, 116.1 + i * 0.1, 39.1)
        for i in range(5)
    ]
    return gpd.GeoDataFrame(
        {"name": [f"poly_{i}" for i in range(5)],
         "land_type": ["011", "012", "021", "999", None]},
        geometry=polygons,
        crs="EPSG:4326",
    )

# tests/steps/vector/test_calculate_area.py

import pytest
import geopandas as gpd
from shapely.geometry import box

from geopipe_agent.steps.vector.calculate_area import calculate_area

class MockStepContext:
    """模拟 StepContext"""

    def __init__(self, input_data, params):
        self._input = input_data
        self._params = params
        self.logger = MockLogger()

    def input(self, name):
        return self._input

    def param(self, name, default=None):
        return self._params.get(name, default)

class MockLogger:
    def info(self, msg): pass
    def debug(self, msg): pass
    def warning(self, msg): pass

class TestCalculateArea:
    def test_basic_area_calculation(self):
        """测试基本面积计算"""
        # 创建已知面积的多边形（投影坐标系）
        polygons = [box(0, 0, 100, 100)]  # 100x100 = 10000 平方米
        gdf = gpd.GeoDataFrame(geometry=polygons, crs="EPSG:3857")

        ctx = MockStepContext(
            input_data=gdf,
            params={"unit": "sqm", "column_name": "area"},
        )

        result = calculate_area(ctx)

        assert "area" in result.output.columns
        assert result.stats["feature_count"] == 1
        assert result.stats["unit"] == "sqm"

    def test_hectare_unit(self):
        """测试公顷单位转换"""
        polygons = [box(0, 0, 100, 100)]  # 10000 sqm = 1 hectare
        gdf = gpd.GeoDataFrame(geometry=polygons, crs="EPSG:3857")

        ctx = MockStepContext(
            input_data=gdf,
            params={"unit": "hectare", "column_name": "area_ha"},
        )

        result = calculate_area(ctx)

        assert "area_ha" in result.output.columns
        assert result.stats["unit"] == "hectare"

    def test_custom_column_name(self):
        """测试自定义列名"""
        polygons = [box(0, 0, 10, 10)]
        gdf = gpd.GeoDataFrame(geometry=polygons, crs="EPSG:3857")

        ctx = MockStepContext(
            input_data=gdf,
            params={"unit": "sqm", "column_name": "custom_area"},
        )

        result = calculate_area(ctx)

        assert "custom_area" in result.output.columns

    def test_stats_output(self):
        """测试统计信息输出"""
        polygons = [box(0, 0, 10, 10), box(0, 0, 20, 20)]
        gdf = gpd.GeoDataFrame(geometry=polygons, crs="EPSG:3857")

        ctx = MockStepContext(
            input_data=gdf,
            params={"unit": "sqm", "column_name": "area"},
        )

        result = calculate_area(ctx)

        assert "min_area" in result.stats
        assert "max_area" in result.stats
        assert "mean_area" in result.stats
        assert "total_area" in result.stats
        assert result.stats["feature_count"] == 2

@step(...)
def my_step(ctx: StepContext) -> StepResult:
    gdf = ctx.input("input")

    # ✓ 在步骤开始时验证输入
    if gdf is None or len(gdf) == 0:
        raise ValueError("输入数据为空")

    column = ctx.param("column")
    if column not in gdf.columns:
        available = ", ".join(gdf.columns.tolist())
        raise ValueError(
            f"列 '{column}' 不存在。可用列: {available}"
        )

    # ✓ 使用日志记录关键信息
    ctx.logger.info(f"处理 {len(gdf)} 条要素")

    # ✓ 不修改原始数据
    result = gdf.copy()
    # 处理逻辑...

    return StepResult(output=result, stats={...})

# ✓ 好的统计信息
stats = {
    "feature_count": 1024,      # 整数类型
    "pass_rate": 0.95,          # 浮点数 0-1 范围
    "total_area": 12345.67,     # 数值类型
    "has_errors": False,        # 布尔类型
    "crs": "EPSG:4326",        # 字符串类型
}

# ✗ 避免在统计中放入大数据
stats = {
    "all_data": gdf.to_dict(),  # 不要放完整数据！
}

# ✓ 正确：复制后修改
result = gdf.copy()
result["new_col"] = values

# ✗ 错误：直接修改输入
gdf["new_col"] = values  # 会影响上游步骤的输出！