drill_db

钻孔数据库业务层：DrillDBManager、drill_conn 与相关功能

职责：提供 Pythonic 的钻孔数据库 API；直接调用 native.prospecting， 不依赖 _adapter.py。 允许类型：pd.DataFrame、np.ndarray、models.、dict、str、int、float、bool 禁止类型：任何 Dm 类型

DrillDBError

Bases: RuntimeError

钻孔数据库操作异常基类

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

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

DrillDBLoadError

Bases: DrillDBError

加载钻孔数据库失败

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

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

DrillDBManager

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

collar property writable

钻孔孔口表（pd.DataFrame）

lithology property writable

钻孔岩性表（pd.DataFrame）

sample property writable

钻孔样品表（pd.DataFrame）

survey property writable

钻孔测斜表（pd.DataFrame）

__getitem__(table_name)

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

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

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

__init__(file_path=None, base_path=None)

初始化钻孔数据库管理器

Parameters:

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

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

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

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

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

close()

释放资源

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

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

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

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

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

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

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

load(file_path, base_path=None)

从文件加载钻孔数据库

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

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

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

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

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

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

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

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

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

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

process_high_grade(params)

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

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

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

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

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

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

Parameters:

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

Returns:

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

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

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

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

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

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

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

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

process_samples(params)

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

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

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

process_steps(params)

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

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

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

reload()

重新加载文件

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

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

save(output_path=None, extend_tables=None)

保存钻孔数据库

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

Parameters

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

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

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

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

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

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

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

save_as_dmg(output_path)

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

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

Parameters:

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

Raises:

Type	Description
DrillDBSaveError	保存失败时抛出

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

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

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

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

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

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

DrillDBProcessError

Bases: DrillDBError

钻孔数据处理失败

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

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

DrillDBSaveError

Bases: DrillDBError

保存钻孔数据库失败

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

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

DrillFunctionWrapper

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

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

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

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

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

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

extra_high_grade_process(params) staticmethod

特高品位处理

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

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

sample_length_combine(params) staticmethod

样长组合

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

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

step_combine(params) staticmethod

台阶组合

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

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

convert_dmd_to_dmg(input_path, output_path)

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

Parameters:

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

Raises:

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

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

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

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

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

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

drill_conn(file_path=None, base_path=None)

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

Parameters:

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

Usage

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

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

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

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

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

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

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

extra_high_grade_process(param)

特高品位处理

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

def extra_high_grade_process(param: Union[dict, str]) -> dict:
    """特高品位处理"""
    try:
        return _native_extra_high_grade_process(param)
    except NativeProcessError as exc:
        raise DrillDBProcessError(str(exc), exc.response) from exc

sample_length_combine(param)

样长组合

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

def sample_length_combine(param: Union[dict, str]) -> dict:
    """样长组合"""
    try:
        return _native_sample_length_combine(param)
    except NativeProcessError as exc:
        raise DrillDBProcessError(str(exc), exc.response) from exc

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

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

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

Parameters

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

step_combine(param)

台阶组合

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

def step_combine(param: Union[dict, str]) -> dict:
    """台阶组合"""
    try:
        return _native_step_combine(param)
    except NativeProcessError as exc:
        raise DrillDBProcessError(str(exc), exc.response) from exc