drill_db

钻孔数据库连接管理器（已弃用兼容 shim）

请迁移至： from dimine_python_sdk.lib.prospecting.api import DrillDBManager, drill_conn from dimine_python_sdk.lib.prospecting._adapter import DrillDBError, DrillDBLoadError, DrillDBProcessError

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\api.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._geo_drill = 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._geo_drill = create_geo_drill()

    # ------------------------------------------------------------------
    # 内部缓存管理
    # ------------------------------------------------------------------
    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

        self._geo_drill, self._ref_dict = load_geo_drill(self.file_path)
        self._invalidate_cache()

    def reload(self):
        """重新加载文件"""
        if not self.file_path:
            raise DrillDBError("未指定文件路径，无法重新加载")
        self._geo_drill, self._ref_dict = load_geo_drill(self.file_path)
        self._invalidate_cache()

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

        完全走 Python 层，不调用 C++ geo_drill_save：
            1. 孔口表   -> {stem}_collar.dmt（DmDataTable.from_dataframe）
            2. 斜测表   -> {stem}_survey.dmt
            3. 岩性表   -> {stem}_lithology.dmt
            4. 样品表   -> {stem}_sample.dmt（可选）
            5. 引用关系 -> {stem}.dmd（DmDataTable.from_dataframe）
        """
        save_path = output_path or self.file_path
        if not save_path:
            raise DrillDBError("未指定保存路径，请传入 output_path 或在初始化/加载时设置 file_path")

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

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

    # ------------------------------------------------------------------
    # 表访问（DataFrame 化）
    # ------------------------------------------------------------------
    @property
    def collar(self) -> "pd.DataFrame":
        """钻孔孔口表（pd.DataFrame）"""
        if self._collar_df is None:
            self._collar_df = get_collar_table(self._geo_drill, self._ref_dict)
        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 = get_survey_table(self._geo_drill, self._ref_dict)
        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 = get_lithology_table(self._geo_drill, self._ref_dict)
        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 = get_sample_table(self._geo_drill, self._ref_dict)
        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:
        """处理样长组合"""
        return DrillFunctionWrapper.sample_length_combine(params)

    def process_steps(self, params: StepCombineParam) -> dict:
        """处理台阶组合"""
        return DrillFunctionWrapper.step_combine(params)

    def process_high_grade(self, params: HighGradeProcessParam) -> dict:
        """处理特高品位"""
        return DrillFunctionWrapper.extra_high_grade_process(params)

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\api.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\api.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._geo_drill = 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._geo_drill = create_geo_drill()

close()

释放资源

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

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

load(file_path, base_path=None)

从文件加载钻孔数据库

Source code in dimine_python_sdk\lib\prospecting\api.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

    self._geo_drill, self._ref_dict = load_geo_drill(self.file_path)
    self._invalidate_cache()

process_high_grade(params)

处理特高品位

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

def process_high_grade(self, params: HighGradeProcessParam) -> dict:
    """处理特高品位"""
    return DrillFunctionWrapper.extra_high_grade_process(params)

process_samples(params)

处理样长组合

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

def process_samples(self, params: SampleLengthCombineParam) -> dict:
    """处理样长组合"""
    return DrillFunctionWrapper.sample_length_combine(params)

process_steps(params)

处理台阶组合

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

def process_steps(self, params: StepCombineParam) -> dict:
    """处理台阶组合"""
    return DrillFunctionWrapper.step_combine(params)

reload()

重新加载文件

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

def reload(self):
    """重新加载文件"""
    if not self.file_path:
        raise DrillDBError("未指定文件路径，无法重新加载")
    self._geo_drill, self._ref_dict = load_geo_drill(self.file_path)
    self._invalidate_cache()

save(output_path=None)

保存钻孔数据库

完全走 Python 层，不调用 C++ geo_drill_save： 1. 孔口表 -> {stem}_collar.dmt（DmDataTable.from_dataframe） 2. 斜测表 -> {stem}_survey.dmt 3. 岩性表 -> {stem}_lithology.dmt 4. 样品表 -> {stem}_sample.dmt（可选） 5. 引用关系 -> {stem}.dmd（DmDataTable.from_dataframe）

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

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

    完全走 Python 层，不调用 C++ geo_drill_save：
        1. 孔口表   -> {stem}_collar.dmt（DmDataTable.from_dataframe）
        2. 斜测表   -> {stem}_survey.dmt
        3. 岩性表   -> {stem}_lithology.dmt
        4. 样品表   -> {stem}_sample.dmt（可选）
        5. 引用关系 -> {stem}.dmd（DmDataTable.from_dataframe）
    """
    save_path = output_path or self.file_path
    if not save_path:
        raise DrillDBError("未指定保存路径，请传入 output_path 或在初始化/加载时设置 file_path")

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

DrillFunctionWrapper

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

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

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

    @staticmethod
    def sample_length_combine(params: SampleLengthCombineParam) -> dict:
        """样长组合"""
        return sample_length_combine(params.model_dump_json())

    @staticmethod
    def step_combine(params: StepCombineParam) -> dict:
        """台阶组合"""
        return step_combine(params.model_dump_json())

    @staticmethod
    def extra_high_grade_process(params: HighGradeProcessParam) -> dict:
        """特高品位处理"""
        return extra_high_grade_process(params.model_dump_json())

extra_high_grade_process(params) staticmethod

特高品位处理

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

@staticmethod
def extra_high_grade_process(params: HighGradeProcessParam) -> dict:
    """特高品位处理"""
    return extra_high_grade_process(params.model_dump_json())

sample_length_combine(params) staticmethod

样长组合

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

@staticmethod
def sample_length_combine(params: SampleLengthCombineParam) -> dict:
    """样长组合"""
    return sample_length_combine(params.model_dump_json())

step_combine(params) staticmethod

台阶组合

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

@staticmethod
def step_combine(params: StepCombineParam) -> dict:
    """台阶组合"""
    return step_combine(params.model_dump_json())

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\api.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()