prospecting

Native Layer v2 勘探模块

钻孔数据库与块段模型封装，对外只暴露 pd.DataFrame / dict / numpy。

BlockModelSession

块段模型会话，内部持有 C++ dmBlockDataNew 对象

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

class BlockModelSession:
    """块段模型会话，内部持有 C++ dmBlockDataNew 对象"""

    def __init__(self, cpp_obj: Any, file_path: str | None = None):
        self._cpp_obj = cpp_obj
        self._file_path = file_path

    @classmethod
    def open(cls, file_path: str) -> "BlockModelSession":
        """加载块段模型"""
        cpp_obj = Dm.dmBlockDataNew()
        if not cpp_obj.Open(file_path):
            raise NativeProspectingError(f"加载块段模型失败: {file_path}")
        return cls(cpp_obj, file_path)

    @property
    def origin(self) -> "np.ndarray":
        """模型原点"""
        import numpy as np

        pt = self._cpp_obj.GetOrigin()
        return np.array([pt.x, pt.y, pt.z], dtype=float)

    @property
    def xyz_length(self) -> list[float]:
        """模型 X/Y/Z 方向总长度"""
        return list(self._cpp_obj.GetXYZLen())

    @property
    def max_level(self) -> int:
        """最大精度层级"""
        return self._cpp_obj.GetMaxLevel()

    @property
    def min_size(self) -> list[float]:
        """最小块段尺寸"""
        return list(self._cpp_obj.GetMinSize())

    @property
    def rotation(self) -> list[float]:
        """模型旋转角度"""
        return list(self._cpp_obj.GetRotation())

    @property
    def field_definitions(self) -> list[NativeFieldDef]:
        """字段定义列表"""
        defs = self._cpp_obj.GetFieldDefine()
        return [NativeFieldDef(name=d["name"], type=d["type"]) for d in defs]

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

    def iter_blocks(self, field_name: str) -> Iterator[dict]:
        """遍历所有块段，逐块返回指定字段数据"""
        self._cpp_obj.InitCursor()
        while self._cpp_obj.AdvCursor() and not self._cpp_obj.IsBeyondBoundaryNode():
            success, data = self._cpp_obj.GetCursorData(field_name)
            if success:
                try:
                    yield json.loads(data)
                except (ValueError, TypeError):
                    yield {"_raw": str(data)}

field_definitions property

字段定义列表

max_level property

最大精度层级

min_size property

最小块段尺寸

origin property

模型原点

rotation property

模型旋转角度

total_block_count property

块段总数量

xyz_length property

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

iter_blocks(field_name)

遍历所有块段，逐块返回指定字段数据

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

def iter_blocks(self, field_name: str) -> Iterator[dict]:
    """遍历所有块段，逐块返回指定字段数据"""
    self._cpp_obj.InitCursor()
    while self._cpp_obj.AdvCursor() and not self._cpp_obj.IsBeyondBoundaryNode():
        success, data = self._cpp_obj.GetCursorData(field_name)
        if success:
            try:
                yield json.loads(data)
            except (ValueError, TypeError):
                yield {"_raw": str(data)}

open(file_path) classmethod

加载块段模型

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

@classmethod
def open(cls, file_path: str) -> "BlockModelSession":
    """加载块段模型"""
    cpp_obj = Dm.dmBlockDataNew()
    if not cpp_obj.Open(file_path):
        raise NativeProspectingError(f"加载块段模型失败: {file_path}")
    return cls(cpp_obj, file_path)

DrillSession

钻孔数据库会话，内部持有 C++ dmGeoDrillData 对象

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

class DrillSession:
    """钻孔数据库会话，内部持有 C++ dmGeoDrillData 对象"""

    def __init__(self, cpp_obj: Any, ref_dict: dict):
        self._cpp_obj = cpp_obj
        self._ref_dict = ref_dict

    @classmethod
    def load(cls, file_path: str) -> "DrillSession":
        """加载钻孔数据库"""
        cpp_obj = Dm.dmGeoDrillData()
        if not cpp_obj.Load(file_path):
            raise NativeProspectingError(f"加载钻孔数据库失败: {file_path}")
        ref_dict = _load_dmd_reference(file_path)
        return cls(cpp_obj, ref_dict)

    @classmethod
    def create_empty(cls) -> "DrillSession":
        """创建空钻孔数据库"""
        return cls(Dm.dmGeoDrillData(), {})

    @property
    def ref_dict(self) -> dict:
        return self._ref_dict

    def _get_table(self, getter_name: str, ref_table_key: str, ref_keys: set, std_map: dict, empty_columns: list[str]) -> "pd.DataFrame":
        """通用获取子表为 DataFrame"""
        require_pandas()
        import pandas as pd

        if not self._ref_dict or not self._ref_dict.get(ref_table_key):
            return pd.DataFrame(columns=empty_columns)

        cpp_table = getattr(self._cpp_obj, getter_name)()
        handle = NativeDataTableHandle(cpp_table)
        df = handle.to_dataframe()

        define = _build_define_from_ref(self._ref_dict, ref_keys)
        return _rename_with_define(df, define, std_map)

    def _set_table(
        self,
        df: "pd.DataFrame",
        getter_name: str,
        mapping: dict,
        required: list[str],
        ref_keys: set,
    ) -> None:
        """通用设置子表"""
        require_pandas()
        _validate_columns(df, mapping, getter_name, required)

        cpp_table = getattr(self._cpp_obj, getter_name)()
        handle = NativeDataTableHandle(cpp_table)
        handle.clear_records()

        define = _build_define_from_ref(self._ref_dict, ref_keys)
        column_mapping = {}
        if define:
            for std_key, actual_col in define.items():
                if actual_col in df.columns:
                    column_mapping[actual_col] = std_key
        for chinese, english in mapping.items():
            if chinese in df.columns and chinese not in column_mapping:
                column_mapping[chinese] = english

        handle.insert_from_dataframe(df, column_mapping=column_mapping)

    def collar_table(self) -> "pd.DataFrame":
        """获取孔口表"""
        return self._get_table(
            "GetCollarTable",
            "COLLAR",
            _COLLAR_REF_KEYS,
            _COLLAR_STD,
            list(COLLAR_COLUMN_MAP.keys()),
        )

    def set_collar_table(self, df: "pd.DataFrame") -> None:
        """设置孔口表"""
        self._set_table(df, "GetCollarTable", COLLAR_COLUMN_MAP, _COLLAR_REQUIRED, _COLLAR_REF_KEYS)

    def survey_table(self) -> "pd.DataFrame":
        """获取测斜表"""
        return self._get_table(
            "GetSurveyTable",
            "SURVEY",
            _SURVEY_REF_KEYS,
            _SURVEY_STD,
            list(SURVEY_COLUMN_MAP.keys()),
        )

    def set_survey_table(self, df: "pd.DataFrame") -> None:
        """设置测斜表"""
        self._set_table(df, "GetSurveyTable", SURVEY_COLUMN_MAP, _SURVEY_REQUIRED, _SURVEY_REF_KEYS)

    def lithology_table(self) -> "pd.DataFrame":
        """获取岩性表"""
        return self._get_table(
            "GetLithologyTable",
            "LITHOLOGY",
            _LITHOLOGY_REF_KEYS,
            _LITHOLOGY_STD,
            list(LITHOLOGY_COLUMN_MAP.keys()),
        )

    def set_lithology_table(self, df: "pd.DataFrame") -> None:
        """设置岩性表"""
        self._set_table(df, "GetLithologyTable", LITHOLOGY_COLUMN_MAP, _LITHOLOGY_REQUIRED, _LITHOLOGY_REF_KEYS)

    def sample_table(self) -> "pd.DataFrame":
        """获取样品表"""
        return self._get_table(
            "GetSampleTable",
            "SAMPLE",
            _SAMPLE_REF_KEYS,
            _SAMPLE_STD,
            list(SAMPLE_COLUMN_MAP.keys()),
        )

    def set_sample_table(self, df: "pd.DataFrame") -> None:
        """设置样品表"""
        self._set_table(df, "GetSampleTable", SAMPLE_COLUMN_MAP, _SAMPLE_REQUIRED, _SAMPLE_REF_KEYS)

    def save_as_dmg(self, file_path: str) -> None:
        """保存为 .dmg 格式"""
        try:
            self._cpp_obj.Save(file_path)
        except Exception as exc:
            raise NativeProspectingError(f"保存 .dmg 失败: {file_path}") from exc

collar_table()

获取孔口表

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

def collar_table(self) -> "pd.DataFrame":
    """获取孔口表"""
    return self._get_table(
        "GetCollarTable",
        "COLLAR",
        _COLLAR_REF_KEYS,
        _COLLAR_STD,
        list(COLLAR_COLUMN_MAP.keys()),
    )

create_empty() classmethod

创建空钻孔数据库

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

@classmethod
def create_empty(cls) -> "DrillSession":
    """创建空钻孔数据库"""
    return cls(Dm.dmGeoDrillData(), {})

lithology_table()

获取岩性表

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

def lithology_table(self) -> "pd.DataFrame":
    """获取岩性表"""
    return self._get_table(
        "GetLithologyTable",
        "LITHOLOGY",
        _LITHOLOGY_REF_KEYS,
        _LITHOLOGY_STD,
        list(LITHOLOGY_COLUMN_MAP.keys()),
    )

load(file_path) classmethod

加载钻孔数据库

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

@classmethod
def load(cls, file_path: str) -> "DrillSession":
    """加载钻孔数据库"""
    cpp_obj = Dm.dmGeoDrillData()
    if not cpp_obj.Load(file_path):
        raise NativeProspectingError(f"加载钻孔数据库失败: {file_path}")
    ref_dict = _load_dmd_reference(file_path)
    return cls(cpp_obj, ref_dict)

sample_table()

获取样品表

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

def sample_table(self) -> "pd.DataFrame":
    """获取样品表"""
    return self._get_table(
        "GetSampleTable",
        "SAMPLE",
        _SAMPLE_REF_KEYS,
        _SAMPLE_STD,
        list(SAMPLE_COLUMN_MAP.keys()),
    )

save_as_dmg(file_path)

保存为 .dmg 格式

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

def save_as_dmg(self, file_path: str) -> None:
    """保存为 .dmg 格式"""
    try:
        self._cpp_obj.Save(file_path)
    except Exception as exc:
        raise NativeProspectingError(f"保存 .dmg 失败: {file_path}") from exc

set_collar_table(df)

设置孔口表

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

def set_collar_table(self, df: "pd.DataFrame") -> None:
    """设置孔口表"""
    self._set_table(df, "GetCollarTable", COLLAR_COLUMN_MAP, _COLLAR_REQUIRED, _COLLAR_REF_KEYS)

set_lithology_table(df)

设置岩性表

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

def set_lithology_table(self, df: "pd.DataFrame") -> None:
    """设置岩性表"""
    self._set_table(df, "GetLithologyTable", LITHOLOGY_COLUMN_MAP, _LITHOLOGY_REQUIRED, _LITHOLOGY_REF_KEYS)

set_sample_table(df)

设置样品表

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

def set_sample_table(self, df: "pd.DataFrame") -> None:
    """设置样品表"""
    self._set_table(df, "GetSampleTable", SAMPLE_COLUMN_MAP, _SAMPLE_REQUIRED, _SAMPLE_REF_KEYS)

set_survey_table(df)

设置测斜表

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

def set_survey_table(self, df: "pd.DataFrame") -> None:
    """设置测斜表"""
    self._set_table(df, "GetSurveyTable", SURVEY_COLUMN_MAP, _SURVEY_REQUIRED, _SURVEY_REF_KEYS)

survey_table()

获取测斜表

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

def survey_table(self) -> "pd.DataFrame":
    """获取测斜表"""
    return self._get_table(
        "GetSurveyTable",
        "SURVEY",
        _SURVEY_REF_KEYS,
        _SURVEY_STD,
        list(SURVEY_COLUMN_MAP.keys()),
    )

block_constrain_to_dmc_file(block_model_file, constraint_json, output_file)

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

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

def block_constrain_to_dmc_file(
    block_model_file: str,
    constraint_json: str,
    output_file: str,
) -> None:
    """块约束保存到约束结果 dmc 文件"""
    ok = Dm.BlockConstrainToDmcFile(block_model_file, constraint_json, output_file)
    if not ok:
        raise NativeProspectingError(
            f"块约束保存失败: {block_model_file} -> {output_file}"
        )

calculate_reserves(block_model_file, constraint_json, reserve_json)

储量计算

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

def calculate_reserves(
    block_model_file: str,
    constraint_json: str,
    reserve_json: str,
) -> dict:
    """储量计算"""
    result = Dm.ReservesCalculate(
        block_model_file,
        _serialize_constraints(constraint_json),
        _serialize_param(reserve_json),
    )
    message = check_bool_message(result, NativeProspectingError, "储量计算")
    return {"success": True, "message": message}

create_block_model(params)

创建空块段模型

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

def create_block_model(params: dict | str) -> dict:
    """创建空块段模型"""
    t = _serialize_param(params)
    raw = Dm.CreateBlockModel(t)
    return parse_json_response(raw, NativeProspectingError, "创建块段模型")

distance_power_evaluation(block_model_file, constraint_json, evaluation_json, overwrite=True)

距离幂估值

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

def distance_power_evaluation(
    block_model_file: str,
    constraint_json: str,
    evaluation_json: str,
    overwrite: bool = True,
) -> dict:
    """距离幂估值"""
    print(f"[DEBUG] Dm.DistancePowerEvaluationValue 入参:")
    print(f"  block_model_file: {block_model_file}")
    print(f"  constraint_json: {constraint_json}")
    print(f"  evaluation_json: {evaluation_json}")
    print(f"  overwrite: {overwrite}")

    result = Dm.DistancePowerEvaluationValue(
        block_model_file,
        _serialize_constraints(constraint_json),
        _serialize_param(evaluation_json),
        overwrite,
    )
    message = check_bool_message(result, NativeProspectingError, "距离幂估值")
    return {"success": True, "message": message}

extra_high_grade_process(param)

特高品位处理

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

def extra_high_grade_process(param: dict | str) -> dict:
    """特高品位处理"""
    raw = Dm.ExtraHighGradeProcess(_serialize_param(param))
    return parse_json_response(raw, NativeProspectingError, "特高品位处理")

kriging_evaluation(block_model_file, constraint_json, evaluation_json, overwrite=True)

克里格估值

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

def kriging_evaluation(
    block_model_file: str,
    constraint_json: str,
    evaluation_json: str,
    overwrite: bool = True,
) -> dict:
    """克里格估值"""
    result = Dm.KrigEvaluationValue(
        block_model_file,
        _serialize_constraints(constraint_json),
        _serialize_param(evaluation_json),
        overwrite,
    )
    message = check_bool_message(result, NativeProspectingError, "克里格估值")
    return {"success": True, "message": message}

sample_length_combine(param)

样长组合

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

def sample_length_combine(param: dict | str) -> dict:
    """样长组合"""
    raw = Dm.SampleLengthCombine(_serialize_param(param))
    return parse_json_response(raw, NativeProspectingError, "样长组合")

step_combine(param)

台阶组合

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

def step_combine(param: dict | str) -> dict:
    """台阶组合"""
    raw = Dm.StepCombine(_serialize_param(param))
    return parse_json_response(raw, NativeProspectingError, "台阶组合")