api

lib.prospecting — 勘探/储量模块

BlockDataError

Bases: RuntimeError

块段模型操作异常基类

Source code in dimine_python_sdk\lib\prospecting\block_model.py

class BlockDataError(RuntimeError):
    """块段模型操作异常基类"""
    pass

BlockModelCommonParams

Bases: ParamModel

块段模型估值通用参数基类（距离幂/克里格共用）

Source code in dimine_python_sdk\lib\prospecting\models.py

class BlockModelCommonParams(ParamModel):
    """块段模型估值通用参数基类（距离幂/克里格共用）"""

    sample_file: str = Field(..., description="dmg样品数据文件路径")
    variables: str = Field(..., alias="variable", description="估值变量即元素列表字段，多个元素用逗号隔开，元素即对应钻孔数据库中元素名称")
    extra_attributes: str = Field(..., alias="extra_attribute", description="额外属性字段（如到样品点平均距离、到样品点最近距离、样品点数目、工程数目等）,支持输入多个属性名称，多个属性用逗号隔开")
    min_value: Optional[float] = Field(default=0.0001, description="估值最小值")
    max_value: Optional[float] = Field(default=9999, description="估值最大值")
    single_block_min: Optional[float] = Field(default=4, description="单块最小尺寸")
    single_block_max: Optional[float] = Field(default=8, description="单块最大尺寸")
    sub_block_main: Optional[int] = Field(default=1, description="主轴离散块数，在计算某个块品位储量时，可以先分割成多个子块用于更精确计算。")
    sub_block_second: Optional[int] = Field(default=1, description="次轴离散块数，在计算某个块品位储量时，可以先分割成多个子块用于更精确计算。")
    sub_block_short: Optional[int] = Field(default=1, description="短轴离散块数，在计算某个块品位储量时，可以先分割成多个子块用于更精确计算。")
    angle_main: Optional[float] = Field(default=10, description="椭球体参数：搜索角度：主轴方向角度（度）")
    angle_second: Optional[float] = Field(default=0, description="椭球体参数：搜索角度：次轴方向角度（度）")
    angle_short: Optional[float] = Field(default=15, description="椭球体参数：搜索角度：短方向角度（度）")
    main_radius: Optional[float] = Field(default=1536, gt=0, description="椭球体参数：主轴半径。主方向搜索半径,单位m")
    second_main_rate: Optional[float] = Field(
        default=0.8, ge=0, le=1, description="椭球体参数：次方向搜索半径比例（相对于主方向）即：次轴/主轴 的比例，根据主轴半径可以算出次轴半径"
    )
    short_main_rate: Optional[float] = Field(
        default=0.2, ge=0, le=1, description="椭球体参数：短方向搜索半径比例（相对于主方向）即：短轴/主轴 的比例，根据主轴半径可以算出短轴半径"
    )
    octant_max: Optional[int] = Field(default=0, ge=0, description="八分圆最大值")
    project_count_min: Optional[int] = Field(default=0, ge=0, description="工程数最小值")
    single_project_sample_max: Optional[int] = Field(default=0, ge=0, description="单工程样品数最大值")

BlockModelCreateError

Bases: BlockDataError

创建块段模型失败

Source code in dimine_python_sdk\lib\prospecting\block_model.py

class BlockModelCreateError(BlockDataError):
    """创建块段模型失败"""
    pass

BlockModelDistancePowerParams

Bases: BlockModelCommonParams

距离幂估值专属参数类（继承通用参数+新增距离幂特有参数）

Source code in dimine_python_sdk\lib\prospecting\models.py

class BlockModelDistancePowerParams(BlockModelCommonParams):
    """距离幂估值专属参数类（继承通用参数+新增距离幂特有参数）"""

    power: Optional[int] = Field(default=2, description="距离幂次（如 2）")

BlockModelEvaluator

块段模型估值与储量计算器

所有方法接受块段模型文件路径，失败时抛出自定义异常（不返回 tuple）：

evaluator = BlockModelEvaluator()
result = evaluator.distance_power_evaluation(
    "model.dmb", constraint, idw_params
)
# result -> {"success": True, "message": "..."}

Raises:

Type	Description
BlockModelEvaluationError	估值失败
ReservesCalculateError	储量计算失败

Source code in dimine_python_sdk\lib\prospecting\block_model.py

class BlockModelEvaluator:
    """
    块段模型估值与储量计算器

    所有方法接受块段模型文件路径，失败时抛出自定义异常（不返回 tuple）：

        evaluator = BlockModelEvaluator()
        result = evaluator.distance_power_evaluation(
            "model.dmb", constraint, idw_params
        )
        # result -> {"success": True, "message": "..."}

    Raises:
        BlockModelEvaluationError: 估值失败
        ReservesCalculateError: 储量计算失败
    """

    @staticmethod
    def distance_power_evaluation(
        block_model_file: str,
        constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
        evaluation_params: BlockModelDistancePowerParams,
        overwrite_result: bool = True,
    ) -> dict:
        """距离幂估值计算"""
        constraint_json = _serialize_constraints(constraint_params)
        evaluation_json = json.dumps(evaluation_params.to_dict(), ensure_ascii=False)
        try:
            return _native_distance_power_evaluation(
                block_model_file, constraint_json, evaluation_json, overwrite_result
            )
        except Exception as exc:
            raise BlockModelEvaluationError(f"距离幂估值失败: {exc}") from exc

    @staticmethod
    def kriging_evaluation(
        block_model_file: str,
        constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
        evaluation_params: BlockModelKrigingParams,
        overwrite_result: bool = True,
    ) -> dict:
        """克里格估值计算"""
        constraint_json = _serialize_constraints(constraint_params)
        evaluation_json = json.dumps(evaluation_params.to_dict(), ensure_ascii=False)
        try:
            return _native_kriging_evaluation(
                block_model_file, constraint_json, evaluation_json, overwrite_result
            )
        except Exception as exc:
            raise BlockModelEvaluationError(f"克里格估值失败: {exc}") from exc

    @staticmethod
    def calculate_reserves(
        block_model_file: str,
        constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
        reserve_params: ReservesCalculateParam,
    ) -> dict:
        """储量计算"""
        constraint_json = _serialize_constraints(constraint_params)
        reserve_json = json.dumps(reserve_params.to_dict(), ensure_ascii=False)
        try:
            return _native_calculate_reserves(
                block_model_file, constraint_json, reserve_json
            )
        except Exception as exc:
            raise ReservesCalculateError(f"储量计算失败: {exc}") from exc

calculate_reserves(block_model_file, constraint_params, reserve_params) staticmethod

储量计算

Source code in dimine_python_sdk\lib\prospecting\block_model.py

@staticmethod
def calculate_reserves(
    block_model_file: str,
    constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
    reserve_params: ReservesCalculateParam,
) -> dict:
    """储量计算"""
    constraint_json = _serialize_constraints(constraint_params)
    reserve_json = json.dumps(reserve_params.to_dict(), ensure_ascii=False)
    try:
        return _native_calculate_reserves(
            block_model_file, constraint_json, reserve_json
        )
    except Exception as exc:
        raise ReservesCalculateError(f"储量计算失败: {exc}") from exc

distance_power_evaluation(block_model_file, constraint_params, evaluation_params, overwrite_result=True) staticmethod

距离幂估值计算

Source code in dimine_python_sdk\lib\prospecting\block_model.py

@staticmethod
def distance_power_evaluation(
    block_model_file: str,
    constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
    evaluation_params: BlockModelDistancePowerParams,
    overwrite_result: bool = True,
) -> dict:
    """距离幂估值计算"""
    constraint_json = _serialize_constraints(constraint_params)
    evaluation_json = json.dumps(evaluation_params.to_dict(), ensure_ascii=False)
    try:
        return _native_distance_power_evaluation(
            block_model_file, constraint_json, evaluation_json, overwrite_result
        )
    except Exception as exc:
        raise BlockModelEvaluationError(f"距离幂估值失败: {exc}") from exc

kriging_evaluation(block_model_file, constraint_params, evaluation_params, overwrite_result=True) staticmethod

克里格估值计算

Source code in dimine_python_sdk\lib\prospecting\block_model.py

@staticmethod
def kriging_evaluation(
    block_model_file: str,
    constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
    evaluation_params: BlockModelKrigingParams,
    overwrite_result: bool = True,
) -> dict:
    """克里格估值计算"""
    constraint_json = _serialize_constraints(constraint_params)
    evaluation_json = json.dumps(evaluation_params.to_dict(), ensure_ascii=False)
    try:
        return _native_kriging_evaluation(
            block_model_file, constraint_json, evaluation_json, overwrite_result
        )
    except Exception as exc:
        raise BlockModelEvaluationError(f"克里格估值失败: {exc}") from exc

BlockModelKrigingParams

Bases: BlockModelCommonParams

克里格估值专属参数类（继承通用参数+新增克里格特有参数）

Source code in dimine_python_sdk\lib\prospecting\models.py

class BlockModelKrigingParams(BlockModelCommonParams):
    """克里格估值专属参数类（继承通用参数+新增克里格特有参数）"""

    krig_type: Literal[0,1,2] = Field(default=1, description="克里格类型，0-简单克里格，1-普通克里格，2-对数正太克里格")
    sk: Optional[int] = Field(default=0, description="克里格相关参数")
    c0: Optional[int] = Field(default=0, description="块金值")
    variational_function: list = Field(
        default_factory=lambda: [{"type": 0, "range": 22, "cc": 0.162586}],
        description="变差函数参数列表",
    )

BlockModelLoadError

Bases: BlockDataError

加载块段模型失败

Source code in dimine_python_sdk\lib\prospecting\block_model.py

class BlockModelLoadError(BlockDataError):
    """加载块段模型失败"""
    pass

CreateBlockModelParams

Bases: ParamModel

创建空块段模型参数，即 注意：为什么没有输入文件，因为这只是创建空块段模型，只需要矿体包围盒范围即可

Source code in dimine_python_sdk\lib\prospecting\models.py

class CreateBlockModelParams(ParamModel):
    """创建空块段模型参数，即
    注意：为什么没有输入文件，因为这只是创建空块段模型，只需要矿体包围盒范围即可"""

    file_name: str = Field(..., description="输出块段模型文件路径，块段模型是dmb文件")
    origin_x: float = Field(..., description="起点 X，即包围盒的X最小点")
    origin_y: float = Field(..., description="起点 Y，即包围盒的Y最小点")
    origin_z: float = Field(..., description="起点 Z，即包围盒的Z最小点")
    rotate_x: Optional[float] = Field(default=0, description="绕 X 轴旋转角度")
    rotate_y: Optional[float] = Field(default=0, description="绕 Y 轴旋转角度")
    rotate_z: Optional[float] = Field(default=0, description="绕 Z 轴旋转角度")
    size_x: Optional[float] = Field(default=1, description="单元块 X 尺寸")
    size_y: Optional[float] = Field(default=1, description="单元块 Y 尺寸")
    size_z: Optional[float] = Field(default=1, description="单元块 Z 尺寸")
    length_x: float = Field(..., description="X 方向延伸长度，即包围盒的X方向长度")
    length_y: float = Field(..., description="Y 方向延伸长度，即包围盒的Y方向长度")
    length_z: float = Field(..., description="Z 方向延伸长度，即包围盒的Z方向长度")
    field_definition: list[FieldDefinition] = Field(
        default_factory=lambda: [
            FieldDefinition(field_name="岩石类型", field_type="字节型"),
            FieldDefinition(field_name="体重", field_type="浮点型"),
        ],
        description="块段属性定义清单，每个块段的属性字段名称和类型",
    )

DTMConstraint

Bases: ParamModel

DTM 约束 (type=1)：基于 DTM 模型的上部/下部空间约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class DTMConstraint(ParamModel):
    """DTM 约束 (type=1)：基于 DTM 模型的上部/下部空间约束"""

    type: Literal[1] = Field(default=1, description="DTM约束类型")
    file: str = Field(..., description="DTM 文件路径")
    range: int = Field(default=0, description="约束范围，0=上部，1=下部")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

DmBlockData

块段模型管理器

管理单个块段模型的加载与属性访问。所有几何数据属性直接返回， 无需 getter 前缀：

block = DmBlockData("model.dmb")
origin = block.origin           # np.ndarray
xyz_len = block.xyz_length      # list[float]
max_level = block.max_level     # int
fc = block.field_definitions    # list[dict]
total = block.total_block_count # int

# 遍历块段数据
for data in block.iter_blocks("品位"):
    print(data)

静态方法（无需实例化）： result = DmBlockData.create_block_model(params) DmBlockData.block_constrain_to_dmc_file(block_file, constraint, output)

Source code in dimine_python_sdk\lib\prospecting\block_model.py

class DmBlockData:
    """
    块段模型管理器

    管理单个块段模型的加载与属性访问。所有几何数据属性直接返回，
    无需 getter 前缀：

        block = DmBlockData("model.dmb")
        origin = block.origin           # np.ndarray
        xyz_len = block.xyz_length      # list[float]
        max_level = block.max_level     # int
        fc = block.field_definitions    # list[dict]
        total = block.total_block_count # int

        # 遍历块段数据
        for data in block.iter_blocks("品位"):
            print(data)

    静态方法（无需实例化）：
        result = DmBlockData.create_block_model(params)
        DmBlockData.block_constrain_to_dmc_file(block_file, constraint, output)
    """

    def __init__(self, file_path: str | None = None):
        """
        Args:
            file_path: 块段模型文件路径。为 None 时需手动调用 open()
        """
        self._session: BlockModelSession | None = None
        if file_path is not None:
            self.open(file_path)

    def open(self, file_path: str) -> None:
        """
        加载块段模型文件

        Raises:
            BlockModelLoadError: 加载失败时抛出
        """
        try:
            self._session = BlockModelSession.open(file_path)
        except Exception as exc:
            raise BlockModelLoadError(f"加载块段模型失败: {file_path}") from exc

    # ------------------------------------------------------------------
    # 属性（只读）
    # ------------------------------------------------------------------
    @property
    def origin(self):  # -> np.ndarray:
        """模型原点坐标"""
        self._check_open()

        origin = self._session.origin  # np.ndarray
        return origin

    @property
    def xyz_length(self) -> list:
        """模型 X/Y/Z 方向总长度"""
        self._check_open()
        return list(self._session.xyz_length)

    @property
    def max_level(self) -> int:
        """模型最大精度层级"""
        self._check_open()
        return self._session.max_level

    @property
    def min_size(self) -> list:
        """最小块段尺寸"""
        self._check_open()
        return list(self._session.min_size)

    @property
    def rotation(self) -> list:
        """模型旋转角度"""
        self._check_open()
        return list(self._session.rotation)

    @property
    def field_definitions(self) -> list:
        """字段定义列表"""
        self._check_open()
        return [
            {"name": d.name, "type": d.type, "size": d.size}
            for d in self._session.field_definitions
        ]

    @property
    def total_block_count(self) -> int:
        """块段总数量"""
        self._check_open()
        return self._session.total_block_count

    # ------------------------------------------------------------------
    # 迭代器
    # ------------------------------------------------------------------
    def iter_blocks(self, field_name: str) -> Generator[dict, None, None]:
        """遍历块段模型的所有块段数据"""
        self._check_open()
        return self._session.iter_blocks(field_name)

    # ------------------------------------------------------------------
    # 内部
    # ------------------------------------------------------------------
    def _check_open(self) -> None:
        if self._session is None:
            raise BlockDataError("请先调用 open() 加载块段模型")

    # ------------------------------------------------------------------
    # 静态方法
    # ------------------------------------------------------------------
    @staticmethod
    def create_block_model(params: CreateBlockModelParams) -> dict:
        """
        创建空块段模型

        Args:
            params: CreateBlockModelParams 参数实例

        Returns:
            dict: 创建结果 {"state": "ok", ...}

        Raises:
            BlockModelCreateError: 创建失败时抛出
        """
        json_param = json.dumps(params.to_dict(), ensure_ascii=False)
        try:
            response = _native_create_block_model(json_param)
        except Exception as exc:
            raise BlockModelCreateError(f"创建块段模型失败: {exc}") from exc
        if response.get("state") == "failed":
            raise BlockModelCreateError(f"创建块段模型失败: {response.get('message')}")
        return response

    @staticmethod
    def block_constrain_to_dmc_file(
        block_model_file: str,
        constrain_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
        constrain_result_file: str,
    ) -> bool:
        """
        块约束保存到约束结果 dmc 文件

        Args:
            block_model_file: 块段模型文件路径
            constrain_params: 约束参数
            constrain_result_file: 约束结果文件路径

        Returns:
            bool
        """
        json_param = _serialize_constraints(constrain_params)
        try:
            _native_block_constrain_to_dmc_file(
                block_model_file, json_param, constrain_result_file
            )
        except Exception as exc:
            raise BlockDataError(
                f"块约束保存失败: 源文件={block_model_file}, 约束文件={constrain_result_file}"
            ) from exc
        return True

field_definitions property

字段定义列表

max_level property

模型最大精度层级

min_size property

最小块段尺寸

origin property

模型原点坐标

rotation property

模型旋转角度

total_block_count property

块段总数量

xyz_length property

模型 X/Y/Z 方向总长度

__init__(file_path=None)

Parameters:

Name	Type	Description	Default
file_path	str | None	块段模型文件路径。为 None 时需手动调用 open()	None

Source code in dimine_python_sdk\lib\prospecting\block_model.py

def __init__(self, file_path: str | None = None):
    """
    Args:
        file_path: 块段模型文件路径。为 None 时需手动调用 open()
    """
    self._session: BlockModelSession | None = None
    if file_path is not None:
        self.open(file_path)

block_constrain_to_dmc_file(block_model_file, constrain_params, constrain_result_file) staticmethod

块约束保存到约束结果 dmc 文件

Parameters:

Name	Type	Description	Default
block_model_file	str	块段模型文件路径	required
constrain_params	list[BlockModelConstraintItem] | BlockModelConstraintItem	约束参数	required
constrain_result_file	str	约束结果文件路径	required

Returns:

Type	Description
bool	bool

Source code in dimine_python_sdk\lib\prospecting\block_model.py

@staticmethod
def block_constrain_to_dmc_file(
    block_model_file: str,
    constrain_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
    constrain_result_file: str,
) -> bool:
    """
    块约束保存到约束结果 dmc 文件

    Args:
        block_model_file: 块段模型文件路径
        constrain_params: 约束参数
        constrain_result_file: 约束结果文件路径

    Returns:
        bool
    """
    json_param = _serialize_constraints(constrain_params)
    try:
        _native_block_constrain_to_dmc_file(
            block_model_file, json_param, constrain_result_file
        )
    except Exception as exc:
        raise BlockDataError(
            f"块约束保存失败: 源文件={block_model_file}, 约束文件={constrain_result_file}"
        ) from exc
    return True

create_block_model(params) staticmethod

创建空块段模型

Parameters:

Name	Type	Description	Default
params	CreateBlockModelParams	CreateBlockModelParams 参数实例	required

Returns:

Name	Type	Description
dict	dict	创建结果 {"state": "ok", ...}

Raises:

Type	Description
BlockModelCreateError	创建失败时抛出

Source code in dimine_python_sdk\lib\prospecting\block_model.py

@staticmethod
def create_block_model(params: CreateBlockModelParams) -> dict:
    """
    创建空块段模型

    Args:
        params: CreateBlockModelParams 参数实例

    Returns:
        dict: 创建结果 {"state": "ok", ...}

    Raises:
        BlockModelCreateError: 创建失败时抛出
    """
    json_param = json.dumps(params.to_dict(), ensure_ascii=False)
    try:
        response = _native_create_block_model(json_param)
    except Exception as exc:
        raise BlockModelCreateError(f"创建块段模型失败: {exc}") from exc
    if response.get("state") == "failed":
        raise BlockModelCreateError(f"创建块段模型失败: {response.get('message')}")
    return response

iter_blocks(field_name)

遍历块段模型的所有块段数据

Source code in dimine_python_sdk\lib\prospecting\block_model.py

def iter_blocks(self, field_name: str) -> Generator[dict, None, None]:
    """遍历块段模型的所有块段数据"""
    self._check_open()
    return self._session.iter_blocks(field_name)

open(file_path)

加载块段模型文件

Raises:

Type	Description
BlockModelLoadError	加载失败时抛出

Source code in dimine_python_sdk\lib\prospecting\block_model.py

def open(self, file_path: str) -> None:
    """
    加载块段模型文件

    Raises:
        BlockModelLoadError: 加载失败时抛出
    """
    try:
        self._session = BlockModelSession.open(file_path)
    except Exception as exc:
        raise BlockModelLoadError(f"加载块段模型失败: {file_path}") from exc

DrillDBError

Bases: RuntimeError

钻孔数据库操作异常基类

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

class DrillDBError(RuntimeError):
    """钻孔数据库操作异常基类"""
    pass

DrillDBLoadError

Bases: DrillDBError

加载钻孔数据库失败

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

class DrillDBLoadError(DrillDBError):
    """加载钻孔数据库失败"""
    pass

DrillDBManager

钻孔数据库管理器（业务层新版）

管理单个钻孔数据库的加载和操作。 所有表访问返回 pandas DataFrame，列名为中文标准格式：

db = DrillDBManager("钻孔数据.dmd")
db.collar['工程号']     # 取整列（Series）
db.collar.iloc[0]         # 取单行
len(db.collar)            # 记录数
db.collar.columns         # 字段名列表（中文）
df = db.collar            # 直接获得 pd.DataFrame

标准输入格式： COLLAR_COLUMN_MAP — 孔口表中文→英文列名映射 SURVEY_COLUMN_MAP — 测斜表中文→英文列名映射 LITHOLOGY_COLUMN_MAP — 岩性表中文→英文列名映射 SAMPLE_COLUMN_MAP — 样品表中文→英文列名映射

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

class DrillDBManager:
    """
    钻孔数据库管理器（业务层新版）

    管理单个钻孔数据库的加载和操作。
    所有表访问返回 pandas DataFrame，列名为中文标准格式：

        db = DrillDBManager("钻孔数据.dmd")
        db.collar['工程号']     # 取整列（Series）
        db.collar.iloc[0]         # 取单行
        len(db.collar)            # 记录数
        db.collar.columns         # 字段名列表（中文）
        df = db.collar            # 直接获得 pd.DataFrame

    标准输入格式：
        COLLAR_COLUMN_MAP    — 孔口表中文→英文列名映射
        SURVEY_COLUMN_MAP    — 测斜表中文→英文列名映射
        LITHOLOGY_COLUMN_MAP — 岩性表中文→英文列名映射
        SAMPLE_COLUMN_MAP    — 样品表中文→英文列名映射
    """

    COLLAR_COLUMN_MAP = COLLAR_COLUMN_MAP
    SURVEY_COLUMN_MAP = SURVEY_COLUMN_MAP
    LITHOLOGY_COLUMN_MAP = LITHOLOGY_COLUMN_MAP
    SAMPLE_COLUMN_MAP = SAMPLE_COLUMN_MAP

    def __init__(
        self, file_path: Optional[str] = None, base_path: Optional[str] = None
    ):
        """
        初始化钻孔数据库管理器

        Args:
            file_path: dmd 文件路径（可以是相对路径）。为 None 时创建空对象，
                      需通过 setter 赋值 DataFrame 后再 save。
            base_path: 基础路径，用于相对路径解析
        """
        self.file_path: Optional[str] = None
        self._session: Optional[_NativeDrillSession] = None
        self._ref_dict: Optional[dict] = None
        self._collar_df: Optional["pd.DataFrame"] = None
        self._survey_df: Optional["pd.DataFrame"] = None
        self._lithology_df: Optional["pd.DataFrame"] = None
        self._sample_df: Optional["pd.DataFrame"] = None

        if file_path is not None:
            self.load(file_path, base_path=base_path)
        else:
            self._session = _NativeDrillSession.create_empty()

    # ------------------------------------------------------------------
    # 内部缓存管理
    # ------------------------------------------------------------------
    def _invalidate_cache(self):
        """使缓存的 DataFrame 失效"""
        self._collar_df = None
        self._survey_df = None
        self._lithology_df = None
        self._sample_df = None

    # ------------------------------------------------------------------
    # 生命周期
    # ------------------------------------------------------------------
    def load(self, file_path: str, base_path: Optional[str] = None) -> None:
        """从文件加载钻孔数据库"""
        if base_path:
            full_path = Path(base_path) / file_path
            if not full_path.exists():
                full_path = Path(file_path)
            self.file_path = str(full_path)
        else:
            self.file_path = file_path

        try:
            self._session = _NativeDrillSession.load(self.file_path)
        except Exception as exc:
            raise DrillDBLoadError(f"加载钻孔数据库失败: {self.file_path}") from exc
        self._ref_dict = self._session.ref_dict
        self._invalidate_cache()

    def reload(self):
        """重新加载文件"""
        if not self.file_path:
            raise DrillDBError("未指定文件路径，无法重新加载")
        try:
            self._session = _NativeDrillSession.load(self.file_path)
        except Exception as exc:
            raise DrillDBLoadError(f"加载钻孔数据库失败: {self.file_path}") from exc
        self._ref_dict = self._session.ref_dict
        self._invalidate_cache()

    def save(
        self,
        output_path: Optional[str] = None,
        extend_tables: Optional[List[Dict[str, Any]]] = None,
    ) -> None:
        """
        保存钻孔数据库

        完全走 Python 层，不调用 C++ geo_drill_save：
            1. 孔口表   -> {stem}_collar.dmt
            2. 斜测表   -> {stem}_survey.dmt
            3. 岩性表   -> {stem}_lithology.dmt
            4. 样品表   -> {stem}_sample.dmt（可选）
            5. 扩展表   -> {ext.dmt_filename}（可选，用于隐式建模等）
            6. 引用关系 -> {stem}.dmd

        Parameters
        ----------
        output_path : str, optional
            保存路径
        extend_tables : List[Dict], optional
            扩展表列表，每个 dict 含 df / dmt_filename / field_list
        """
        save_path = output_path or self.file_path
        if not save_path:
            raise DrillDBError(
                "未指定保存路径，请传入 output_path 或在初始化/加载时设置 file_path"
            )

        if not str(save_path).lower().endswith(".dmd"):
            raise DrillDBError("钻孔数据库只能保存成dmd文件")

        save_drill_database(
            self.collar,
            self.survey,
            self.lithology,
            str(save_path),
            sample_df=self.sample,
            extend_tables=extend_tables,
        )

    def save_as_dmg(self, output_path: Union[str, Path]) -> None:
        """
        将钻孔数据库保存为 .dmg 可视化格式。

        先将内存中四个缓存表同步回 native session，再调用 native save_as_dmg。

        Args:
            output_path: 目标 .dmg 文件路径（无扩展名时自动补全）

        Raises:
            DrillDBSaveError: 保存失败时抛出
        """
        if self._session is None:
            raise DrillDBSaveError("钻孔数据库未加载或已关闭")

        save_path = _normalize_dmg_path(output_path)
        try:
            self._session.set_collar_table(self.collar)
            self._session.set_survey_table(self.survey)
            self._session.set_lithology_table(self.lithology)
            self._session.set_sample_table(self.sample)
            self._session.save_as_dmg(save_path)
        except Exception as exc:
            raise DrillDBSaveError(f"保存 .dmg 钻孔数据库失败: {save_path}") from exc

    def close(self):
        """释放资源"""
        self._invalidate_cache()
        self._ref_dict = None
        self._session = None

    # ------------------------------------------------------------------
    # 表访问（DataFrame 化）
    # ------------------------------------------------------------------
    @property
    def collar(self) -> "pd.DataFrame":
        """钻孔孔口表（pd.DataFrame）"""
        if self._collar_df is None:
            self._collar_df = self._session.collar_table()
        return self._collar_df

    @collar.setter
    def collar(self, df: "pd.DataFrame") -> None:
        """设置钻孔孔口表（仅更新内存缓存）"""
        _validate_columns(df, COLLAR_COLUMN_MAP, "孔口表", required=_COLLAR_REQUIRED)
        self._collar_df = df

    @property
    def survey(self) -> "pd.DataFrame":
        """钻孔测斜表（pd.DataFrame）"""
        if self._survey_df is None:
            self._survey_df = self._session.survey_table()
        return self._survey_df

    @survey.setter
    def survey(self, df: "pd.DataFrame") -> None:
        """设置钻孔测斜表（仅更新内存缓存）"""
        _validate_columns(df, SURVEY_COLUMN_MAP, "测斜表", required=_SURVEY_REQUIRED)
        self._survey_df = df

    @property
    def lithology(self) -> "pd.DataFrame":
        """钻孔岩性表（pd.DataFrame）"""
        if self._lithology_df is None:
            self._lithology_df = self._session.lithology_table()
        return self._lithology_df

    @lithology.setter
    def lithology(self, df: "pd.DataFrame") -> None:
        """设置钻孔岩性表（仅更新内存缓存）"""
        _validate_columns(
            df, LITHOLOGY_COLUMN_MAP, "岩性表", required=_LITHOLOGY_REQUIRED
        )
        self._lithology_df = df

    @property
    def sample(self) -> "pd.DataFrame":
        """钻孔样品表（pd.DataFrame）"""
        if self._sample_df is None:
            self._sample_df = self._session.sample_table()
        return self._sample_df

    @sample.setter
    def sample(self, df: "pd.DataFrame") -> None:
        """设置钻孔样品表（仅更新内存缓存）"""
        _validate_columns(df, SAMPLE_COLUMN_MAP, "样品表", required=_SAMPLE_REQUIRED)
        self._sample_df = df

    def __getitem__(self, table_name: str) -> "pd.DataFrame":
        """统一表访问：db['collar'] / db['survey'] / db['lithology'] / db['sample']"""
        if table_name == "collar":
            return self.collar
        if table_name == "survey":
            return self.survey
        if table_name == "lithology":
            return self.lithology
        if table_name == "sample":
            return self.sample
        raise KeyError(
            f"未知表: {table_name}。支持的表: 'collar', 'survey', 'lithology', 'sample'"
        )

    # ------------------------------------------------------------------
    # 钻孔处理功能
    # ------------------------------------------------------------------
    def process_samples(self, params: SampleLengthCombineParam) -> dict:
        """处理样长组合（C++ 接口）"""
        return DrillFunctionWrapper.sample_length_combine(params)

    def process_steps(self, params: StepCombineParam) -> dict:
        """处理台阶组合（C++ 接口）"""
        return DrillFunctionWrapper.step_combine(params)

    def process_high_grade(self, params: HighGradeProcessParam) -> dict:
        """处理特高品位（调用 C++ extra_high_grade_process 接口）"""
        return DrillFunctionWrapper.extra_high_grade_process(params)

    def process_high_grade_samples(
        self,
        grade_fields: List[str],
        group_mode: str = "by_hole",
        average_multiple: Optional[int] = None,
        use_frequency: bool = False,
        frequency: float = 0.95,
        replace_method: int = 2,
        max_depth_gap: float = 0.5,
        keep_raw_grades: bool = True,
    ):
        """特高品位处理（纯 Python/pandas 实现，不依赖 C++ 接口）

        对多个品位字段依次执行特高品位识别与替换，返回处理后的完整样品表。

        Args:
            grade_fields: 品位字段列表，如 ["TFe", "TiO2"]
            group_mode: 分组模式 ("by_hole" / "by_section" / "global")
            average_multiple: 平均值倍数，None=按CV自动判定6/7/8
            use_frequency: 启用累积频率法
            frequency: 累积频率阈值 (0.95 / 0.975)
            replace_method: 替换方式 (0=剔除 1=给定值 2=相邻平均 3=矿体平均 4=单一工程 5=截止品位)
            max_depth_gap: 深度连续性容差(m)
            keep_raw_grades: 是否保留原始品位为 {field}_raw 列

        Returns:
            (processed_df, summary) 元组:
                processed_df — 处理后的 DataFrame（每字段有 {field}_processed 列）
                summary     — {"特高样品总数": N, "TFe": {...}, "TiO2": {...}, "分组模式": ..., "品位字段": [...]}
        """
        processed_df = self.sample.copy()
        total_flagged = 0
        per_field = {}

        for gf in grade_fields:
            params = HighGradeParams(
                grade_field=gf,
                group_mode=group_mode,
                average_multiple=average_multiple,
                use_frequency=use_frequency,
                frequency=frequency,
                replace_method=replace_method,
                max_depth_gap=max_depth_gap,
            )
            df, hg_sum = HighGradeProcessor(params).process(
                processed_df, collar_df=self.collar
            )
            if keep_raw_grades:
                processed_df[f"{gf}_raw"] = processed_df[gf].copy()
            processed_df[gf] = df[f"{gf}_processed"]
            processed_df[f"{gf}_processed"] = df[f"{gf}_processed"]
            total_flagged += hg_sum["特高样品数"]
            per_field[gf] = hg_sum

        summary = {
            "特高样品总数": total_flagged,
            **per_field,
            "分组模式": group_mode,
            "品位字段": grade_fields,
        }
        return processed_df, summary

    def composite_samples(
        self,
        composite_length: float = 2.0,
        grade_fields: Optional[List[str]] = None,
        combine_percent: float = 0.75,
        start_mode: str = "first_sample",
        discard_low_coverage: bool = False,
        include_3d_coords: bool = True,
        auto_detect_grade_fields: bool = True,
    ) -> Dict[str, Any]:
        """样长组合（纯 Python/pandas 实现，不依赖 C++ 接口）"""
        # lazy import — 避免模块导入期循环依赖
        from dimine_python_sdk.lib.prospecting.compositing import (
            composite_samples as _composite,
        )

        return _composite(
            self,
            composite_length=composite_length,
            grade_fields=grade_fields,
            combine_percent=combine_percent,
            start_mode=start_mode,
            discard_low_coverage=discard_low_coverage,
            include_3d_coords=include_3d_coords,
            auto_detect_grade_fields=auto_detect_grade_fields,
        )

    def process_and_composite(
        self,
        grade_fields: List[str],
        group_mode: str = "by_hole",
        average_multiple: Optional[int] = None,
        use_frequency: bool = False,
        frequency: float = 0.95,
        replace_method: int = 2,
        max_depth_gap: float = 0.5,
        keep_raw_grades: bool = True,
        composite_length: float = 2.0,
        combine_percent: float = 0.75,
        start_mode: str = "first_sample",
        discard_low_coverage: bool = False,
        include_3d_coords: bool = True,
    ):
        """特高品位处理 → 样长组合 串联调用"""
        # [1] 特高品位处理
        processed_df, hg_summary = self.process_high_grade_samples(
            grade_fields=grade_fields,
            group_mode=group_mode,
            average_multiple=average_multiple,
            use_frequency=use_frequency,
            frequency=frequency,
            replace_method=replace_method,
            max_depth_gap=max_depth_gap,
            keep_raw_grades=keep_raw_grades,
        )

        # [2] 样长组合 — 直接传入 processed_df，不修改 self.sample
        from dimine_python_sdk.lib.prospecting.compositing import composite_samples as _composite

        composited_df, comp_summary = _composite(
            processed_df,
            composite_length=composite_length,
            grade_fields=grade_fields,
            combine_percent=combine_percent,
            start_mode=start_mode,
            discard_low_coverage=discard_low_coverage,
            include_3d_coords=include_3d_coords,
            auto_detect_grade_fields=False,
            survey_df=self.survey,
            collar_df=self.collar,
        )

        return composited_df, {"high_grade": hg_summary, "compositing": comp_summary}

collar property writable

钻孔孔口表（pd.DataFrame）

lithology property writable

钻孔岩性表（pd.DataFrame）

sample property writable

钻孔样品表（pd.DataFrame）

survey property writable

钻孔测斜表（pd.DataFrame）

__getitem__(table_name)

统一表访问：db['collar'] / db['survey'] / db['lithology'] / db['sample']

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def __getitem__(self, table_name: str) -> "pd.DataFrame":
    """统一表访问：db['collar'] / db['survey'] / db['lithology'] / db['sample']"""
    if table_name == "collar":
        return self.collar
    if table_name == "survey":
        return self.survey
    if table_name == "lithology":
        return self.lithology
    if table_name == "sample":
        return self.sample
    raise KeyError(
        f"未知表: {table_name}。支持的表: 'collar', 'survey', 'lithology', 'sample'"
    )

__init__(file_path=None, base_path=None)

初始化钻孔数据库管理器

Parameters:

Name	Type	Description	Default
file_path	Optional[str]	dmd 文件路径（可以是相对路径）。为 None 时创建空对象， 需通过 setter 赋值 DataFrame 后再 save。	None
base_path	Optional[str]	基础路径，用于相对路径解析	None

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def __init__(
    self, file_path: Optional[str] = None, base_path: Optional[str] = None
):
    """
    初始化钻孔数据库管理器

    Args:
        file_path: dmd 文件路径（可以是相对路径）。为 None 时创建空对象，
                  需通过 setter 赋值 DataFrame 后再 save。
        base_path: 基础路径，用于相对路径解析
    """
    self.file_path: Optional[str] = None
    self._session: Optional[_NativeDrillSession] = None
    self._ref_dict: Optional[dict] = None
    self._collar_df: Optional["pd.DataFrame"] = None
    self._survey_df: Optional["pd.DataFrame"] = None
    self._lithology_df: Optional["pd.DataFrame"] = None
    self._sample_df: Optional["pd.DataFrame"] = None

    if file_path is not None:
        self.load(file_path, base_path=base_path)
    else:
        self._session = _NativeDrillSession.create_empty()

close()

释放资源

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def close(self):
    """释放资源"""
    self._invalidate_cache()
    self._ref_dict = None
    self._session = None

composite_samples(composite_length=2.0, grade_fields=None, combine_percent=0.75, start_mode='first_sample', discard_low_coverage=False, include_3d_coords=True, auto_detect_grade_fields=True)

样长组合（纯 Python/pandas 实现，不依赖 C++ 接口）

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def composite_samples(
    self,
    composite_length: float = 2.0,
    grade_fields: Optional[List[str]] = None,
    combine_percent: float = 0.75,
    start_mode: str = "first_sample",
    discard_low_coverage: bool = False,
    include_3d_coords: bool = True,
    auto_detect_grade_fields: bool = True,
) -> Dict[str, Any]:
    """样长组合（纯 Python/pandas 实现，不依赖 C++ 接口）"""
    # lazy import — 避免模块导入期循环依赖
    from dimine_python_sdk.lib.prospecting.compositing import (
        composite_samples as _composite,
    )

    return _composite(
        self,
        composite_length=composite_length,
        grade_fields=grade_fields,
        combine_percent=combine_percent,
        start_mode=start_mode,
        discard_low_coverage=discard_low_coverage,
        include_3d_coords=include_3d_coords,
        auto_detect_grade_fields=auto_detect_grade_fields,
    )

load(file_path, base_path=None)

从文件加载钻孔数据库

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def load(self, file_path: str, base_path: Optional[str] = None) -> None:
    """从文件加载钻孔数据库"""
    if base_path:
        full_path = Path(base_path) / file_path
        if not full_path.exists():
            full_path = Path(file_path)
        self.file_path = str(full_path)
    else:
        self.file_path = file_path

    try:
        self._session = _NativeDrillSession.load(self.file_path)
    except Exception as exc:
        raise DrillDBLoadError(f"加载钻孔数据库失败: {self.file_path}") from exc
    self._ref_dict = self._session.ref_dict
    self._invalidate_cache()

process_and_composite(grade_fields, group_mode='by_hole', average_multiple=None, use_frequency=False, frequency=0.95, replace_method=2, max_depth_gap=0.5, keep_raw_grades=True, composite_length=2.0, combine_percent=0.75, start_mode='first_sample', discard_low_coverage=False, include_3d_coords=True)

特高品位处理 → 样长组合 串联调用

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def process_and_composite(
    self,
    grade_fields: List[str],
    group_mode: str = "by_hole",
    average_multiple: Optional[int] = None,
    use_frequency: bool = False,
    frequency: float = 0.95,
    replace_method: int = 2,
    max_depth_gap: float = 0.5,
    keep_raw_grades: bool = True,
    composite_length: float = 2.0,
    combine_percent: float = 0.75,
    start_mode: str = "first_sample",
    discard_low_coverage: bool = False,
    include_3d_coords: bool = True,
):
    """特高品位处理 → 样长组合 串联调用"""
    # [1] 特高品位处理
    processed_df, hg_summary = self.process_high_grade_samples(
        grade_fields=grade_fields,
        group_mode=group_mode,
        average_multiple=average_multiple,
        use_frequency=use_frequency,
        frequency=frequency,
        replace_method=replace_method,
        max_depth_gap=max_depth_gap,
        keep_raw_grades=keep_raw_grades,
    )

    # [2] 样长组合 — 直接传入 processed_df，不修改 self.sample
    from dimine_python_sdk.lib.prospecting.compositing import composite_samples as _composite

    composited_df, comp_summary = _composite(
        processed_df,
        composite_length=composite_length,
        grade_fields=grade_fields,
        combine_percent=combine_percent,
        start_mode=start_mode,
        discard_low_coverage=discard_low_coverage,
        include_3d_coords=include_3d_coords,
        auto_detect_grade_fields=False,
        survey_df=self.survey,
        collar_df=self.collar,
    )

    return composited_df, {"high_grade": hg_summary, "compositing": comp_summary}

process_high_grade(params)

处理特高品位（调用 C++ extra_high_grade_process 接口）

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def process_high_grade(self, params: HighGradeProcessParam) -> dict:
    """处理特高品位（调用 C++ extra_high_grade_process 接口）"""
    return DrillFunctionWrapper.extra_high_grade_process(params)

process_high_grade_samples(grade_fields, group_mode='by_hole', average_multiple=None, use_frequency=False, frequency=0.95, replace_method=2, max_depth_gap=0.5, keep_raw_grades=True)

特高品位处理（纯 Python/pandas 实现，不依赖 C++ 接口）

对多个品位字段依次执行特高品位识别与替换，返回处理后的完整样品表。

Parameters:

Name	Type	Description	Default
grade_fields	List[str]	品位字段列表，如 ["TFe", "TiO2"]	required
group_mode	str	分组模式 ("by_hole" / "by_section" / "global")	'by_hole'
average_multiple	Optional[int]	平均值倍数，None=按CV自动判定6/7/8	None
use_frequency	bool	启用累积频率法	False
frequency	float	累积频率阈值 (0.95 / 0.975)	0.95
replace_method	int	替换方式 (0=剔除 1=给定值 2=相邻平均 3=矿体平均 4=单一工程 5=截止品位)	2
max_depth_gap	float	深度连续性容差(m)	0.5
keep_raw_grades	bool	是否保留原始品位为 {field}_raw 列	True

Returns:

Type	Description
	(processed_df, summary) 元组: processed_df — 处理后的 DataFrame（每字段有 {field}_processed 列） summary — {"特高样品总数": N, "TFe": {...}, "TiO2": {...}, "分组模式": ..., "品位字段": [...]}

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def process_high_grade_samples(
    self,
    grade_fields: List[str],
    group_mode: str = "by_hole",
    average_multiple: Optional[int] = None,
    use_frequency: bool = False,
    frequency: float = 0.95,
    replace_method: int = 2,
    max_depth_gap: float = 0.5,
    keep_raw_grades: bool = True,
):
    """特高品位处理（纯 Python/pandas 实现，不依赖 C++ 接口）

    对多个品位字段依次执行特高品位识别与替换，返回处理后的完整样品表。

    Args:
        grade_fields: 品位字段列表，如 ["TFe", "TiO2"]
        group_mode: 分组模式 ("by_hole" / "by_section" / "global")
        average_multiple: 平均值倍数，None=按CV自动判定6/7/8
        use_frequency: 启用累积频率法
        frequency: 累积频率阈值 (0.95 / 0.975)
        replace_method: 替换方式 (0=剔除 1=给定值 2=相邻平均 3=矿体平均 4=单一工程 5=截止品位)
        max_depth_gap: 深度连续性容差(m)
        keep_raw_grades: 是否保留原始品位为 {field}_raw 列

    Returns:
        (processed_df, summary) 元组:
            processed_df — 处理后的 DataFrame（每字段有 {field}_processed 列）
            summary     — {"特高样品总数": N, "TFe": {...}, "TiO2": {...}, "分组模式": ..., "品位字段": [...]}
    """
    processed_df = self.sample.copy()
    total_flagged = 0
    per_field = {}

    for gf in grade_fields:
        params = HighGradeParams(
            grade_field=gf,
            group_mode=group_mode,
            average_multiple=average_multiple,
            use_frequency=use_frequency,
            frequency=frequency,
            replace_method=replace_method,
            max_depth_gap=max_depth_gap,
        )
        df, hg_sum = HighGradeProcessor(params).process(
            processed_df, collar_df=self.collar
        )
        if keep_raw_grades:
            processed_df[f"{gf}_raw"] = processed_df[gf].copy()
        processed_df[gf] = df[f"{gf}_processed"]
        processed_df[f"{gf}_processed"] = df[f"{gf}_processed"]
        total_flagged += hg_sum["特高样品数"]
        per_field[gf] = hg_sum

    summary = {
        "特高样品总数": total_flagged,
        **per_field,
        "分组模式": group_mode,
        "品位字段": grade_fields,
    }
    return processed_df, summary

process_samples(params)

处理样长组合（C++ 接口）

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def process_samples(self, params: SampleLengthCombineParam) -> dict:
    """处理样长组合（C++ 接口）"""
    return DrillFunctionWrapper.sample_length_combine(params)

process_steps(params)

处理台阶组合（C++ 接口）

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def process_steps(self, params: StepCombineParam) -> dict:
    """处理台阶组合（C++ 接口）"""
    return DrillFunctionWrapper.step_combine(params)

reload()

重新加载文件

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def reload(self):
    """重新加载文件"""
    if not self.file_path:
        raise DrillDBError("未指定文件路径，无法重新加载")
    try:
        self._session = _NativeDrillSession.load(self.file_path)
    except Exception as exc:
        raise DrillDBLoadError(f"加载钻孔数据库失败: {self.file_path}") from exc
    self._ref_dict = self._session.ref_dict
    self._invalidate_cache()

save(output_path=None, extend_tables=None)

保存钻孔数据库

完全走 Python 层，不调用 C++ geo_drill_save： 1. 孔口表 -> {stem}_collar.dmt 2. 斜测表 -> {stem}_survey.dmt 3. 岩性表 -> {stem}_lithology.dmt 4. 样品表 -> {stem}_sample.dmt（可选） 5. 扩展表 -> {ext.dmt_filename}（可选，用于隐式建模等） 6. 引用关系 -> {stem}.dmd

Parameters

output_path : str, optional 保存路径 extend_tables : List[Dict], optional 扩展表列表，每个 dict 含 df / dmt_filename / field_list

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def save(
    self,
    output_path: Optional[str] = None,
    extend_tables: Optional[List[Dict[str, Any]]] = None,
) -> None:
    """
    保存钻孔数据库

    完全走 Python 层，不调用 C++ geo_drill_save：
        1. 孔口表   -> {stem}_collar.dmt
        2. 斜测表   -> {stem}_survey.dmt
        3. 岩性表   -> {stem}_lithology.dmt
        4. 样品表   -> {stem}_sample.dmt（可选）
        5. 扩展表   -> {ext.dmt_filename}（可选，用于隐式建模等）
        6. 引用关系 -> {stem}.dmd

    Parameters
    ----------
    output_path : str, optional
        保存路径
    extend_tables : List[Dict], optional
        扩展表列表，每个 dict 含 df / dmt_filename / field_list
    """
    save_path = output_path or self.file_path
    if not save_path:
        raise DrillDBError(
            "未指定保存路径，请传入 output_path 或在初始化/加载时设置 file_path"
        )

    if not str(save_path).lower().endswith(".dmd"):
        raise DrillDBError("钻孔数据库只能保存成dmd文件")

    save_drill_database(
        self.collar,
        self.survey,
        self.lithology,
        str(save_path),
        sample_df=self.sample,
        extend_tables=extend_tables,
    )

save_as_dmg(output_path)

将钻孔数据库保存为 .dmg 可视化格式。

先将内存中四个缓存表同步回 native session，再调用 native save_as_dmg。

Parameters:

Name	Type	Description	Default
output_path	Union[str, Path]	目标 .dmg 文件路径（无扩展名时自动补全）	required

Raises:

Type	Description
DrillDBSaveError	保存失败时抛出

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def save_as_dmg(self, output_path: Union[str, Path]) -> None:
    """
    将钻孔数据库保存为 .dmg 可视化格式。

    先将内存中四个缓存表同步回 native session，再调用 native save_as_dmg。

    Args:
        output_path: 目标 .dmg 文件路径（无扩展名时自动补全）

    Raises:
        DrillDBSaveError: 保存失败时抛出
    """
    if self._session is None:
        raise DrillDBSaveError("钻孔数据库未加载或已关闭")

    save_path = _normalize_dmg_path(output_path)
    try:
        self._session.set_collar_table(self.collar)
        self._session.set_survey_table(self.survey)
        self._session.set_lithology_table(self.lithology)
        self._session.set_sample_table(self.sample)
        self._session.save_as_dmg(save_path)
    except Exception as exc:
        raise DrillDBSaveError(f"保存 .dmg 钻孔数据库失败: {save_path}") from exc

DrillDBProcessError

Bases: DrillDBError

钻孔数据处理失败

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

class DrillDBProcessError(DrillDBError):
    """钻孔数据处理失败"""
    def __init__(self, message: str, response: Optional[dict] = None):
        super().__init__(message)
        self.response = response

DrillDBSaveError

Bases: DrillDBError

保存钻孔数据库失败

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

class DrillDBSaveError(DrillDBError):
    """保存钻孔数据库失败"""
    pass

DrillFunctionWrapper

钻孔相关功能的封装器，提供更简洁的接口

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

class DrillFunctionWrapper:
    """
    钻孔相关功能的封装器，提供更简洁的接口
    """

    @staticmethod
    def sample_length_combine(params: SampleLengthCombineParam) -> dict:
        """样长组合"""
        # combine_percent 传给 C++ 前需 ×100 取整
        if params.combine_percent is not None:
            params.combine_percent = int(params.combine_percent * 100)
        # output_file 未指定时默认使用 {input_file stem}_combined.dmg
        if params.output_file is None:
            stem = Path(params.input_file).stem
            params.output_file = f"{stem}_combined.dmg"
        return sample_length_combine(params.model_dump_json())

    @staticmethod
    def step_combine(params: StepCombineParam) -> dict:
        """台阶组合"""
        if params.output_file is None:
            stem = Path(params.input_file).stem
            params.output_file = f"{stem}_step.dmg"
        return step_combine(params.model_dump_json())

    @staticmethod
    def extra_high_grade_process(params: HighGradeProcessParam) -> dict:
        """特高品位处理"""
        if params.result_field is None:
            params.result_field = f"{params.grade_field}_processed"
        return extra_high_grade_process(params.model_dump_json())

extra_high_grade_process(params) staticmethod

特高品位处理

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

@staticmethod
def extra_high_grade_process(params: HighGradeProcessParam) -> dict:
    """特高品位处理"""
    if params.result_field is None:
        params.result_field = f"{params.grade_field}_processed"
    return extra_high_grade_process(params.model_dump_json())

sample_length_combine(params) staticmethod

样长组合

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

@staticmethod
def sample_length_combine(params: SampleLengthCombineParam) -> dict:
    """样长组合"""
    # combine_percent 传给 C++ 前需 ×100 取整
    if params.combine_percent is not None:
        params.combine_percent = int(params.combine_percent * 100)
    # output_file 未指定时默认使用 {input_file stem}_combined.dmg
    if params.output_file is None:
        stem = Path(params.input_file).stem
        params.output_file = f"{stem}_combined.dmg"
    return sample_length_combine(params.model_dump_json())

step_combine(params) staticmethod

台阶组合

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

@staticmethod
def step_combine(params: StepCombineParam) -> dict:
    """台阶组合"""
    if params.output_file is None:
        stem = Path(params.input_file).stem
        params.output_file = f"{stem}_step.dmg"
    return step_combine(params.model_dump_json())

EntityConstraint

Bases: ParamModel

实体约束 (type=0)：基于实体模型的内部/外部空间约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class EntityConstraint(ParamModel):
    """实体约束 (type=0)：基于实体模型的内部/外部空间约束"""

    type: Literal[0] = Field(default=0, description="实体约束类型")
    file: str = Field(..., description="约束实体dmf文件路径,即矿体模型文件路径")
    range: int = Field(default=0, description="约束范围，0=内部，1=外部")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

FieldDefinition

Bases: ParamModel

块段模型属性字段定义

Source code in dimine_python_sdk\lib\prospecting\models.py

class FieldDefinition(ParamModel):
    """块段模型属性字段定义"""

    field_name: str = Field(..., description="字段名称")
    field_type: Literal["字节型", "整型", "浮点型", "双精度型", "字符串型"] = Field(
        ..., description="字段类型"
    )
    field_length: Optional[int] = Field(default=120, description="字段长度")

HighGradeParams dataclass

特高品位处理参数

Attributes:

Name	Type	Description
grade_field	str	品位字段名（如 "CU"、"TFe"）
group_mode	GroupMode	分组维度 - "by_hole": 按工程号分组 - "by_section": 按勘探线分组（需传入 collar_df） - "global": 全矿区统一计算（默认）
use_average_multiple	bool	启用平均值倍数法
average_multiple	Optional[int]	倍数系数。设为 None（默认）时按品位变化系数(CV)自动判定： CV<0.5→6倍, 0.5≤CV<1.0→7倍, CV≥1.0→8倍。 也可手动指定 6/7/8。
use_frequency	bool	启用累积频率法
frequency	float	累积频率阈值（常用 0.95 / 0.975 / 0.977）
replace_method	ReplaceMethod	替换方式 0 - 剔除法（设为 NaN） 1 - 给定值法 2 - 相邻样品平均值法（默认，长度加权） 3 - 矿体平均品位法 4 - 单一工程法 5 - 截止品位法（cap 到阈值）
assign_value	float	方法 1 的给定值
max_adjacent_search	int	方法 2 相邻搜索最大扩展步数（默认 3）
max_depth_gap	float	方法 2 深度连续性容差（m）。候选样品与当前样品的 FROM/TO 间隙超过此值则视为不连续，跳过该候选。 默认 0.5m；设为 0 表示要求严格连续。
average_method	AverageMethod	均值计算方式 - "arithmetic": 算术平均 - "length_weighted": 长度加权平均（默认）
result_field	Optional[str]	结果字段名（None 时默认为 {grade_field}_processed）

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

@dataclass
class HighGradeParams:
    """特高品位处理参数

    Attributes:
        grade_field: 品位字段名（如 "CU"、"TFe"）
        group_mode: 分组维度
            - "by_hole":    按工程号分组
            - "by_section": 按勘探线分组（需传入 collar_df）
            - "global":     全矿区统一计算（默认）
        use_average_multiple: 启用平均值倍数法
        average_multiple: 倍数系数。设为 None（默认）时按品位变化系数(CV)自动判定：
                          CV<0.5→6倍, 0.5≤CV<1.0→7倍, CV≥1.0→8倍。
                          也可手动指定 6/7/8。
        use_frequency: 启用累积频率法
        frequency: 累积频率阈值（常用 0.95 / 0.975 / 0.977）
        replace_method: 替换方式
            0 - 剔除法（设为 NaN）
            1 - 给定值法
            2 - 相邻样品平均值法（默认，长度加权）
            3 - 矿体平均品位法
            4 - 单一工程法
            5 - 截止品位法（cap 到阈值）
        assign_value: 方法 1 的给定值
        max_adjacent_search: 方法 2 相邻搜索最大扩展步数（默认 3）
        max_depth_gap: 方法 2 深度连续性容差（m）。候选样品与当前样品的
                       FROM/TO 间隙超过此值则视为不连续，跳过该候选。
                       默认 0.5m；设为 0 表示要求严格连续。
        average_method: 均值计算方式
            - "arithmetic":      算术平均
            - "length_weighted": 长度加权平均（默认）
        result_field: 结果字段名（None 时默认为 {grade_field}_processed）
    """

    grade_field: str
    group_mode: GroupMode = "global"

    # 阈值策略
    use_average_multiple: bool = True
    average_multiple: Optional[int] = None  # None=按CV自动判定6/7/8
    use_frequency: bool = False
    frequency: float = 0.95

    # 替换策略
    replace_method: ReplaceMethod = 2
    assign_value: float = 0.0
    max_adjacent_search: int = 3
    max_depth_gap: float = 0.5

    # 计算方式
    average_method: AverageMethod = "length_weighted"

    # 输出
    result_field: Optional[str] = None

    def __post_init__(self):
        if self.result_field is None:
            self.result_field = f"{self.grade_field}_processed"

HighGradeProcessParam

Bases: ParamModel

特高品位处理参数

Source code in dimine_python_sdk\lib\prospecting\models.py

class HighGradeProcessParam(ParamModel):
    """特高品位处理参数"""

    input_file: str = Field(..., description="输入dmg文件路径")
    grade_field: str = Field(..., description="品位字段名")
    process_mode: Optional[Literal[0, 1]] = Field(
        default=0, description="处理模式（0:国内；1:国外）"
    )
    average_multiple: Optional[Literal[6, 7, 8]] = Field(
        default=6, description="平均值倍数"
    )
    frequency: Optional[float] = Field(
        default=0.95,
        ge=0,
        le=1,
        description="累积频率阈值（0-1 之间,常见值：0.95, 0.975, 0.977）",
    )
    replace_method: Optional[Literal[0, 1, 2, 3, 4, 5]] = Field(
        default=2,
        description="替换方式（0:剔除法，1:给定值，2:相邻样品平均值，3:矿体平均品位法，4:单一工程法，5:截止品位法）",
    )
    assign_value: Optional[float] = Field(
        default=0.0,
        description="给定值方法对应的值,(replace_method=1时，默认值为0， replace_method=2时，默认值为2)",
    )
    adjoin_average: Optional[float] = Field(default=2, description="相邻样品平均值")
    average_method: Optional[Literal[0, 1]] = Field(
        default=1, description="平均值计算方式（0:算数平均，1:长度加权平均）"
    )
    contain_mode: Optional[Literal[0, 1]] = Field(
        default=0, description="包含模式（0:包含特高品位，1:不包含）"
    )
    result_field: Optional[str] = Field(
        default=None, description="结果字段名（未设置时默认为 {grade_field}_processed）"
    )

HighGradeProcessor

特高品位处理器

对样品表 DataFrame 执行特高品位识别与替换，支持： - 平均值倍数法 / 累积频率法双阈值策略 - 六种替换方法（剔除、给定值、相邻平均、矿体平均、单一工程、截止品位） - 单次处理，不做迭代收敛

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

class HighGradeProcessor:
    """特高品位处理器

    对样品表 DataFrame 执行特高品位识别与替换，支持：
    - 平均值倍数法 / 累积频率法双阈值策略
    - 六种替换方法（剔除、给定值、相邻平均、矿体平均、单一工程、截止品位）
    - 单次处理，不做迭代收敛
    """

    def __init__(self, params: HighGradeParams):
        self.params = params
        # 运行时状态
        self.thresholds_: Optional[pd.Series] = None   # 各分组阈值
        self.group_labels_: Optional[pd.Series] = None  # 每个样品所属分组标签

    # ------------------------------------------------------------------
    # 公开入口
    # ------------------------------------------------------------------

    def process(
        self,
        sample_df: pd.DataFrame,
        collar_df: Optional[pd.DataFrame] = None,
    ) -> Dict[str, Any]:
        """执行特高品位处理（单次处理，不迭代）

        Args:
            sample_df: 样品表 DataFrame，需包含 [工程号, 样本编号, 起始, 结束] 及品位列。
            collar_df: 孔口表 DataFrame（group_mode="by_section" 时必需）。

        Returns:
            (processed_df, summary) 元组:
                processed_df — 处理后的完整样品 DataFrame（新增 result_field 列）
                summary     — 汇总统计 dict
        """
        self._reset_state()

        df = sample_df.copy()
        df = self._clean_data(df)
        df["_sample_length"] = self._compute_length(df)
        df["_original_grade"] = df[self.params.grade_field].copy()

        # 分组标签
        self.group_labels_ = self._build_group_labels(df, collar_df)

        # 初始化结果列
        result_field = self.params.result_field
        df[result_field] = df[self.params.grade_field].copy()

        # --- 1. 计算阈值 ---
        thresholds = self._compute_thresholds(df, self.group_labels_)
        self.thresholds_ = thresholds

        # --- 2. 标记超阈值样品 ---
        sample_thresholds = self.group_labels_.map(thresholds)
        flagged_mask = df[result_field] > sample_thresholds

        if flagged_mask.any():
            # --- 3. 替换 ---
            replacements = self._compute_replacements(df, flagged_mask, thresholds)
            df.loc[flagged_mask, result_field] = replacements

        # --- 4. 构建输出 ---
        return self._build_output(df, result_field, flagged_mask)

    # ------------------------------------------------------------------
    # [1] 数据清洗
    # ------------------------------------------------------------------

    def _clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """剔除无效样品"""
        required = ["工程号", "样本编号", "起始", "结束"]
        for col in required:
            if col not in df.columns:
                raise ValueError(f"样品表缺少必要列: {col}")

        grade_col = self.params.grade_field
        if grade_col not in df.columns:
            raise ValueError(f"样品表缺少品位列: {grade_col}")

        # 剔除品位为 NaN 的行
        before = len(df)
        df = df.dropna(subset=[grade_col]).copy()
        if len(df) < before:
            print(f"[清洗] 剔除 {before - len(df)} 条品位为 NaN 的样品")

        # 剔除样长 ≤ 0 的样品
        length = df["结束"] - df["起始"]
        invalid_len = length <= 0
        if invalid_len.any():
            print(f"[清洗] 剔除 {invalid_len.sum()} 条样长 ≤ 0 的样品")
            df = df[~invalid_len].copy()

        # 确保工程号为字符串
        df["工程号"] = df["工程号"].astype(str)

        return df.reset_index(drop=True)

    @staticmethod
    def _compute_length(df: pd.DataFrame) -> pd.Series:
        """计算样长"""
        return df["结束"] - df["起始"]

    # ------------------------------------------------------------------
    # [2] 分组
    # ------------------------------------------------------------------

    def _build_group_labels(
        self, df: pd.DataFrame, collar_df: Optional[pd.DataFrame]
    ) -> pd.Series:
        """为每个样品生成分组标签"""
        mode = self.params.group_mode

        if mode == "by_hole":
            return df["工程号"]

        elif mode == "by_section":
            if collar_df is not None:
                section_map = collar_df.set_index("工程号")["勘探线"]
            else:
                raise ValueError("group_mode='by_section' 需要传入 collar_df")
            return df["工程号"].map(section_map).fillna("未分组")

        elif mode == "global":
            return pd.Series("全矿区", index=df.index)

        else:
            raise ValueError(f"未知分组模式: {mode}")

    # ------------------------------------------------------------------
    # [3] 阈值计算
    # ------------------------------------------------------------------

    def _compute_thresholds(self, df: pd.DataFrame, labels: pd.Series) -> pd.Series:
        """计算各分组的特高品位阈值，取双策略中较严格者"""
        source = df["_original_grade"]

        thresholds = None

        if self.params.use_average_multiple:
            t_avg = self._average_multiple_threshold(df, source, labels)
            thresholds = t_avg

        if self.params.use_frequency:
            t_freq = self._frequency_threshold(source, labels)
            if thresholds is None:
                thresholds = t_freq
            else:
                thresholds = pd.concat([thresholds, t_freq], axis=1).min(axis=1)

        if thresholds is None:
            raise ValueError("至少需要启用一种阈值策略")

        return thresholds

    # ------------------------------------------------------------------
    # CV 自动判定倍数
    # ------------------------------------------------------------------

    @staticmethod
    def _detect_multiple(values: pd.Series) -> int:
        """按品位变化系数(CV)自动选择倍数

        规范规定：
          - 品位变化均匀 → 6 倍（CV < 0.5）
          - 品位变化较均匀 → 7 倍（0.5 ≤ CV < 1.0）
          - 品位变化不均匀 → 8 倍（CV ≥ 1.0）
        """
        if len(values) < 2:
            return 6
        mean = values.mean()
        if mean <= 0:
            return 6
        cv = values.std() / mean
        if cv < 0.5:
            return 6
        elif cv < 1.0:
            return 7
        else:
            return 8

    # ------------------------------------------------------------------
    # 阈值计算方法
    # ------------------------------------------------------------------

    def _average_multiple_threshold(
        self,
        df: pd.DataFrame,
        source_values: pd.Series,
        labels: pd.Series,
    ) -> pd.Series:
        """平均值倍数法：阈值 = 分组品位均值 × 倍数（按CV自动判定或手动指定）"""
        if self.params.average_method == "length_weighted":
            lengths = df["_sample_length"]
            grouped = pd.DataFrame({
                "label": labels,
                "value": source_values,
                "length": lengths,
            })
            means = grouped.groupby("label").apply(
                lambda g: np.average(g["value"], weights=g["length"])
                if g["length"].sum() > 0 else np.nan,
                include_groups=False,
            )
        else:
            means = source_values.groupby(labels).mean()

        # 确定各分组的倍数
        if self.params.average_multiple is not None:
            multiples = pd.Series(self.params.average_multiple, index=means.index)
        else:
            multiples = source_values.groupby(labels).apply(
                self._detect_multiple
            )

        result = (means * multiples).dropna()

        # 记录各分组使用的倍数供输出参考
        self._detected_multiples_ = multiples.to_dict()

        return result

    def _frequency_threshold(
        self,
        source_values: pd.Series,
        labels: pd.Series,
    ) -> pd.Series:
        """累积频率法：阈值 = 第 N 百分位数"""
        q = self.params.frequency
        thresholds = source_values.groupby(labels).quantile(q)
        return thresholds.dropna()

    # ------------------------------------------------------------------
    # [4] 替换值计算
    # ------------------------------------------------------------------

    def _compute_replacements(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
        thresholds: pd.Series,
    ) -> pd.Series:
        """对标记样品计算替换值"""
        method = self.params.replace_method
        flagged_indices = df.index[flagged_mask]

        if method == 0:
            return self._replace_by_discard(flagged_indices)
        elif method == 1:
            return self._replace_by_fixed(flagged_indices)
        elif method == 2:
            return self._replace_by_adjacent(df, flagged_mask)
        elif method == 3:
            return self._replace_by_orebody_average(df, flagged_mask)
        elif method == 4:
            return self._replace_by_hole_average(df, flagged_mask)
        elif method == 5:
            return self._replace_by_cap(df, flagged_mask, thresholds)
        else:
            raise ValueError(f"未知替换方法: {method}")

    def _replace_by_discard(self, indices: pd.Index) -> pd.Series:
        """方法 0：剔除（设为 NaN）"""
        return pd.Series(np.nan, index=indices)

    def _replace_by_fixed(self, indices: pd.Index) -> pd.Series:
        """方法 1：给定固定值"""
        return pd.Series(self.params.assign_value, index=indices)

    def _replace_by_adjacent(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
    ) -> pd.Series:
        """方法 2：相邻未标记样品的长度加权平均

        在同一钻孔内向上/向下搜索未标记样品（跳过已标记），
        最多扩展 max_adjacent_search 步。

        深度连续性校验：候选样品与当前样品的 FROM/TO 间隙超过
        max_depth_gap 时视为不连续，跳过该候选继续向外搜索。
        """
        results = {}
        max_step = self.params.max_adjacent_search
        max_gap = self.params.max_depth_gap

        for hole, hole_df in df.groupby("工程号"):
            hole_indices = hole_df.index

            for idx in hole_indices[flagged_mask.loc[hole_indices]]:
                pos = hole_df.index.get_loc(idx)
                current_from = df.at[idx, "起始"]
                current_to = df.at[idx, "结束"]

                # 向上搜索
                above_values = []
                above_lengths = []
                step = 1
                while step <= max_step and (pos - step) >= 0:
                    candidate_idx = hole_indices[pos - step]
                    candidate = df.loc[candidate_idx]

                    gap = abs(candidate["结束"] - current_from)
                    if gap > max_gap:
                        step += 1
                        continue

                    if not flagged_mask.get(candidate_idx, False):
                        above_values.append(candidate["_original_grade"])
                        above_lengths.append(candidate["_sample_length"])
                        break
                    step += 1

                # 向下搜索
                below_values = []
                below_lengths = []
                step = 1
                while step <= max_step and (pos + step) < len(hole_indices):
                    candidate_idx = hole_indices[pos + step]
                    candidate = df.loc[candidate_idx]

                    gap = abs(candidate["起始"] - current_to)
                    if gap > max_gap:
                        step += 1
                        continue

                    if not flagged_mask.get(candidate_idx, False):
                        below_values.append(candidate["_original_grade"])
                        below_lengths.append(candidate["_sample_length"])
                        break
                    step += 1

                all_vals = above_values + below_values
                all_lens = above_lengths + below_lengths

                if all_vals:
                    if self.params.average_method == "length_weighted" and sum(all_lens) > 0:
                        replacement = np.average(all_vals, weights=all_lens)
                    else:
                        replacement = np.mean(all_vals)
                else:
                    # 回退：该孔内所有未标记样品的均值
                    hole_unflagged_mask = ~flagged_mask.loc[hole_indices]
                    hole_unflagged = hole_indices[hole_unflagged_mask]
                    if len(hole_unflagged) > 0:
                        if self.params.average_method == "length_weighted":
                            vals = df.loc[hole_unflagged, "_original_grade"]
                            lens = df.loc[hole_unflagged, "_sample_length"]
                            replacement = np.average(vals, weights=lens) if lens.sum() > 0 else np.nan
                        else:
                            replacement = df.loc[hole_unflagged, "_original_grade"].mean()
                    else:
                        replacement = np.nan

                results[idx] = replacement

        return pd.Series(results)

    def _replace_by_orebody_average(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
    ) -> pd.Series:
        """方法 3：矿体内未标记样品的长度加权平均"""
        unflagged_mask = ~flagged_mask
        labels = self.group_labels_

        results = {}
        for idx in df.index[flagged_mask]:
            group = labels.at[idx]
            group_mask = (labels == group) & unflagged_mask
            group_df = df.loc[group_mask]
            if len(group_df) == 0:
                results[idx] = np.nan
                continue
            if self.params.average_method == "length_weighted":
                results[idx] = np.average(
                    group_df["_original_grade"],
                    weights=group_df["_sample_length"]
                )
            else:
                results[idx] = group_df["_original_grade"].mean()

        return pd.Series(results)

    def _replace_by_hole_average(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
    ) -> pd.Series:
        """方法 4：同一钻孔内未标记样品的长度加权平均"""
        unflag = ~flagged_mask

        results = {}
        for idx in df.index[flagged_mask]:
            hole = df.at[idx, "工程号"]
            hole_mask = (df["工程号"] == hole) & unflag
            hole_df = df.loc[hole_mask]
            if len(hole_df) == 0:
                results[idx] = np.nan
                continue
            if self.params.average_method == "length_weighted":
                results[idx] = np.average(
                    hole_df["_original_grade"],
                    weights=hole_df["_sample_length"]
                )
            else:
                results[idx] = hole_df["_original_grade"].mean()

        return pd.Series(results)

    def _replace_by_cap(
        self,
        df: pd.DataFrame,
        flagged_mask: pd.Series,
        thresholds: pd.Series,
    ) -> pd.Series:
        """方法 5：截止品位法（cap 到分组阈值）"""
        results = {}
        group_thresholds = self.group_labels_.map(thresholds)
        for idx in df.index[flagged_mask]:
            results[idx] = group_thresholds.at[idx]
        return pd.Series(results)

    # ------------------------------------------------------------------
    # [5] 输出构建
    # ------------------------------------------------------------------

    def _build_output(
        self,
        df: pd.DataFrame,
        result_field: str,
        flagged_mask: pd.Series,
    ) -> Dict[str, Any]:
        """构建最终返回字典"""
        out_df = df.drop(columns=["_sample_length", "_original_grade"], errors="ignore")

        # 标记明细
        flagged_indices = df.index[flagged_mask]
        flagged_samples = []
        for idx in flagged_indices:
            flagged_samples.append({
                "工程号": df.at[idx, "工程号"],
                "样本编号": df.at[idx, "样本编号"],
                "起始": df.at[idx, "起始"],
                "结束": df.at[idx, "结束"],
                "原始品位": df.at[idx, "_original_grade"],
                "处理后品位": df.at[idx, result_field],
                "所属分组": str(self.group_labels_.at[idx]) if self.group_labels_ is not None else None,
            })

        # 分组阈值
        final_thresholds = {}
        if self.thresholds_ is not None:
            for label, thr in self.thresholds_.items():
                final_thresholds[str(label)] = float(thr)

        # CV 自动判定的倍数
        detected_multiples = getattr(self, '_detected_multiples_', None) or {}

        # 汇总
        total = len(df)
        flagged_count = int(flagged_mask.sum())
        original_mean = float(df["_original_grade"].mean())
        processed_mean = float(df[result_field].mean())

        summary = {
            "总样品数": total,
            "特高样品数": flagged_count,
            "特高占比": f"{flagged_count / total * 100:.2f}%" if total > 0 else "0%",
            "原始品位均值": round(original_mean, 4),
            "处理后品位均值": round(processed_mean, 4),
            "均值降幅": f"{(original_mean - processed_mean) / original_mean * 100:.2f}%"
            if original_mean > 0 else "N/A",
            "分组模式": self.params.group_mode,
            "替换方法": _REPLACE_METHOD_NAMES.get(self.params.replace_method, str(self.params.replace_method)),
        }

        return out_df, summary

    def _reset_state(self):
        """重置运行时状态"""
        self.thresholds_ = None
        self.group_labels_ = None

process(sample_df, collar_df=None)

执行特高品位处理（单次处理，不迭代）

Parameters:

Name	Type	Description	Default
sample_df	DataFrame	样品表 DataFrame，需包含 [工程号, 样本编号, 起始, 结束] 及品位列。	required
collar_df	Optional[DataFrame]	孔口表 DataFrame（group_mode="by_section" 时必需）。	None

Returns:

Type	Description
Dict[str, Any]	(processed_df, summary) 元组: processed_df — 处理后的完整样品 DataFrame（新增 result_field 列） summary — 汇总统计 dict

Source code in dimine_python_sdk\lib\prospecting\high_grade.py

def process(
    self,
    sample_df: pd.DataFrame,
    collar_df: Optional[pd.DataFrame] = None,
) -> Dict[str, Any]:
    """执行特高品位处理（单次处理，不迭代）

    Args:
        sample_df: 样品表 DataFrame，需包含 [工程号, 样本编号, 起始, 结束] 及品位列。
        collar_df: 孔口表 DataFrame（group_mode="by_section" 时必需）。

    Returns:
        (processed_df, summary) 元组:
            processed_df — 处理后的完整样品 DataFrame（新增 result_field 列）
            summary     — 汇总统计 dict
    """
    self._reset_state()

    df = sample_df.copy()
    df = self._clean_data(df)
    df["_sample_length"] = self._compute_length(df)
    df["_original_grade"] = df[self.params.grade_field].copy()

    # 分组标签
    self.group_labels_ = self._build_group_labels(df, collar_df)

    # 初始化结果列
    result_field = self.params.result_field
    df[result_field] = df[self.params.grade_field].copy()

    # --- 1. 计算阈值 ---
    thresholds = self._compute_thresholds(df, self.group_labels_)
    self.thresholds_ = thresholds

    # --- 2. 标记超阈值样品 ---
    sample_thresholds = self.group_labels_.map(thresholds)
    flagged_mask = df[result_field] > sample_thresholds

    if flagged_mask.any():
        # --- 3. 替换 ---
        replacements = self._compute_replacements(df, flagged_mask, thresholds)
        df.loc[flagged_mask, result_field] = replacements

    # --- 4. 构建输出 ---
    return self._build_output(df, result_field, flagged_mask)

ParamModel

Bases: BaseModel

参数模型基类，提供 model_dump 的 to_dict 别名以兼容旧代码

Source code in dimine_python_sdk\lib\_base.py

class ParamModel(BaseModel):
    """参数模型基类，提供 model_dump 的 to_dict 别名以兼容旧代码"""

    model_config = ConfigDict(populate_by_name=True)

    def to_dict(self) -> Dict[str, Any]:
        return self.model_dump(by_alias=True)

PlaneConstraint

Bases: ParamModel

平面约束 (type=4)：由平面方程 ax+by+cz+d=0 定义的上部/下部约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class PlaneConstraint(ParamModel):
    """平面约束 (type=4)：由平面方程 ax+by+cz+d=0 定义的上部/下部约束"""

    type: Literal[4] = Field(default=4, description="平面约束类型")
    a: float = Field(..., description="平面方程系数 a（法向量 X 分量），通过ax+by+cz+d=0构成一个平面")
    b: float = Field(..., description="平面方程系数 b（法向量 Y 分量），通过ax+by+cz+d=0构成一个平面")
    c: float = Field(..., description="平面方程系数 c（法向量 Z 分量），通过ax+by+cz+d=0构成一个平面")
    d: float = Field(..., description="平面方程常数 d，通过ax+by+cz+d=0构成一个平面")
    range: int = Field(default=0, description="约束范围，0=上部，1=下部")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

PolylineConstraint

Bases: ParamModel

闭合多段线约束 (type=3)：基于闭合多段线 + 坐标轴范围的约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class PolylineConstraint(ParamModel):
    """闭合多段线约束 (type=3)：基于闭合多段线 + 坐标轴范围的约束"""

    type: Literal[3] = Field(default=3, description="闭合多段线约束类型")
    file: str = Field(..., description="闭合多段线dmf文件路径")
    range: int = Field(default=0, description="约束范围，0=内部，1=外部")
    attribute: Literal["X","Y","Z"] = Field(default="Z", description="坐标轴：X、Y 或 Z")
    from_: Optional[float] = Field(
        default=-100000, alias="from", description="坐标轴起始值"
    )
    to: Optional[float] = Field(default=100000, description="坐标轴结束值")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

ReservesCalculateError

Bases: BlockDataError

储量计算失败

Source code in dimine_python_sdk\lib\prospecting\block_model.py

class ReservesCalculateError(BlockDataError):
    """储量计算失败"""
    def __init__(self, message: str, response: dict | None = None):
        super().__init__(message)
        self.response = response

ReservesCalculateParam

Bases: ParamModel

估值算量参数

Source code in dimine_python_sdk\lib\prospecting\models.py

class ReservesCalculateParam(ParamModel):
    """估值算量参数"""

    weight_field: str = Field(..., description="体重字段")
    default_weight: Optional[float] = Field(default=2.76, description="默认体重值")
    main_field: str = Field(..., description="主统计字段")
    other_fields: list[str] = Field(default_factory=list, description="其他统计字段列表")

SampleCompositingParams dataclass

样长组合参数

Attributes:

Name	Type	Description
composite_length	float	组合样长度（米），必须 > 0。 建议取值为块体尺寸的 1/4 到 1/3。
grade_fields	List[str]	需要做长度加权平均的品位字段名列表。 字段必须存在于样品表中，且为数值类型。
combine_percent	float	最小覆盖率（0-1），组合区间内实际样品覆盖长度 低于此比例时标记为低质量组合样。 默认 0.75（与 DIMINE C++ 接口默认值一致）。
start_mode	str	组合起始模式。 - 'first_sample': 从每个钻孔第一个样品起始深度开始（默认） - 'collar': 从孔口（深度0）开始 - 数字: 从指定的绝对深度开始（如 5.0）
discard_low_coverage	bool	尾部截断模式。 为 True 时，从孔口方向找到最后一个满足覆盖率要求的区间， 丢弃其之后的所有区间（即"尾部截断"）。 注意：截断点之前的中间低覆盖率区间仍会被保留。 为 False 时保留所有区间，通过 "低覆盖率" 列标记。
include_3d_coords	bool	是否计算组合样段中点的三维坐标。 需要提供 survey 和 collar 表。
min_samples_per_composite	int	每个组合样最少包含的原始样品数（警告阈值）。
auto_detect_grade_fields	bool	是否自动检测样品表中的数值型品位字段。 为 True 时，grade_fields 中未列出的数值列也会被组合。 自动排除标准列（工程号、样本编号、坐标等）。

Source code in dimine_python_sdk\lib\prospecting\compositing.py

@dataclass
class SampleCompositingParams:
    """样长组合参数

    Attributes:
        composite_length: 组合样长度（米），必须 > 0。
                          建议取值为块体尺寸的 1/4 到 1/3。
        grade_fields: 需要做长度加权平均的品位字段名列表。
                      字段必须存在于样品表中，且为数值类型。
        combine_percent: 最小覆盖率（0-1），组合区间内实际样品覆盖长度
                         低于此比例时标记为低质量组合样。
                         默认 0.75（与 DIMINE C++ 接口默认值一致）。
        start_mode: 组合起始模式。
                     - 'first_sample': 从每个钻孔第一个样品起始深度开始（默认）
                     - 'collar': 从孔口（深度0）开始
                     - 数字: 从指定的绝对深度开始（如 5.0）
        discard_low_coverage: 尾部截断模式。
                              为 True 时，从孔口方向找到最后一个满足覆盖率要求的区间，
                              丢弃其之后的所有区间（即"尾部截断"）。
                              注意：截断点之前的中间低覆盖率区间仍会被保留。
                              为 False 时保留所有区间，通过 "低覆盖率" 列标记。
        include_3d_coords: 是否计算组合样段中点的三维坐标。
                           需要提供 survey 和 collar 表。
        min_samples_per_composite: 每个组合样最少包含的原始样品数（警告阈值）。
        auto_detect_grade_fields: 是否自动检测样品表中的数值型品位字段。
                                  为 True 时，grade_fields 中未列出的数值列也会被组合。
                                  自动排除标准列（工程号、样本编号、坐标等）。
    """

    composite_length: float = 1.0
    grade_fields: List[str] = field(default_factory=list)
    combine_percent: float = 0.75
    start_mode: str = "first_sample"  # 'first_sample' | 'collar' | numeric string
    discard_low_coverage: bool = False
    include_3d_coords: bool = True
    min_samples_per_composite: int = 1
    auto_detect_grade_fields: bool = True

    _VALID_START_MODES = frozenset({"first_sample", "collar"})

    def __post_init__(self):
        if self.composite_length <= 0:
            raise ValueError(
                f"composite_length 必须 > 0，当前值: {self.composite_length}"
            )
        if not (0 <= self.combine_percent <= 1):
            raise ValueError(
                f"combine_percent 必须在 0-1 之间，当前值: {self.combine_percent}"
            )
        # 校验 start_mode：允许 'first_sample'、'collar'，或可转为 float 的字符串
        if (
            isinstance(self.start_mode, str)
            and self.start_mode not in self._VALID_START_MODES
        ):
            try:
                float(self.start_mode)
            except ValueError:
                raise ValueError(
                    f"start_mode 必须是 'first_sample'、'collar' 或数字，"
                    f"当前值: '{self.start_mode}'"
                )

SampleCompositingProcessor

样长组合处理器

对钻孔样品表执行样长组合操作。核心流程：

1. 数据预处理（NaN 清理、类型转换、数据校验）
2. 按钻孔（工程号）分组
3. 对每个钻孔独立执行长度加权组合
4. 汇总所有钻孔的组合结果
5. 生成统计报告

Usage

params = SampleCompositingParams( composite_length=2.0, grade_fields=['TFe', 'TiO2'], ) processor = SampleCompositingProcessor(params) result = processor.process(sample_df, survey_df, collar_df)

Source code in dimine_python_sdk\lib\prospecting\compositing.py

class SampleCompositingProcessor:
    """样长组合处理器

    对钻孔样品表执行样长组合操作。核心流程：

        1. 数据预处理（NaN 清理、类型转换、数据校验）
        2. 按钻孔（工程号）分组
        3. 对每个钻孔独立执行长度加权组合
        4. 汇总所有钻孔的组合结果
        5. 生成统计报告

    Usage:
        params = SampleCompositingParams(
            composite_length=2.0,
            grade_fields=['TFe', 'TiO2'],
        )
        processor = SampleCompositingProcessor(params)
        result = processor.process(sample_df, survey_df, collar_df)
    """

    def __init__(self, params: Optional[SampleCompositingParams] = None, **kwargs):
        """
        Args:
            params: SampleCompositingParams 参数对象。
                    为 None 时从 **kwargs 构建。
            **kwargs: 直接传入 SampleCompositingParams 的参数。
        """
        if params is None:
            params = SampleCompositingParams(**kwargs)
        self.params = params

    # ------------------------------------------------------------------
    # 主入口
    # ------------------------------------------------------------------

    def process(
        self,
        sample_df: pd.DataFrame,
        survey_df: Optional[pd.DataFrame] = None,
        collar_df: Optional[pd.DataFrame] = None,
    ) -> Dict[str, Any]:
        """
        执行样长组合。

        Args:
            sample_df: 样品表 DataFrame（中文标准列名：工程号, 起始, 结束, ...）
            survey_df: 测斜表 DataFrame（可选，用于计算3D坐标）
            collar_df: 孔口表 DataFrame（可选，用于计算3D坐标）

        Returns:
            (composited_df, summary) 元组:
                composited_df — 组合后的样品 DataFrame
                summary      — 汇总统计 dict
        """
        p = self.params

        # --- 拷贝输入 ---
        samples = sample_df.copy()

        # --- 数据类型预处理 ---
        # 确保起始/结束为数值类型，转换失败的 NaN 提前清理
        for col in ["起始", "结束"]:
            if col in samples.columns:
                samples[col] = pd.to_numeric(samples[col], errors="coerce")

        # 提前统计并报告因转换失败而产生的 NaN
        nan_before = len(samples)
        samples = samples.dropna(subset=["起始", "结束"])
        nan_dropped = nan_before - len(samples)
        if nan_dropped > 0:
            _logger.warning(
                f"样品表中 {nan_dropped} 行因 '起始'/'结束' 为非数值而被丢弃。"
                f"原始行数: {nan_before}, 保留行数: {len(samples)}"
            )

        # 过滤 工程号 为空的行
        samples = samples.dropna(subset=["工程号"])
        samples["工程号"] = samples["工程号"].astype(str)

        # 空数据快速返回
        if len(samples) == 0:
            return pd.DataFrame(), {"原始样品总数": 0, "组合样品总数": 0}

        # --- 自动检测品位字段 ---
        grade_fields = list(p.grade_fields)
        if p.auto_detect_grade_fields:
            for col in samples.columns:
                if col not in _STANDARD_SAMPLE_COLS and col not in grade_fields:
                    if pd.api.types.is_numeric_dtype(samples[col]):
                        grade_fields.append(col)

        if not grade_fields:
            raise ValueError(
                "未指定品位字段，且样品表中未检测到数值型品位列。"
                "请通过 grade_fields 参数明确指定。"
            )

        # 验证品位字段存在
        missing_fields = [gf for gf in grade_fields if gf not in samples.columns]
        if missing_fields:
            raise ValueError(
                f"品位字段 {missing_fields} 在样品表中不存在。"
                f"可用列: {list(samples.columns)}"
            )

        # --- 构建孔口信息查找表 ---
        collar_lookup: Dict[str, Dict[str, float]] = {}
        if collar_df is not None:
            for _, row in collar_df.iterrows():
                hole_id = str(row.get("工程号", ""))
                collar_lookup[hole_id] = {
                    "横坐标": float(row.get("横坐标", 0)),
                    "纵坐标": float(row.get("纵坐标", 0)),
                    "高程": float(row.get("高程", 0)),
                }

        # --- 构建测斜数据查找表 ---
        survey_lookup: Dict[str, pd.DataFrame] = {}
        if survey_df is not None:
            for hole_id, group in survey_df.groupby("工程号"):
                survey_lookup[str(hole_id)] = group.copy()

        # --- 确定起始深度 ---
        start_depth_override = None
        if isinstance(p.start_mode, (int, float)):
            start_depth_override = float(p.start_mode)
        elif isinstance(p.start_mode, str):
            try:
                start_depth_override = float(p.start_mode)
            except ValueError:
                pass  # 不是数字字符串，保持为 None

        # --- 按钻孔分组处理 ---
        hole_ids = sorted(samples["工程号"].unique())
        all_composited: List[pd.DataFrame] = []
        all_stats: List[Dict[str, Any]] = []
        hole_data: Dict[str, pd.DataFrame] = {}
        all_validation_warnings: List[str] = []
        total_original = 0
        total_composited = 0

        for hole_id in hole_ids:
            hole_samples = (
                samples[samples["工程号"] == hole_id].sort_values("起始").copy()
            )
            if len(hole_samples) == 0:
                continue

            collar_info = collar_lookup.get(str(hole_id))
            survey_hole = survey_lookup.get(str(hole_id))

            comp_df, stats = _composite_single_hole(
                hole_id=str(hole_id),
                samples=hole_samples,
                composite_length=p.composite_length,
                grade_fields=grade_fields,
                combine_percent=p.combine_percent,
                start_mode=p.start_mode,
                start_depth_override=start_depth_override,
                discard_low_coverage=p.discard_low_coverage,
                collar_info=collar_info,
                survey_hole=survey_hole,
                include_3d=p.include_3d_coords,
            )

            total_original += stats.get("n_original", 0)
            total_composited += stats.get("n_composited", 0)
            all_stats.append(stats)

            if not comp_df.empty:
                all_composited.append(comp_df)
                hole_data[str(hole_id)] = comp_df

        # --- 汇总 ---
        if all_composited:
            composited_df = pd.concat(all_composited, ignore_index=True)
        else:
            cols = ["工程号", "样本编号", "起始", "结束", "样长"] + grade_fields
            if p.include_3d_coords:
                cols += ["X", "Y", "Z"]
            composited_df = pd.DataFrame(columns=cols)

        # --- 统计 ---
        per_hole_df = pd.DataFrame(all_stats) if all_stats else pd.DataFrame()

        summary: Dict[str, Any] = {
            "总钻孔数": len(hole_ids),
            "有效钻孔数": len([s for s in all_stats if s.get("n_composited", 0) > 0]),
            "原始样品总数": total_original,
            "组合样品总数": total_composited,
            "压缩比": (
                f"{total_original} → {total_composited}"
                if total_composited > 0
                else "N/A"
            ),
            "组合长度": p.composite_length,
            "最小覆盖率": p.combine_percent,
            "起始模式": p.start_mode,
            "品位字段": grade_fields,
        }

        if not per_hole_df.empty:
            low_cov_total = per_hole_df["low_coverage_count"].sum()
            summary["低覆盖率样本数"] = int(low_cov_total)
            if "mean_coverage" in per_hole_df.columns:
                summary["平均覆盖率"] = float(per_hole_df["mean_coverage"].mean())
            val_warn_total = per_hole_df["validation_warnings"].sum()
            if val_warn_total > 0:
                summary["数据质量警告数"] = int(val_warn_total)

        return composited_df, summary

__init__(params=None, **kwargs)

Parameters:

Name	Type	Description	Default
params	Optional[SampleCompositingParams]	SampleCompositingParams 参数对象。 为 None 时从 **kwargs 构建。	None
**kwargs		直接传入 SampleCompositingParams 的参数。	{}

Source code in dimine_python_sdk\lib\prospecting\compositing.py

def __init__(self, params: Optional[SampleCompositingParams] = None, **kwargs):
    """
    Args:
        params: SampleCompositingParams 参数对象。
                为 None 时从 **kwargs 构建。
        **kwargs: 直接传入 SampleCompositingParams 的参数。
    """
    if params is None:
        params = SampleCompositingParams(**kwargs)
    self.params = params

process(sample_df, survey_df=None, collar_df=None)

执行样长组合。

Parameters:

Name	Type	Description	Default
sample_df	DataFrame	样品表 DataFrame（中文标准列名：工程号, 起始, 结束, ...）	required
survey_df	Optional[DataFrame]	测斜表 DataFrame（可选，用于计算3D坐标）	None
collar_df	Optional[DataFrame]	孔口表 DataFrame（可选，用于计算3D坐标）	None

Returns:

Type	Description
Dict[str, Any]	(composited_df, summary) 元组: composited_df — 组合后的样品 DataFrame summary — 汇总统计 dict

Source code in dimine_python_sdk\lib\prospecting\compositing.py

def process(
    self,
    sample_df: pd.DataFrame,
    survey_df: Optional[pd.DataFrame] = None,
    collar_df: Optional[pd.DataFrame] = None,
) -> Dict[str, Any]:
    """
    执行样长组合。

    Args:
        sample_df: 样品表 DataFrame（中文标准列名：工程号, 起始, 结束, ...）
        survey_df: 测斜表 DataFrame（可选，用于计算3D坐标）
        collar_df: 孔口表 DataFrame（可选，用于计算3D坐标）

    Returns:
        (composited_df, summary) 元组:
            composited_df — 组合后的样品 DataFrame
            summary      — 汇总统计 dict
    """
    p = self.params

    # --- 拷贝输入 ---
    samples = sample_df.copy()

    # --- 数据类型预处理 ---
    # 确保起始/结束为数值类型，转换失败的 NaN 提前清理
    for col in ["起始", "结束"]:
        if col in samples.columns:
            samples[col] = pd.to_numeric(samples[col], errors="coerce")

    # 提前统计并报告因转换失败而产生的 NaN
    nan_before = len(samples)
    samples = samples.dropna(subset=["起始", "结束"])
    nan_dropped = nan_before - len(samples)
    if nan_dropped > 0:
        _logger.warning(
            f"样品表中 {nan_dropped} 行因 '起始'/'结束' 为非数值而被丢弃。"
            f"原始行数: {nan_before}, 保留行数: {len(samples)}"
        )

    # 过滤 工程号 为空的行
    samples = samples.dropna(subset=["工程号"])
    samples["工程号"] = samples["工程号"].astype(str)

    # 空数据快速返回
    if len(samples) == 0:
        return pd.DataFrame(), {"原始样品总数": 0, "组合样品总数": 0}

    # --- 自动检测品位字段 ---
    grade_fields = list(p.grade_fields)
    if p.auto_detect_grade_fields:
        for col in samples.columns:
            if col not in _STANDARD_SAMPLE_COLS and col not in grade_fields:
                if pd.api.types.is_numeric_dtype(samples[col]):
                    grade_fields.append(col)

    if not grade_fields:
        raise ValueError(
            "未指定品位字段，且样品表中未检测到数值型品位列。"
            "请通过 grade_fields 参数明确指定。"
        )

    # 验证品位字段存在
    missing_fields = [gf for gf in grade_fields if gf not in samples.columns]
    if missing_fields:
        raise ValueError(
            f"品位字段 {missing_fields} 在样品表中不存在。"
            f"可用列: {list(samples.columns)}"
        )

    # --- 构建孔口信息查找表 ---
    collar_lookup: Dict[str, Dict[str, float]] = {}
    if collar_df is not None:
        for _, row in collar_df.iterrows():
            hole_id = str(row.get("工程号", ""))
            collar_lookup[hole_id] = {
                "横坐标": float(row.get("横坐标", 0)),
                "纵坐标": float(row.get("纵坐标", 0)),
                "高程": float(row.get("高程", 0)),
            }

    # --- 构建测斜数据查找表 ---
    survey_lookup: Dict[str, pd.DataFrame] = {}
    if survey_df is not None:
        for hole_id, group in survey_df.groupby("工程号"):
            survey_lookup[str(hole_id)] = group.copy()

    # --- 确定起始深度 ---
    start_depth_override = None
    if isinstance(p.start_mode, (int, float)):
        start_depth_override = float(p.start_mode)
    elif isinstance(p.start_mode, str):
        try:
            start_depth_override = float(p.start_mode)
        except ValueError:
            pass  # 不是数字字符串，保持为 None

    # --- 按钻孔分组处理 ---
    hole_ids = sorted(samples["工程号"].unique())
    all_composited: List[pd.DataFrame] = []
    all_stats: List[Dict[str, Any]] = []
    hole_data: Dict[str, pd.DataFrame] = {}
    all_validation_warnings: List[str] = []
    total_original = 0
    total_composited = 0

    for hole_id in hole_ids:
        hole_samples = (
            samples[samples["工程号"] == hole_id].sort_values("起始").copy()
        )
        if len(hole_samples) == 0:
            continue

        collar_info = collar_lookup.get(str(hole_id))
        survey_hole = survey_lookup.get(str(hole_id))

        comp_df, stats = _composite_single_hole(
            hole_id=str(hole_id),
            samples=hole_samples,
            composite_length=p.composite_length,
            grade_fields=grade_fields,
            combine_percent=p.combine_percent,
            start_mode=p.start_mode,
            start_depth_override=start_depth_override,
            discard_low_coverage=p.discard_low_coverage,
            collar_info=collar_info,
            survey_hole=survey_hole,
            include_3d=p.include_3d_coords,
        )

        total_original += stats.get("n_original", 0)
        total_composited += stats.get("n_composited", 0)
        all_stats.append(stats)

        if not comp_df.empty:
            all_composited.append(comp_df)
            hole_data[str(hole_id)] = comp_df

    # --- 汇总 ---
    if all_composited:
        composited_df = pd.concat(all_composited, ignore_index=True)
    else:
        cols = ["工程号", "样本编号", "起始", "结束", "样长"] + grade_fields
        if p.include_3d_coords:
            cols += ["X", "Y", "Z"]
        composited_df = pd.DataFrame(columns=cols)

    # --- 统计 ---
    per_hole_df = pd.DataFrame(all_stats) if all_stats else pd.DataFrame()

    summary: Dict[str, Any] = {
        "总钻孔数": len(hole_ids),
        "有效钻孔数": len([s for s in all_stats if s.get("n_composited", 0) > 0]),
        "原始样品总数": total_original,
        "组合样品总数": total_composited,
        "压缩比": (
            f"{total_original} → {total_composited}"
            if total_composited > 0
            else "N/A"
        ),
        "组合长度": p.composite_length,
        "最小覆盖率": p.combine_percent,
        "起始模式": p.start_mode,
        "品位字段": grade_fields,
    }

    if not per_hole_df.empty:
        low_cov_total = per_hole_df["low_coverage_count"].sum()
        summary["低覆盖率样本数"] = int(low_cov_total)
        if "mean_coverage" in per_hole_df.columns:
            summary["平均覆盖率"] = float(per_hole_df["mean_coverage"].mean())
        val_warn_total = per_hole_df["validation_warnings"].sum()
        if val_warn_total > 0:
            summary["数据质量警告数"] = int(val_warn_total)

    return composited_df, summary

SampleLengthCombineParam

Bases: ParamModel

样长组合参数

Source code in dimine_python_sdk\lib\prospecting\models.py

class SampleLengthCombineParam(ParamModel):
    """样长组合参数"""

    input_file: str = Field(..., description="输入dmg文件")
    combine_length: Optional[float] = Field(
        default=1, gt=0, description="组合长度（m），必须大于 0"
    )
    combine_percent: Optional[float] = Field(
        default=0.75,
        ge=0,
        le=1,
        description="组合百分比，0-1 之间（传给 C++ 前会 ×100 取整）",
    )
    output_file: Optional[str] = Field(
        default=None,
        description="输出dmg文件（未设置时默认为 {input_file stem}_combined.dmg）",
    )

SearchEllipsoidParams

Bases: ParamModel

搜索椭球体参数，可直接映射到 BlockModelDistancePowerParams。

三个角度依次为 Z-Y-X 欧拉角序列： R = Rz(π/2 − angle_main) × Ry(angle_second) × Rx(angle_short)

其中 R 的列向量为 [PC1, PC2, PC3]。

Source code in dimine_python_sdk\lib\prospecting\ellipsoid.py

class SearchEllipsoidParams(ParamModel):
    """搜索椭球体参数，可直接映射到 ``BlockModelDistancePowerParams``。

    三个角度依次为 Z-Y-X 欧拉角序列：
        R = Rz(π/2 − angle_main) × Ry(angle_second) × Rx(angle_short)

    其中 R 的列向量为 [PC1, PC2, PC3]。
    """

    # ---- 形状参数 ----
    main_radius: float
    """主轴半径（m），椭球体最长半轴"""
    second_radius: float
    """次轴半径（m），椭球体中等半轴"""
    short_radius: float
    """短轴半径（m），椭球体最短半轴"""
    second_main_rate: float
    """次轴/主轴比例"""
    short_main_rate: float
    """短轴/主轴比例"""

    # ---- 方位参数（SDK 映射） ----
    angle_main: float
    """方位角（°），主轴水平投影方向，0°=正北，90°=正东"""
    angle_second: float
    """倾伏角（°），主轴与水平面夹角，正值=向下，0°=水平，90°=垂直"""
    angle_short: float
    """倾角（°），椭球体绕主轴旋转角，控制短轴（厚度方向）的侧倾"""

    # ---- 辅助信息 ----
    center: Tuple[float, float, float]
    """椭球体中心坐标 (x, y, z)"""
    pc1_direction: Tuple[float, float, float]
    """主轴方向单位向量（走向方向）"""
    pc2_direction: Tuple[float, float, float]
    """次轴方向单位向量（倾向方向）"""
    pc3_direction: Tuple[float, float, float]
    """短轴方向单位向量（厚度方向）"""

    def to_block_model_params(self) -> dict:
        """提取 ``BlockModelDistancePowerParams`` 所需的 6 个字段。

        Returns:
            dict，包含 main_radius、second_main_rate、short_main_rate、
            angle_main、angle_second、angle_short
        """
        return {
            "main_radius": self.main_radius,
            "second_main_rate": self.second_main_rate,
            "short_main_rate": self.short_main_rate,
            "angle_main": self.angle_main,
            "angle_second": self.angle_second,
            "angle_short": self.angle_short,
        }

angle_main instance-attribute

方位角（°），主轴水平投影方向，0°=正北，90°=正东

angle_second instance-attribute

倾伏角（°），主轴与水平面夹角，正值=向下，0°=水平，90°=垂直

angle_short instance-attribute

倾角（°），椭球体绕主轴旋转角，控制短轴（厚度方向）的侧倾

center instance-attribute

椭球体中心坐标 (x, y, z)

main_radius instance-attribute

主轴半径（m），椭球体最长半轴

pc1_direction instance-attribute

主轴方向单位向量（走向方向）

pc2_direction instance-attribute

次轴方向单位向量（倾向方向）

pc3_direction instance-attribute

短轴方向单位向量（厚度方向）

second_main_rate instance-attribute

次轴/主轴比例

second_radius instance-attribute

次轴半径（m），椭球体中等半轴

short_main_rate instance-attribute

短轴/主轴比例

short_radius instance-attribute

短轴半径（m），椭球体最短半轴

to_block_model_params()

提取 BlockModelDistancePowerParams 所需的 6 个字段。

Returns:

Type	Description
dict	dict，包含 main_radius、second_main_rate、short_main_rate、
dict	angle_main、angle_second、angle_short

Source code in dimine_python_sdk\lib\prospecting\ellipsoid.py

def to_block_model_params(self) -> dict:
    """提取 ``BlockModelDistancePowerParams`` 所需的 6 个字段。

    Returns:
        dict，包含 main_radius、second_main_rate、short_main_rate、
        angle_main、angle_second、angle_short
    """
    return {
        "main_radius": self.main_radius,
        "second_main_rate": self.second_main_rate,
        "short_main_rate": self.short_main_rate,
        "angle_main": self.angle_main,
        "angle_second": self.angle_second,
        "angle_short": self.angle_short,
    }

StepCombineParam

Bases: ParamModel

台阶组合参数

Source code in dimine_python_sdk\lib\prospecting\models.py

class StepCombineParam(ParamModel):
    """台阶组合参数"""

    input_file: str = Field(..., description="输入dmg文件")
    step_height: Optional[float] = Field(default=2, description="台阶高度")
    start_height: float = Field(..., description="开始高程")
    end_height: float = Field(..., description="结束高程")
    calculate_model: Optional[int] = Field(
        default=1, description="计算方式:0-算术平均，1-长度加权平均"
    )
    low_dip: Optional[float] = Field(
        default=0, description="忽略的最小倾角，即忽略倾角小于等于正负该值的钻孔"
    )
    output_file: Optional[str] = Field(
        default=None,
        description="输出dmg文件（未设置时默认为 {input_file stem}_step.dmg）",
    )

ThresholdConstraint

Bases: ParamModel

阈值约束 (type=2)：基于属性值比较的约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class ThresholdConstraint(ParamModel):
    """阈值约束 (type=2)：基于属性值比较的约束"""

    type: Literal[2] = Field(default=2, description="阈值约束类型")
    attribute: str = Field(..., description="属性字段名称（如 CU）")
    compare: str = Field(default="等于", description="比较运算符：等于、大于、大于等于、小于、小于等于、不等于")
    value: str = Field(..., description="属性值（阈值）")
    is_numeric: bool = Field(default=True, description="是否为数值类型")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

XPlaneConstraint

Bases: ParamModel

X 平面约束 (type=5)：基于 X 坐标值的上部/下部约束，一种特殊的平面约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class XPlaneConstraint(ParamModel):
    """X 平面约束 (type=5)：基于 X 坐标值的上部/下部约束，一种特殊的平面约束"""

    type: Literal[5] = Field(default=5, description="X平面约束类型")
    x: float = Field(..., description="X 坐标值")
    range: int = Field(default=0, description="约束范围，0=上部，1=下部")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

YPlaneConstraint

Bases: ParamModel

Y 平面约束 (type=6)：基于 Y 坐标值的上部/下部约束，一种特殊的平面约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class YPlaneConstraint(ParamModel):
    """Y 平面约束 (type=6)：基于 Y 坐标值的上部/下部约束，一种特殊的平面约束"""

    type: Literal[6] = Field(default=6, description="约束类型")
    y: float = Field(..., description="Y 坐标值")
    range: int = Field(default=0, description="约束范围，0=上部，1=下部")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

ZPlaneConstraint

Bases: ParamModel

Z 平面约束 (type=7)：基于 Z 坐标值（高程）的上部/下部约束，一种特殊的平面约束

Source code in dimine_python_sdk\lib\prospecting\models.py

class ZPlaneConstraint(ParamModel):
    """Z 平面约束 (type=7)：基于 Z 坐标值（高程）的上部/下部约束，一种特殊的平面约束"""

    type: Literal[7] = Field(default=7, description="约束类型")
    z: float = Field(..., description="Z 坐标值（高程）")
    range: int = Field(default=0, description="约束范围，0=上部，1=下部")
    inside_level: int = Field(default=8, description="内部尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    boundary_level: int = Field(default=11, description="边界尺寸层级，0=2560*2560*1280，1=1280*1280*640,2=640*640*320,...，10=2.5*2.5*1.25，11=1.25*1.25*0.625,...,13=0.3125*0.3125*0.15625 每增加一个层级，每个边在前一个等级基础上除以2，以此类推")
    bool_operate: str = Field(
        default="and", description='布尔运算符，"and" 或 "or"'
    )

composite_samples(db_or_samples, composite_length=2.0, grade_fields=None, combine_percent=0.75, discard_low_coverage=False, start_mode='first_sample', include_3d_coords=True, auto_detect_grade_fields=True, survey_df=None, collar_df=None)

样长组合便捷函数。

支持两种输入模式： 1. 从 DrillDBManager 直接传入： result = composite_samples(db, composite_length=2.0, grade_fields=['TFe']) 2. 从独立的 DataFrame 传入： result = composite_samples( sample_df, composite_length=2.0, grade_fields=['TFe', 'TiO2'], survey_df=survey_df, collar_df=collar_df )

discard_low_coverage 策略： - False（默认）：保留所有组合区间，通过 "低覆盖率" 列标记 - True：尾部截断——找到最后一个合格区间，丢弃其之后的所有区间。 如需丢弃所有不合格区间（包括中间的），请在结果上执行： df = df[~df['低覆盖率']]

Parameters:

Name	Type	Description	Default
db_or_samples		DrillDBManager 实例 或 样品表 pd.DataFrame	required
composite_length	float	组合样长度（米），默认 2.0	2.0
grade_fields	Optional[List[str]]	品位字段列表。为 None 时自动检测数值列	None
combine_percent	float	最小覆盖率（0-1），默认 0.75	0.75
discard_low_coverage	bool	尾部截断模式，默认 False	False
start_mode	str	起始模式，默认 'first_sample'	'first_sample'
include_3d_coords	bool	是否计算3D坐标	True
auto_detect_grade_fields	bool	是否自动检测品位字段	True
survey_df	Optional[DataFrame]	测斜表（仅当 db_or_samples 为 DataFrame 时需要）	None
collar_df	Optional[DataFrame]	孔口表（仅当 db_or_samples 为 DataFrame 时需要）	None

Returns:

Type	Description
Dict[str, Any]	(composited_df, summary) 元组

Source code in dimine_python_sdk\lib\prospecting\compositing.py

def composite_samples(
    db_or_samples,
    composite_length: float = 2.0,
    grade_fields: Optional[List[str]] = None,
    combine_percent: float = 0.75,
    discard_low_coverage: bool = False,
    start_mode: str = "first_sample",
    include_3d_coords: bool = True,
    auto_detect_grade_fields: bool = True,
    survey_df: Optional[pd.DataFrame] = None,
    collar_df: Optional[pd.DataFrame] = None,
) -> Dict[str, Any]:
    """
    样长组合便捷函数。

    支持两种输入模式：
        1. 从 DrillDBManager 直接传入：
           result = composite_samples(db, composite_length=2.0, grade_fields=['TFe'])
        2. 从独立的 DataFrame 传入：
           result = composite_samples(
               sample_df, composite_length=2.0, grade_fields=['TFe', 'TiO2'],
               survey_df=survey_df, collar_df=collar_df
           )

    discard_low_coverage 策略：
        - False（默认）：保留所有组合区间，通过 "低覆盖率" 列标记
        - True：尾部截断——找到最后一个合格区间，丢弃其之后的所有区间。
          如需丢弃所有不合格区间（包括中间的），请在结果上执行：
          df = df[~df['低覆盖率']]

    Args:
        db_or_samples: DrillDBManager 实例 或 样品表 pd.DataFrame
        composite_length: 组合样长度（米），默认 2.0
        grade_fields: 品位字段列表。为 None 时自动检测数值列
        combine_percent: 最小覆盖率（0-1），默认 0.75
        discard_low_coverage: 尾部截断模式，默认 False
        start_mode: 起始模式，默认 'first_sample'
        include_3d_coords: 是否计算3D坐标
        auto_detect_grade_fields: 是否自动检测品位字段
        survey_df: 测斜表（仅当 db_or_samples 为 DataFrame 时需要）
        collar_df: 孔口表（仅当 db_or_samples 为 DataFrame 时需要）

    Returns:
        (composited_df, summary) 元组
    """
    from dimine_python_sdk.lib.prospecting.drill_db import DrillDBManager

    if isinstance(db_or_samples, DrillDBManager):
        db = db_or_samples
        sample_df = db.sample.copy()
        if survey_df is None:
            survey_df = db.survey.copy()
        if collar_df is None:
            collar_df = db.collar.copy()
    elif isinstance(db_or_samples, pd.DataFrame):
        sample_df = db_or_samples.copy()
    else:
        raise TypeError(
            f"db_or_samples 必须是 DrillDBManager 或 pd.DataFrame，"
            f"当前类型: {type(db_or_samples)}"
        )

    params = SampleCompositingParams(
        composite_length=composite_length,
        grade_fields=grade_fields or [],
        combine_percent=combine_percent,
        start_mode=start_mode,
        discard_low_coverage=discard_low_coverage,
        include_3d_coords=include_3d_coords,
        auto_detect_grade_fields=auto_detect_grade_fields,
    )

    processor = SampleCompositingProcessor(params)
    return processor.process(sample_df, survey_df, collar_df)

compute_search_ellipsoid(points, voxel_size=5.0)

从点云计算搜索椭球体参数。

流程：体素降采样 → MVEE → 角度分解。

Parameters:

Name	Type	Description	Default
points	ndarray	点云坐标，shape=(N, 3)，(x=东, y=北, z=高程)	required
voxel_size	Optional[float]	体素降采样尺寸（米），设为 None 或 0 则跳过降采样。 推荐 5.0 m，可有效消除三角网密度不均的影响。	5.0

Returns:

Type	Description
SearchEllipsoidParams	SearchEllipsoidParams，包含 6 个核心参数及辅助信息

Raises:

Type	Description
ValueError	点云数量不足（< 4）

Example::

params = compute_search_ellipsoid(points, voxel_size=5.0)
# 直接用于距离幂估值
eval_params = BlockModelDistancePowerParams(
    **params.to_block_model_params(),
    sample_file="样品.dmg",
    variables="CU",
    extra_attributes="样品点数目",
)

Source code in dimine_python_sdk\lib\prospecting\ellipsoid.py

def compute_search_ellipsoid(
    points: np.ndarray,
    voxel_size: Optional[float] = 5.0,
) -> SearchEllipsoidParams:
    """
    从点云计算搜索椭球体参数。

    流程：体素降采样 → MVEE → 角度分解。

    Args:
        points: 点云坐标，shape=(N, 3)，(x=东, y=北, z=高程)
        voxel_size: 体素降采样尺寸（米），设为 None 或 0 则跳过降采样。
                    推荐 5.0 m，可有效消除三角网密度不均的影响。

    Returns:
        SearchEllipsoidParams，包含 6 个核心参数及辅助信息

    Raises:
        ValueError: 点云数量不足（< 4）

    Example::

        params = compute_search_ellipsoid(points, voxel_size=5.0)
        # 直接用于距离幂估值
        eval_params = BlockModelDistancePowerParams(
            **params.to_block_model_params(),
            sample_file="样品.dmg",
            variables="CU",
            extra_attributes="样品点数目",
        )
    """
    if points.ndim != 2 or points.shape[1] < 3:
        raise ValueError(f"点云应为 (N, 3) 形状，当前: {points.shape}")

    if len(points) < 4:
        raise ValueError(f"点云数量不足：需要至少 4 个点，当前 {len(points)} 个")

    # 1. 体素降采样
    if voxel_size and voxel_size > 0:
        pts = _voxel_downsample(points, voxel_size)
    else:
        pts = points

    # 2. MVEE
    axes, directions, center = _compute_mvee(pts)

    # 3. 角度分解
    main_r, second_r, short_r, az, pl, dip = _decompose_angles(axes, directions)

    # 4. 比例
    second_main_rate = second_r / main_r if main_r > 0 else 0.0
    short_main_rate = short_r / main_r if main_r > 0 else 0.0

    return SearchEllipsoidParams(
        main_radius=round(main_r, 2),
        second_radius=round(second_r, 2),
        short_radius=round(short_r, 2),
        second_main_rate=round(second_main_rate, 4),
        short_main_rate=round(short_main_rate, 4),
        angle_main=round(az, 2),
        angle_second=round(pl, 2),
        angle_short=round(dip, 2),
        center=tuple(round(float(c), 2) for c in center),
        pc1_direction=tuple(round(float(v), 4) for v in directions[0]),
        pc2_direction=tuple(round(float(v), 4) for v in directions[1]),
        pc3_direction=tuple(round(float(v), 4) for v in directions[2]),
    )

compute_search_ellipsoid_from_dmf(dmf_path, voxel_size=5.0)

从 DMF 矿体模型文件直接计算搜索椭球体参数。

自动提取所有 Shell 实体的顶点，执行体素降采样与 MVEE 计算。

Parameters:

Name	Type	Description	Default
dmf_path	str	DMF 矿体模型文件路径	required
voxel_size	float	体素降采样尺寸（米），推荐 5.0 m	5.0

Returns:

Type	Description
SearchEllipsoidParams	SearchEllipsoidParams

Raises:

Type	Description
FileNotFoundError	DMF 文件不存在
ValueError	DMF 中无 Shell 实体

Source code in dimine_python_sdk\lib\prospecting\ellipsoid.py

def compute_search_ellipsoid_from_dmf(
    dmf_path: str,
    voxel_size: float = 5.0,
) -> SearchEllipsoidParams:
    """
    从 DMF 矿体模型文件直接计算搜索椭球体参数。

    自动提取所有 Shell 实体的顶点，执行体素降采样与 MVEE 计算。

    Args:
        dmf_path: DMF 矿体模型文件路径
        voxel_size: 体素降采样尺寸（米），推荐 5.0 m

    Returns:
        SearchEllipsoidParams

    Raises:
        FileNotFoundError: DMF 文件不存在
        ValueError: DMF 中无 Shell 实体
    """
    from pathlib import Path

    from dimine_python_sdk.lib.io.dmf_file import DmfFile

    path = Path(dmf_path)
    if not path.exists():
        raise FileNotFoundError(f"DMF 文件不存在: {dmf_path}")

    # 提取所有 Shell 顶点
    dmf = DmfFile(str(path))
    all_vertices = []
    for layer in dmf.layers:
        for entity in layer.entities:
            if hasattr(entity, 'type') and entity.type == 'shell':
                tin = entity.geometry
                if tin is not None and hasattr(tin, 'points'):
                    pts = np.asarray(tin.points, dtype=np.float64)
                    if pts.ndim == 2 and pts.shape[1] >= 3:
                        all_vertices.append(pts[:, :3])
    dmf.close()

    if not all_vertices:
        raise ValueError(f"DMF 中未找到 Shell 实体: {dmf_path}")

    points = np.vstack(all_vertices)
    return compute_search_ellipsoid(points, voxel_size=voxel_size)

convert_dmd_to_dmg(input_path, output_path)

将 .dmd 钻孔数据库直接转换为 .dmg 格式。

Parameters:

Name	Type	Description	Default
input_path	Union[str, Path]	源 .dmd 文件路径	required
output_path	Union[str, Path]	目标 .dmg 文件路径（无扩展名时自动补全）	required

Raises:

Type	Description
DrillDBLoadError	加载 .dmd 失败
DrillDBSaveError	保存 .dmg 失败

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def convert_dmd_to_dmg(input_path: Union[str, Path], output_path: Union[str, Path]) -> None:
    """
    将 .dmd 钻孔数据库直接转换为 .dmg 格式。

    Args:
        input_path: 源 .dmd 文件路径
        output_path: 目标 .dmg 文件路径（无扩展名时自动补全）

    Raises:
        DrillDBLoadError: 加载 .dmd 失败
        DrillDBSaveError: 保存 .dmg 失败
    """
    try:
        session = _NativeDrillSession.load(str(input_path))
    except Exception as exc:
        raise DrillDBLoadError(f"加载钻孔数据库失败: {input_path}") from exc

    save_path = _normalize_dmg_path(output_path)
    try:
        session.save_as_dmg(save_path)
    except Exception as exc:
        raise DrillDBSaveError(f"保存 .dmg 钻孔数据库失败: {save_path}") from exc

drill_conn(file_path=None, base_path=None)

钻孔数据库连接上下文管理器

Parameters:

Name	Type	Description	Default
file_path	Optional[str]	dmd 文件路径。为 None 时创建空对象，可赋值 DataFrame 后 save。	None
base_path	Optional[str]	基础路径，用于相对路径解析	None

Usage

with drill_conn("database.dmd") as db: print(db.collar.iloc[0]) df = db.collar

with drill_conn() as db: db.collar = df_collar db.save("output.dmd")

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

@contextmanager
def drill_conn(
    file_path: Optional[str] = None, base_path: Optional[str] = None
) -> Generator[DrillDBManager, None, None]:
    """
    钻孔数据库连接上下文管理器

    Args:
        file_path: dmd 文件路径。为 None 时创建空对象，可赋值 DataFrame 后 save。
        base_path: 基础路径，用于相对路径解析

    Usage:
        with drill_conn("database.dmd") as db:
            print(db.collar.iloc[0])
            df = db.collar

        with drill_conn() as db:
            db.collar = df_collar
            db.save("output.dmd")
    """
    db = DrillDBManager(file_path, base_path=base_path)
    try:
        yield db
    finally:
        db.close()

save_drill_database(collar_df, survey_df, lithology_df, save_path, sample_df=None, extend_tables=None)

保存钻孔数据库：子表 .dmt + .dmd 引用文件。

完全走 Python 层，不调用 C++ geo_drill_save；直接通过 NativeDataTableHandle 与底层 CDataTable 交互。

Parameters

extend_tables : List[Dict], optional 扩展表列表，每个元素为 dict： - df: pd.DataFrame 扩展表数据 - dmt_filename: str .dmt 文件名（如 "分层表.dmt"） - field_list: str 字段列表（如 "钻孔编号,从,至,地层"） 用于隐式地层建模等场景，最终写入 ExtendTableCount / ExtendTable{N}Path 等引用字段。

Source code in dimine_python_sdk\lib\prospecting\drill_db.py

def save_drill_database(
    collar_df: "pd.DataFrame",
    survey_df: "pd.DataFrame",
    lithology_df: "pd.DataFrame",
    save_path: str,
    sample_df: "pd.DataFrame" = None,
    extend_tables: Optional[List[Dict[str, Any]]] = None,
) -> None:
    """
    保存钻孔数据库：子表 .dmt + .dmd 引用文件。

    完全走 Python 层，不调用 C++ geo_drill_save；直接通过
    ``NativeDataTableHandle`` 与底层 CDataTable 交互。

    Parameters
    ----------
    extend_tables : List[Dict], optional
        扩展表列表，每个元素为 dict：
        - df: pd.DataFrame         扩展表数据
        - dmt_filename: str        .dmt 文件名（如 "分层表.dmt"）
        - field_list: str          字段列表（如 "钻孔编号,从,至,地层"）
        用于隐式地层建模等场景，最终写入 ExtendTableCount / ExtendTable{N}Path 等引用字段。
    """
    try:
        import pandas as pd
    except ImportError as exc:
        raise ImportError("使用 save_drill_database() 需要安装 pandas: uv add pandas") from exc

    path = Path(save_path).resolve()
    directory = path.parent
    stem = path.stem

    directory.mkdir(parents=True, exist_ok=True)

    dmt_paths = {
        "collar": directory / f"{stem}_collar.dmt",
        "survey": directory / f"{stem}_survey.dmt",
        "lithology": directory / f"{stem}_lithology.dmt",
    }

    NativeDataTableHandle.from_dataframe(collar_df).save(str(dmt_paths["collar"]))
    NativeDataTableHandle.from_dataframe(survey_df).save(str(dmt_paths["survey"]))

    tmp_collar_map = {}
    for name in collar_df.columns:
        if name in COLLAR_COLUMN_MAP:
            tmp_collar_map[COLLAR_COLUMN_MAP[name]] = name

    tmp_survey_map = {}
    for name in survey_df.columns:
        if name in SURVEY_COLUMN_MAP:
            tmp_survey_map[SURVEY_COLUMN_MAP[name]] = name

    table_count = 2
    data = {
        "TABLE_NAME": None,
        "DRILL_NAME": None,
        "INDEX_FIRST": None,
        "REPEAT_COUNT": None,
        "TABLE_MARK": 3,
        "COLLAR": f"{stem}_collar",
        "BHID": tmp_collar_map.get("BHID", None),
        "EAST": tmp_collar_map.get("EAST", None),
        "NOTRH": tmp_collar_map.get("NOTRH", None),
        "ELEVATION": tmp_collar_map.get("ELEVATION", None),
        "TOTALDEPTH": tmp_collar_map.get("TOTALDEPTH", None),
        "SECTION": tmp_collar_map.get("SECTION", None),
        "工程类型": "工程类型",
        "OTHERFIELDS": tmp_collar_map.get("OTHERFIELDS", None),
        "SURVEY": f"{stem}_survey",
        "SURBHID": tmp_survey_map.get("SURBHID", None),
        "SDEPTH": tmp_survey_map.get("SDEPTH", None),
        "AZIMUTH": tmp_survey_map.get("AZIMUTH", None),
        "DIP": tmp_survey_map.get("DIP", None),
    }

    if lithology_df is not None and not lithology_df.empty:
        NativeDataTableHandle.from_dataframe(lithology_df).save(
            str(dmt_paths["lithology"])
        )
        table_count += 1
        tmp_lithology_map = {}
        for name in lithology_df.columns:
            if name in LITHOLOGY_COLUMN_MAP:
                tmp_lithology_map[LITHOLOGY_COLUMN_MAP[name]] = name

        data.update({
            "LITHOLOGY": f"{stem}_lithology",
            "LITHBHID": tmp_lithology_map.get("LITHBHID", None),
            "LITHFROM": tmp_lithology_map.get("LITHFROM", None),
            "LITHTO": tmp_lithology_map.get("LITHTO", None),
            "ROCK-TYPE": tmp_lithology_map.get("ROCK-TYPE", None),
            "ElementList": None,
        })

    if sample_df is not None and not sample_df.empty:
        dmt_paths["sample"] = directory / f"{stem}_sample.dmt"
        NativeDataTableHandle.from_dataframe(sample_df).save(str(dmt_paths["sample"]))
        table_count += 1
        tmp_sample_map = {}
        for name in sample_df.columns:
            if name in SAMPLE_COLUMN_MAP:
                tmp_sample_map[SAMPLE_COLUMN_MAP[name]] = name

        data.update({
            "SAMPLE": f"{stem}_sample",
            "SAMBHID": tmp_sample_map.get("SAMBHID", None),
            "SAMPLE-ID": tmp_sample_map.get("SAMPLE-ID", None),
            "SAMFROM": tmp_sample_map.get("SAMFROM", None),
            "SAMTO": tmp_sample_map.get("SAMTO", None),
        })

    if extend_tables:
        ext_count = len(extend_tables)
        data["ExtendTableCount"] = str(ext_count)
        for i, ext in enumerate(extend_tables, start=1):
            ext_df = ext["df"]
            ext_filename = ext["dmt_filename"]
            ext_fields = ext["field_list"]
            ext_dmt_path = directory / ext_filename
            NativeDataTableHandle.from_dataframe(ext_df).save(str(ext_dmt_path))
            data[f"ExtendTable{i}Path"] = ext_filename
            data[f"ExtendTable{i}FieldList"] = ext_fields
            table_count += 1

    data["TABLE_MARK"] = table_count
    ref_df = pd.DataFrame([data])
    ref_df.TABLE_MARK = ref_df.TABLE_MARK.astype(int)

    _DMD_EXCLUDED = {"TABLE_NAME", "DRILL_NAME", "INDEX_FIRST", "REPEAT_COUNT", "ElementList"}
    for k, v in data.items():
        if v is None:
            _DMD_EXCLUDED.add(k)

    handle = NativeDataTableHandle.create()
    handle.insert_from_dataframe(
        ref_df, exclude_columns=list(_DMD_EXCLUDED)
    )
    handle.save(str(path))

钻孔数据库 DMD 转 DMG

除上述应用层 API 外，lib.prospecting 还提供独立的 convert_dmd_to_dmg() 函数，用于将数据表形式存储的钻孔数据库（.dmd）打包为三维模型 + 数据形式（.dmg），无需实例化 DrillDBManager。

from dimine_python_sdk.lib.prospecting import convert_dmd_to_dmg

convert_dmd_to_dmg("input.dmd", "output.dmg")