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]:
        """最小块段尺寸（对应 max_level 层级的块段尺寸）"""
        return list(self._cpp_obj.GetMinSize())

    def get_level_size(self, level: int) -> list[float]:
        """获取指定层级的块段尺寸 [size_x, size_y, size_z]

        以 min_size（= max_level 时的尺寸）为基准反推：
            size_at_level(L) = min_size * 2^(max_level - L)

        Args:
            level: 层级（0 ~ max_level），0 为最粗粒度、max_level 为最细粒度

        Returns:
            [size_x, size_y, size_z]

        Raises:
            ValueError: level 超出 [0, max_level] 范围
        """
        if level < 0 or level > self.max_level:
            raise ValueError(
                f"level 必须在 [0, {self.max_level}] 范围内，实际: {level}"
            )
        factor = 2 ** (self.max_level - level)
        return [s * factor for s in self.min_size]

    def level_size_map(self) -> dict[int, list[float]]:
        """获取 0 ~ max_level 所有层级的块段尺寸对照表

        Returns:
            {level: [size_x, size_y, size_z], ...}
        """
        return {lv: self.get_level_size(lv) for lv in range(self.max_level + 1)}

    @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: Optional[str]=None) -> Iterator[dict]:
        """遍历所有块段，逐块返回指定字段数据"""
        self._cpp_obj.InitCursor()
        while 0 == self._cpp_obj.AdvCursor():    # 返回0 表示有数据，其他值表示没有数据了
            if self._cpp_obj.IsBeyondBoundaryNode():
                continue
            success, data = self._cpp_obj.GetCursorData(field_name or "")
            if success:
                try:
                    record = json.loads(data)
                    # 过滤：若所有 "距离" 开头的字段值均为 0，跳过该记录
                    if all(v == 0 for k, v in record.items() if k.startswith("距离")):
                        continue
                    # 去掉不需要的字段
                    _DROP_KEYS = {"xLen", "yLen", "zen", "I", "J", "K", "编号"}
                    record = {k: v for k, v in record.items() if k not in _DROP_KEYS}
                    yield record
                except (ValueError, TypeError):
                    continue
    def close(self) -> None:
        """释放 C++ 对象，解除文件占用"""
        self._cpp_obj = None
        self._file_path = None

field_definitions property

字段定义列表

max_level property

最大精度层级

min_size property

最小块段尺寸（对应 max_level 层级的块段尺寸）

origin property

模型原点

rotation property

模型旋转角度

total_block_count property

块段总数量

xyz_length property

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

close()

释放 C++ 对象，解除文件占用

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

def close(self) -> None:
    """释放 C++ 对象，解除文件占用"""
    self._cpp_obj = None
    self._file_path = None

get_level_size(level)

获取指定层级的块段尺寸 [size_x, size_y, size_z]

以 min_size（= max_level 时的尺寸）为基准反推： size_at_level(L) = min_size * 2^(max_level - L)

Parameters:

Name	Type	Description	Default
level	int	层级（0 ~ max_level），0 为最粗粒度、max_level 为最细粒度	required

Returns:

Type	Description
list[float]	[size_x, size_y, size_z]

Raises:

Type	Description
ValueError	level 超出 [0, max_level] 范围

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

def get_level_size(self, level: int) -> list[float]:
    """获取指定层级的块段尺寸 [size_x, size_y, size_z]

    以 min_size（= max_level 时的尺寸）为基准反推：
        size_at_level(L) = min_size * 2^(max_level - L)

    Args:
        level: 层级（0 ~ max_level），0 为最粗粒度、max_level 为最细粒度

    Returns:
        [size_x, size_y, size_z]

    Raises:
        ValueError: level 超出 [0, max_level] 范围
    """
    if level < 0 or level > self.max_level:
        raise ValueError(
            f"level 必须在 [0, {self.max_level}] 范围内，实际: {level}"
        )
    factor = 2 ** (self.max_level - level)
    return [s * factor for s in self.min_size]

iter_blocks(field_name=None)

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

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

def iter_blocks(self, field_name: Optional[str]=None) -> Iterator[dict]:
    """遍历所有块段，逐块返回指定字段数据"""
    self._cpp_obj.InitCursor()
    while 0 == self._cpp_obj.AdvCursor():    # 返回0 表示有数据，其他值表示没有数据了
        if self._cpp_obj.IsBeyondBoundaryNode():
            continue
        success, data = self._cpp_obj.GetCursorData(field_name or "")
        if success:
            try:
                record = json.loads(data)
                # 过滤：若所有 "距离" 开头的字段值均为 0，跳过该记录
                if all(v == 0 for k, v in record.items() if k.startswith("距离")):
                    continue
                # 去掉不需要的字段
                _DROP_KEYS = {"xLen", "yLen", "zen", "I", "J", "K", "编号"}
                record = {k: v for k, v in record.items() if k not in _DROP_KEYS}
                yield record
            except (ValueError, TypeError):
                continue

level_size_map()

获取 0 ~ max_level 所有层级的块段尺寸对照表

Returns:

Type	Description
dict[int, list[float]]	{level: [size_x, size_y, size_z], ...}

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

def level_size_map(self) -> dict[int, list[float]]:
    """获取 0 ~ max_level 所有层级的块段尺寸对照表

    Returns:
        {level: [size_x, size_y, size_z], ...}
    """
    return {lv: self.get_level_size(lv) for lv in range(self.max_level + 1)}

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, file_path: str | None = None):
        self._cpp_obj = cpp_obj
        self._ref_dict = ref_dict
        self._file_path: str | None = file_path
        self._extend_table_metas: list[NativeExtendTableMeta] = self._parse_extend_table_metas(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, file_path)

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

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

    @property
    def file_path(self) -> str | None:
        """DMD 文件路径"""
        return self._file_path

    # ------------------------------------------------------------------
    # 扩展表（附加表）元数据解析
    # ------------------------------------------------------------------

    @staticmethod
    def _parse_extend_table_metas(ref_dict: dict) -> list[NativeExtendTableMeta]:
        """从 ref_dict 解析扩展表元数据。

        ExtendTable{i}FieldList 格式：
        ``<钻孔编号列名>,<起始深度列名>,<结束深度列名>,<标题名称>``

        Returns:
            list of NativeExtendTableMeta
        """
        if not ref_dict:
            return []

        try:
            count = int(ref_dict.get("ExtendTableCount", 0))
        except (ValueError, TypeError):
            return []

        if count <= 0:
            return []

        metas = []
        for i in range(1, count + 1):
            path_key = f"ExtendTable{i}Path"
            fields_key = f"ExtendTable{i}FieldList"
            if path_key not in ref_dict:
                continue

            dmt_filename = ref_dict[path_key]
            field_list = ref_dict.get(fields_key, "")

            # 解析 4 个元数据字段
            field_names = [f.strip() for f in field_list.split(",") if f.strip()]
            hole_id_column = field_names[0] if len(field_names) >= 1 else None
            from_column = field_names[1] if len(field_names) >= 2 else None
            to_column = field_names[2] if len(field_names) >= 3 else None
            display_title = field_names[3] if len(field_names) >= 4 else None

            # table_name 优先使用 display_title，回退到 dmt 文件名
            table_name = display_title if display_title else dmt_filename.replace(".dmt", "")

            metas.append(NativeExtendTableMeta(
                dmt_filename=dmt_filename,
                field_list=field_list,
                hole_id_column=hole_id_column,
                from_column=from_column,
                to_column=to_column,
                display_title=display_title,
                table_name=table_name,
            ))

        return metas

    @property
    def extend_table_count(self) -> int:
        """扩展表数量"""
        return len(self._extend_table_metas)

    def extend_table_metas(self) -> list[NativeExtendTableMeta]:
        """返回扩展表元数据列表（不含 DataFrame）。

        Returns:
            list of NativeExtendTableMeta
        """
        return list(self._extend_table_metas)

    def get_extend_table(self, index: int) -> "pd.DataFrame":
        """按索引加载扩展表数据。

        从 DMT 文件加载数据，保留原始列名不做重命名。

        Args:
            index: 扩展表索引（0-based）

        Returns:
            pd.DataFrame，文件不存在或无元数据时返回空 DataFrame
        """
        require_pandas()
        import pandas as pd

        if index < 0 or index >= len(self._extend_table_metas):
            return pd.DataFrame()

        meta = self._extend_table_metas[index]
        dmt_filename = meta.dmt_filename

        if not self._file_path:
            return pd.DataFrame()

        dmt_path = Path(self._file_path).parent / dmt_filename
        if not dmt_path.exists():
            return pd.DataFrame()

        try:
            handle = NativeDataTableHandle.create()
            handle.load(str(dmt_path))
            return handle.to_dataframe()
        except Exception:
            logger.warning("加载扩展表失败: %s", dmt_path)
            return pd.DataFrame()

    def extend_tables(self) -> list[NativeExtendTable]:
        """加载全部扩展表，返回含数据和元数据的完整列表。

        DMT 文件路径根据 dmd 文件所在目录 + dmt_filename 拼接。

        Returns:
            list of NativeExtendTable
        """
        result = []

        for i in range(len(self._extend_table_metas)):
            df = self.get_extend_table(i)
            meta = self._extend_table_metas[i]
            result.append(NativeExtendTable(
                df=df,
                dmt_filename=meta.dmt_filename,
                field_list=meta.field_list,
                table_name=meta.table_name,
                hole_id_column=meta.hole_id_column,
                from_column=meta.from_column,
                to_column=meta.to_column,
                display_title=meta.display_title,
            ))

        return result

    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 格式

        先通过 GetDisplayDB 获取 dmDbDatabase 对象，保存为临时 .dmf 文件，
        再将 .dmf 拷贝并重命名为 .dmg。
        """
        """保存为 .dmg 格式"""
        try:
            self._cpp_obj.Save(file_path)
            import time
            time.sleep(2)
        except Exception as exc:
            raise NativeProspectingError(f"保存 .dmg 失败: {file_path}") from exc

    def close(self) -> None:
        """释放 C++ 对象，解除文件占用"""
        self._cpp_obj = None
        self._ref_dict = {}
        self._file_path = None
        self._extend_table_metas = []

extend_table_count property

扩展表数量

file_path property

DMD 文件路径

close()

释放 C++ 对象，解除文件占用

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

def close(self) -> None:
    """释放 C++ 对象，解除文件占用"""
    self._cpp_obj = None
    self._ref_dict = {}
    self._file_path = None
    self._extend_table_metas = []

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(), {})

extend_table_metas()

返回扩展表元数据列表（不含 DataFrame）。

Returns:

Type	Description
list[NativeExtendTableMeta]	list of NativeExtendTableMeta

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

def extend_table_metas(self) -> list[NativeExtendTableMeta]:
    """返回扩展表元数据列表（不含 DataFrame）。

    Returns:
        list of NativeExtendTableMeta
    """
    return list(self._extend_table_metas)

extend_tables()

加载全部扩展表，返回含数据和元数据的完整列表。

DMT 文件路径根据 dmd 文件所在目录 + dmt_filename 拼接。

Returns:

Type	Description
list[NativeExtendTable]	list of NativeExtendTable

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

def extend_tables(self) -> list[NativeExtendTable]:
    """加载全部扩展表，返回含数据和元数据的完整列表。

    DMT 文件路径根据 dmd 文件所在目录 + dmt_filename 拼接。

    Returns:
        list of NativeExtendTable
    """
    result = []

    for i in range(len(self._extend_table_metas)):
        df = self.get_extend_table(i)
        meta = self._extend_table_metas[i]
        result.append(NativeExtendTable(
            df=df,
            dmt_filename=meta.dmt_filename,
            field_list=meta.field_list,
            table_name=meta.table_name,
            hole_id_column=meta.hole_id_column,
            from_column=meta.from_column,
            to_column=meta.to_column,
            display_title=meta.display_title,
        ))

    return result

get_extend_table(index)

按索引加载扩展表数据。

从 DMT 文件加载数据，保留原始列名不做重命名。

Parameters:

Name	Type	Description	Default
index	int	扩展表索引（0-based）	required

Returns:

Type	Description
'pd.DataFrame'	pd.DataFrame，文件不存在或无元数据时返回空 DataFrame

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

def get_extend_table(self, index: int) -> "pd.DataFrame":
    """按索引加载扩展表数据。

    从 DMT 文件加载数据，保留原始列名不做重命名。

    Args:
        index: 扩展表索引（0-based）

    Returns:
        pd.DataFrame，文件不存在或无元数据时返回空 DataFrame
    """
    require_pandas()
    import pandas as pd

    if index < 0 or index >= len(self._extend_table_metas):
        return pd.DataFrame()

    meta = self._extend_table_metas[index]
    dmt_filename = meta.dmt_filename

    if not self._file_path:
        return pd.DataFrame()

    dmt_path = Path(self._file_path).parent / dmt_filename
    if not dmt_path.exists():
        return pd.DataFrame()

    try:
        handle = NativeDataTableHandle.create()
        handle.load(str(dmt_path))
        return handle.to_dataframe()
    except Exception:
        logger.warning("加载扩展表失败: %s", dmt_path)
        return pd.DataFrame()

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, file_path)

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 格式

先通过 GetDisplayDB 获取 dmDbDatabase 对象，保存为临时 .dmf 文件， 再将 .dmf 拷贝并重命名为 .dmg。

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

def save_as_dmg(self, file_path: str) -> None:
    """保存为 .dmg 格式

    先通过 GetDisplayDB 获取 dmDbDatabase 对象，保存为临时 .dmf 文件，
    再将 .dmf 拷贝并重命名为 .dmg。
    """
    """保存为 .dmg 格式"""
    try:
        self._cpp_obj.Save(file_path)
        import time
        time.sleep(2)
    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:
    """储量计算"""
    constraint_params = _serialize_constraints(constraint_json)
    reserve_params = _serialize_param(reserve_json)
    logger.debug(
        "Dm.ReservesCalculate 入参:\n"
        "  block_model_file: %s\n"
        "  constraints: %s\n"
        "  reserves: %s",
        block_model_file,
        constraint_params,
        reserve_params,
    )
    result = Dm.ReservesCalculate(
        block_model_file,
        constraint_params,
        reserve_params,
    )
    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:
    """距离幂估值"""
    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:
    """特高品位处理"""
    print(f"extra_high_grade_process: {_serialize_param(param)}")
    raw = Dm.ExtraHighGradeProcess(_serialize_param(param))
    try:
        return parse_json_response(raw, NativeProspectingError, "特高品位处理")
    except NativeProcessError as exc:
        if "所选字段数据为空" in str(exc.response.get("message")):
            return {"state": "true", "message": "所选字段数据为空"}
        raise exc

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, "样长组合")

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

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

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

Parameters

extend_tables : list[dict], optional 扩展表列表，每个元素为 dict： - df: pd.DataFrame 扩展表数据 - dmt_filename: str .dmt 文件名（如 "分层表.dmt"） - field_list: str 字段元数据（"孔号列,从列,至列,标题"），可为空 - hole_id_column: str|null 钻孔编号对应列名（field_list 为空时用于自动构造） - from_column: str|null 起始深度对应列名 - to_column: str|null 结束深度对应列名 - display_title: str|null 标题名称（Dimine UI 显示用） 用于隐式地层建模等场景，最终写入 ExtendTableCount / ExtendTable{N}Path 等引用字段。

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

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

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

    Parameters
    ----------
    extend_tables : list[dict], optional
        扩展表列表，每个元素为 dict：
        - df: pd.DataFrame           扩展表数据
        - dmt_filename: str          .dmt 文件名（如 "分层表.dmt"）
        - field_list: str            字段元数据（"孔号列,从列,至列,标题"），可为空
        - hole_id_column: str|null   钻孔编号对应列名（field_list 为空时用于自动构造）
        - from_column: str|null      起始深度对应列名
        - to_column: str|null        结束深度对应列名
        - display_title: str|null    标题名称（Dimine UI 显示用）
        用于隐式地层建模等场景，最终写入 ExtendTableCount / ExtendTable{N}Path 等引用字段。
    """
    try:
        import pandas as pd
    except ImportError as exc:
        raise ImportError("使用 save_drill_database() 需要安装 pandas: uv add pandas") from exc

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

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

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

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

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

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

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

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

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

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

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

    if extend_tables:
        ext_count = len(extend_tables)
        data["ExtendTableCount"] = str(ext_count)
        for i, ext in enumerate(extend_tables, start=1):
            if isinstance(ext, NativeExtendTable):
                ext_df = ext.df
                ext_filename = ext.dmt_filename
                ext_fields = ext.field_list

                # 自动从元数据构造 field_list（field_list 为空但有元数据时）
                if not ext_fields:
                    hole_id = ext.hole_id_column or ""
                    from_col = ext.from_column or ""
                    to_col = ext.to_column or ""
                    display = ext.display_title or ext.table_name or ""
                    ext_fields = f"{hole_id},{from_col},{to_col},{display}"
            else:
                ext_df = ext["df"]
                ext_filename = ext["dmt_filename"]
                ext_fields = ext.get("field_list", "")

                # 自动从元数据构造 field_list（field_list 为空但有元数据时）
                if not ext_fields:
                    hole_id = ext.get("hole_id_column", "")
                    from_col = ext.get("from_column", "")
                    to_col = ext.get("to_column", "")
                    display = ext.get("display_title", ext.get("table_name", ""))
                    ext_fields = f"{hole_id},{from_col},{to_col},{display}"

            ext_dmt_path = directory / ext_filename
            NativeDataTableHandle.from_dataframe(ext_df).save(str(ext_dmt_path))
            data[f"ExtendTable{i}Path"] = ext_filename
            data[f"ExtendTable{i}FieldList"] = ext_fields
            table_count += 1

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

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

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

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, "台阶组合")