Skip to content

api

lib.prospecting — 勘探/储量模块

BlockDataError

Bases: RuntimeError

块段模型操作异常基类

Source code in dimine_python_sdk\lib\prospecting\block_model.py 
48
49
50
class BlockDataError(RuntimeError):
    """块段模型操作异常基类"""
    pass

BlockModelCommonParams

Bases: ParamModel

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

Source code in dimine_python_sdk\lib\prospecting\models.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class BlockModelCommonParams(ParamModel):
    """块段模型估值通用参数基类（距离幂/克里格共用）"""

    sample_file: str = Field(..., description="dmg样品数据文件路径")
    variables: list[str] = Field(..., alias="variable", description="估值变量即元素列表（如 [CU, AU, AG] 或 [CU]），元素对应钻孔数据库中元素名称")
    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="短轴离散块数，在计算某个块品位储量时，可以先分割成多个子块用于更精确计算。")
    ellipsoid_azimuth: Optional[float] = Field(default=10, alias="angle_main", description="椭球体参数：方位角Azimuth：主轴方向角度（度）")
    ellipsoid_plunge: Optional[float] = Field(default=0, alias="angle_second", description="椭球体参数：倾伏角Plunge：次轴方向角度（度）")
    ellipsoid_dip: Optional[float] = Field(default=15, alias="angle_short", description="椭球体参数：倾角Dip：短方向角度（度）")
    ellipsoid_main_radius: Optional[float] = Field(default=1536, gt=0, alias="main_radius", description="椭球体参数：主轴半径。主方向搜索半径,单位m")
    ellipsoid_second_main_rate: Optional[float] = Field(
        default=0.8, ge=0, le=1, alias="second_main_rate", description="椭球体参数：次方向搜索半径比例（相对于主方向）即：次轴/主轴 的比例，根据主轴半径可以算出次轴半径"
    )
    ellipsoid_short_main_rate: Optional[float] = Field(
        default=0.2, ge=0, le=1, alias="short_main_rate", 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="单工程样品数最大值")

    def model_dump_json(self, **kwargs):
        """序列化为 JSON 字符串，默认使用 alias 输出旧字段名以兼容 C++ 后端"""
        kwargs.setdefault('by_alias', True)
        return super().model_dump_json(**kwargs)

    @field_serializer('variables')
    def _serialize_variables(self, value: list[str]) -> str:
        """序列化时自动将元素列表合并为逗号分隔字符串，兼容 C++ 后端"""
        return ",".join(value)

model_dump_json(**kwargs)

序列化为 JSON 字符串，默认使用 alias 输出旧字段名以兼容 C++ 后端

Source code in dimine_python_sdk\lib\prospecting\models.py 
39
40
41
42
def model_dump_json(self, **kwargs):
    """序列化为 JSON 字符串，默认使用 alias 输出旧字段名以兼容 C++ 后端"""
    kwargs.setdefault('by_alias', True)
    return super().model_dump_json(**kwargs)

BlockModelCreateError

Bases: BlockDataError

创建块段模型失败

Source code in dimine_python_sdk\lib\prospecting\block_model.py 
53
54
55
class BlockModelCreateError(BlockDataError):
    """创建块段模型失败"""
    pass

BlockModelDistancePowerParams

Bases: BlockModelCommonParams

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

Source code in dimine_python_sdk\lib\prospecting\models.py 
50
51
52
53
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

估值失败
Source code in dimine_python_sdk\lib\prospecting\block_model.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
class BlockModelEvaluator:
    """
    块段模型估值与储量计算器

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

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

    Raises:
        BlockModelEvaluationError: 估值失败
    """

    @staticmethod
    def distance_power_evaluation(
        block_model_file: str,
        constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
        evaluation_params: BlockModelDistancePowerParams,
        overwrite_result: bool = True,
    ) -> DistancePowerEvaluationResult:
        """距离幂估值计算

        Returns:
            DistancePowerEvaluationResult: 估值结果对象，包含 success / message / elements
        """
        constraint_json = _serialize_constraints(constraint_params)
        evaluation_json = json.dumps(evaluation_params.to_dict(), ensure_ascii=False)
        try:
            result = _native_distance_power_evaluation(
                block_model_file, constraint_json, evaluation_json, overwrite_result
            )
            logger.info(f"距离幂估值结果: {result}")
            message = result.get("message", "")
            elements = _parse_evaluation_message(message)
            return DistancePowerEvaluationResult(
                success=True,
                elements=elements,
            )
        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:
        """克里格估值计算

        克里格算法底层仅支持单元素估值，当 variables 包含多个元素（逗号分隔）时，
        自动按元素拆分，依次调用底层接口实现多元素估值效果。

        """
        constraint_json = _serialize_constraints(constraint_params)
        params_dict = evaluation_params.to_dict()

        # 估值变量：直接使用 list[str]，逐个调用克里格估值
        elements = evaluation_params.variables
        if not elements:
            raise BlockModelEvaluationError("克里格估值失败: variables 不能为空")

        results = []
        for element in elements:
            params_dict["variable"] = element
            evaluation_json = json.dumps(params_dict, ensure_ascii=False)
            try:
                result = _native_kriging_evaluation(
                    block_model_file, constraint_json, evaluation_json, overwrite_result
                )
                logger.info(f"克里格估值结果（元素={element}）: {result}")
                results.append(result)
            except Exception as exc:
                raise BlockModelEvaluationError(
                    f"克里格估值失败（元素={element}）: {exc}"
                ) from exc

        # 返回最后一个结果（与底层约定一致），附加全部结果列表
        final = results[-1] if results else {}
        final["all_results"] = results
        return final

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

距离幂估值计算

Returns:

Name Type Description
DistancePowerEvaluationResult DistancePowerEvaluationResult

估值结果对象，包含 success / message / elements
Source code in dimine_python_sdk\lib\prospecting\block_model.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@staticmethod
def distance_power_evaluation(
    block_model_file: str,
    constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
    evaluation_params: BlockModelDistancePowerParams,
    overwrite_result: bool = True,
) -> DistancePowerEvaluationResult:
    """距离幂估值计算

    Returns:
        DistancePowerEvaluationResult: 估值结果对象，包含 success / message / elements
    """
    constraint_json = _serialize_constraints(constraint_params)
    evaluation_json = json.dumps(evaluation_params.to_dict(), ensure_ascii=False)
    try:
        result = _native_distance_power_evaluation(
            block_model_file, constraint_json, evaluation_json, overwrite_result
        )
        logger.info(f"距离幂估值结果: {result}")
        message = result.get("message", "")
        elements = _parse_evaluation_message(message)
        return DistancePowerEvaluationResult(
            success=True,
            elements=elements,
        )
    except Exception as exc:
        raise BlockModelEvaluationError(f"距离幂估值失败: {exc}") from exc

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

克里格估值计算

克里格算法底层仅支持单元素估值，当 variables 包含多个元素（逗号分隔）时， 自动按元素拆分，依次调用底层接口实现多元素估值效果。

Source code in dimine_python_sdk\lib\prospecting\block_model.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@staticmethod
def kriging_evaluation(
    block_model_file: str,
    constraint_params: list[BlockModelConstraintItem] | BlockModelConstraintItem,
    evaluation_params: BlockModelKrigingParams,
    overwrite_result: bool = True,
) -> dict:
    """克里格估值计算

    克里格算法底层仅支持单元素估值，当 variables 包含多个元素（逗号分隔）时，
    自动按元素拆分，依次调用底层接口实现多元素估值效果。

    """
    constraint_json = _serialize_constraints(constraint_params)
    params_dict = evaluation_params.to_dict()

    # 估值变量：直接使用 list[str]，逐个调用克里格估值
    elements = evaluation_params.variables
    if not elements:
        raise BlockModelEvaluationError("克里格估值失败: variables 不能为空")

    results = []
    for element in elements:
        params_dict["variable"] = element
        evaluation_json = json.dumps(params_dict, ensure_ascii=False)
        try:
            result = _native_kriging_evaluation(
                block_model_file, constraint_json, evaluation_json, overwrite_result
            )
            logger.info(f"克里格估值结果（元素={element}）: {result}")
            results.append(result)
        except Exception as exc:
            raise BlockModelEvaluationError(
                f"克里格估值失败（元素={element}）: {exc}"
            ) from exc

    # 返回最后一个结果（与底层约定一致），附加全部结果列表
    final = results[-1] if results else {}
    final["all_results"] = results
    return final

BlockModelKrigingParams

Bases: BlockModelCommonParams

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

Source code in dimine_python_sdk\lib\prospecting\models.py 
56
57
58
59
60
61
62
63
64
65
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="变差函数参数列表",
    )

CreateBlockModelParams

Bases: ParamModel

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

Source code in dimine_python_sdk\lib\prospecting\models.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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 
86
87
88
89
90
91
92
93
94
95
96
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"'
    )

DrillFunctionWrapper

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

Source code in dimine_python_sdk\lib\prospecting\drill_db.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class DrillFunctionWrapper:
    """
    钻孔相关功能的封装器，提供更简洁的接口
    """

    @staticmethod
    def sample_length_combine(params: SampleLengthCombineParam) -> dict:
        """样长组合"""
        if params.combine_percent is not None:
            params.combine_percent = int(params.combine_percent * 100)
        if params.output_file is None:
            stem = Path(params.input_file).stem
            params.output_file = f"{stem}_combined.dmg"
        return _native_sample_length_combine(params.model_dump_json(by_alias=True))

    @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 _native_step_combine(params.model_dump_json(by_alias=True))

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

extra_high_grade_process(params) staticmethod

特高品位处理

Source code in dimine_python_sdk\lib\prospecting\drill_db.py 
58
59
60
61
62
63
@staticmethod
def extra_high_grade_process(params: HighGradeProcessParam) -> dict:
    """特高品位处理"""
    if params.result_field is None:
        params.result_field = params.grade_field
    return _native_extra_high_grade_process(params.model_dump_json(by_alias=True))

sample_length_combine(params) staticmethod

样长组合

Source code in dimine_python_sdk\lib\prospecting\drill_db.py 
40
41
42
43
44
45
46
47
48
@staticmethod
def sample_length_combine(params: SampleLengthCombineParam) -> dict:
    """样长组合"""
    if params.combine_percent is not None:
        params.combine_percent = int(params.combine_percent * 100)
    if params.output_file is None:
        stem = Path(params.input_file).stem
        params.output_file = f"{stem}_combined.dmg"
    return _native_sample_length_combine(params.model_dump_json(by_alias=True))

step_combine(params) staticmethod

台阶组合

Source code in dimine_python_sdk\lib\prospecting\drill_db.py 
50
51
52
53
54
55
56
@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 _native_step_combine(params.model_dump_json(by_alias=True))

EntityConstraint

Bases: ParamModel

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

Source code in dimine_python_sdk\lib\prospecting\models.py 
73
74
75
76
77
78
79
80
81
82
83
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 
207
208
209
210
211
212
213
214
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} 即源品位字段名，处理后的结果会把原始数据覆盖）
Source code in dimine_python_sdk\lib\prospecting\high_grade.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@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} 即源品位字段名，处理后的结果会把原始数据覆盖）
    """

    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 = self.grade_field

HighGradeProcessParam

Bases: ParamModel

特高品位处理参数

Source code in dimine_python_sdk\lib\prospecting\models.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
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,
        alias="assign_valule",
        description="给定值方法对应的值,(replace_method=1时，默认值为0， replace_method=2时，默认值为2)",
    )
    adjoin_average: Optional[float] = Field(default=2, description="相邻样品平均值,只有replace_method=2时才有效")
    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}）"
    )

HighGradeProcessor

特高品位处理器

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

Source code in dimine_python_sdk\lib\prospecting\high_grade.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
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

        .. deprecated::
            此方法为纯 Python 实现，已废弃。请改用 C++ 原生接口：
            DrillFunctionWrapper.extra_high_grade_process() 或
            DrillFunctionWrapper.extra_high_grade_process()。
        """
        warnings.warn(
            "HighGradeProcessor.process() 已废弃，"
            "请改用 C++ 原生接口: DrillFunctionWrapper.extra_high_grade_process() "
            "或 DrillFunctionWrapper.extra_high_grade_process()。"
            "详见模块文档。",
            DeprecationWarning,
            stacklevel=3,
        )
        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

.. deprecated:: 此方法为纯 Python 实现，已废弃。请改用 C++ 原生接口： DrillFunctionWrapper.extra_high_grade_process() 或 DrillFunctionWrapper.extra_high_grade_process()。

Source code in dimine_python_sdk\lib\prospecting\high_grade.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
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

    .. deprecated::
        此方法为纯 Python 实现，已废弃。请改用 C++ 原生接口：
        DrillFunctionWrapper.extra_high_grade_process() 或
        DrillFunctionWrapper.extra_high_grade_process()。
    """
    warnings.warn(
        "HighGradeProcessor.process() 已废弃，"
        "请改用 C++ 原生接口: DrillFunctionWrapper.extra_high_grade_process() "
        "或 DrillFunctionWrapper.extra_high_grade_process()。"
        "详见模块文档。",
        DeprecationWarning,
        stacklevel=3,
    )
    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 
15
16
17
18
19
20
21
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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"'
    )

ReservesCalculateParam

Bases: ParamModel

估值算量参数

Source code in dimine_python_sdk\lib\prospecting\models.py 
198
199
200
201
202
203
204
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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
@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 
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
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

        .. deprecated::
            此方法为纯 Python 实现，已废弃。请改用 C++ 原生接口：
            DrillFunctionWrapper.sample_length_combine() 或
            DrillFunctionWrapper.sample_length_combine()。
        """
        warnings.warn(
            "SampleCompositingProcessor.process() 已废弃，"
            "请改用 C++ 原生接口: DrillFunctionWrapper.sample_length_combine() "
            "或 DmdFile。详见模块文档。",
            DeprecationWarning,
            stacklevel=3,
        )
        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 
674
675
676
677
678
679
680
681
682
683
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
Dict[str, Any]

summary — 汇总统计 dict

.. deprecated:: 此方法为纯 Python 实现，已废弃。请改用 C++ 原生接口： DrillFunctionWrapper.sample_length_combine() 或 DrillFunctionWrapper.sample_length_combine()。

Source code in dimine_python_sdk\lib\prospecting\compositing.py 
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
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

    .. deprecated::
        此方法为纯 Python 实现，已废弃。请改用 C++ 原生接口：
        DrillFunctionWrapper.sample_length_combine() 或
        DrillFunctionWrapper.sample_length_combine()。
    """
    warnings.warn(
        "SampleCompositingProcessor.process() 已废弃，"
        "请改用 C++ 原生接口: DrillFunctionWrapper.sample_length_combine() "
        "或 DmdFile。详见模块文档。",
        DeprecationWarning,
        stacklevel=3,
    )
    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 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
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 − ellipsoid_azimuth) × Ry(ellipsoid_plunge) × Rx(ellipsoid_dip)

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

Source code in dimine_python_sdk\lib\prospecting\ellipsoid.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class SearchEllipsoidParams(ParamModel):
    """搜索椭球体参数，可直接映射到 ``BlockModelDistancePowerParams``。

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

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

    # ---- 形状参数 ----
    ellipsoid_main_radius: float = Field(..., alias="main_radius")
    """主轴半径（m），椭球体最长半轴"""
    second_radius: float
    """次轴半径（m），椭球体中等半轴"""
    short_radius: float
    """短轴半径（m），椭球体最短半轴"""
    ellipsoid_second_main_rate: float = Field(..., alias="second_main_rate")
    """次轴/主轴比例"""
    ellipsoid_short_main_rate: float = Field(..., alias="short_main_rate")
    """短轴/主轴比例"""

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

    # ---- 辅助信息 ----
    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.ellipsoid_main_radius,
            "second_main_rate": self.ellipsoid_second_main_rate,
            "short_main_rate": self.ellipsoid_short_main_rate,
            "angle_main": self.ellipsoid_azimuth,
            "angle_second": self.ellipsoid_plunge,
            "angle_short": self.ellipsoid_dip,
        }

center instance-attribute

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

ellipsoid_azimuth = Field(..., alias='angle_main') class-attribute instance-attribute

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

ellipsoid_dip = Field(..., alias='angle_short') class-attribute instance-attribute

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

ellipsoid_main_radius = Field(..., alias='main_radius') class-attribute instance-attribute

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

ellipsoid_plunge = Field(..., alias='angle_second') class-attribute instance-attribute

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

ellipsoid_second_main_rate = Field(..., alias='second_main_rate') class-attribute instance-attribute

次轴/主轴比例

ellipsoid_short_main_rate = Field(..., alias='short_main_rate') class-attribute instance-attribute

短轴/主轴比例

pc1_direction instance-attribute

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

pc2_direction instance-attribute

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

pc3_direction instance-attribute

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

second_radius instance-attribute

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

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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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.ellipsoid_main_radius,
        "second_main_rate": self.ellipsoid_second_main_rate,
        "short_main_rate": self.ellipsoid_short_main_rate,
        "angle_main": self.ellipsoid_azimuth,
        "angle_second": self.ellipsoid_plunge,
        "angle_short": self.ellipsoid_dip,
    }

StepCombineParam

Bases: ParamModel

台阶组合参数

Source code in dimine_python_sdk\lib\prospecting\models.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
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 
 99
100
101
102
103
104
105
106
107
108
109
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 
146
147
148
149
150
151
152
153
154
155
156
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 
159
160
161
162
163
164
165
166
167
168
169
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 
172
173
174
175
176
177
178
179
180
181
182
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"'
    )

block_constrain_to_dmc_file(block_model_file, constrain_params, constrain_result_file)

块约束保存到约束结果 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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
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

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. 从 DmdFile 直接传入： result = composite_samples(dmd, 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

DmdFile 实例 或 样品表 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) 元组

.. deprecated:: 此函数为纯 Python 实现，已废弃。请改用 C++ 原生接口。

Source code in dimine_python_sdk\lib\prospecting\compositing.py 
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
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. 从 DmdFile 直接传入：
           result = composite_samples(dmd, 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: DmdFile 实例 或 样品表 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) 元组

    .. deprecated::
        此函数为纯 Python 实现，已废弃。请改用 C++ 原生接口。
    """
    warnings.warn(
        "composite_samples() 已废弃，请改用 C++ 原生接口: "
        "DrillFunctionWrapper.sample_length_combine() 或 "
        "DrillFunctionWrapper.sample_length_combine()。详见模块文档。",
        DeprecationWarning,
        stacklevel=3,
    )
    from dimine_python_sdk.lib.io.dmd_file import DmdFile

    if isinstance(db_or_samples, DmdFile):
        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 必须是 DmdFile 或 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 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
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(
        ellipsoid_main_radius=round(main_r, 2),
        second_radius=round(second_r, 2),
        short_radius=round(short_r, 2),
        ellipsoid_second_main_rate=round(second_main_rate, 4),
        ellipsoid_short_main_rate=round(short_main_rate, 4),
        ellipsoid_azimuth=round(az, 2),
        ellipsoid_plunge=round(pl, 2),
        ellipsoid_dip=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 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
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)

create_block_model(params)

创建空块段模型

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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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

钻孔数据库 DMD 转 DMG

除上述应用层 API 外，可使用 dimine_python_sdk.lib.io.file_converter 中的 convert_file() 函数完成 .dmd → .dmg 转换：

from dimine_python_sdk.lib.io import convert_file

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