dmg_file

DMG 文件业务层：DmgFile、dmg_conn 与相关功能

职责： - 提供 Pythonic 的 DMG 钻孔数据库三维可视化文件接口； - 从分段 DataFrame 构建 DMG（save_from_segments）。

允许类型：models.、np.ndarray、pd.DataFrame、Path、str、标量禁止类型：Dm 禁止 import DmPyBindInterface

`DmgError`

Bases: RuntimeError

DMG 操作异常基类

Source code in dimine_python_sdk\lib\io\dmg_file.py

class DmgError(RuntimeError):
    """DMG 操作异常基类"""

    pass

`DmgFile`

Bases: DmfFile

DMG 文件管理器，继承自 DmfFile。

DMG 是钻孔数据库的三维可视化格式（一种特殊的 DMF 文件）。当前支持： - 保存：从 DmdFile 保存为 DMG（直接调用 save_as_dmg） - 保存：从分段 DataFrame 保存为 DMG（save_from_segments） - 加载：加载 DMG 文件，通过继承的 DMF API 查看图层和实体结构 - 提取：从实体几何中提取孔口坐标等结构信息（属性数据受限于 C++ API 暂不可读）

用法 - 从 DmdFile 保存 DMG::

from dimine_python_sdk.lib.io import DmdFile
dmd = DmdFile("钻孔数据.dmd")
dmg = DmgFile()
dmg.save_from_drill_db("output.dmg", dmd)

用法 - 从分段 DataFrame 保存 DMG::

dmg = DmgFile()
dmg.save_from_segments("output.dmg", df)

用法 - 加载 DMG 查看结构::

dmg = DmgFile("input.dmg")
print(dmg.info)                 # {file, layers, total_entities, ...}
print(dmg.layer_names)          # ['L0', 'L1', 'L2']
for layer in dmg:               # 遍历图层和实体
    for entity in layer.entities:
        print(type(entity).__name__, entity.geometry)

继承自 DmfFile，可使用所有图层/实体操作方法。

Source code in dimine_python_sdk\lib\io\dmg_file.py

class DmgFile(DmfFile):
    """
    DMG 文件管理器，继承自 ``DmfFile``。

    DMG 是钻孔数据库的三维可视化格式（一种特殊的 DMF 文件）。
    当前支持：
    - **保存**：从 ``DmdFile`` 保存为 DMG（直接调用 ``save_as_dmg``）
    - **保存**：从分段 DataFrame 保存为 DMG（``save_from_segments``）
    - **加载**：加载 DMG 文件，通过继承的 DMF API 查看图层和实体结构
    - **提取**：从实体几何中提取孔口坐标等结构信息（属性数据受限于 C++ API 暂不可读）

    用法 - 从 DmdFile 保存 DMG::

        from dimine_python_sdk.lib.io import DmdFile
        dmd = DmdFile("钻孔数据.dmd")
        dmg = DmgFile()
        dmg.save_from_drill_db("output.dmg", dmd)

    用法 - 从分段 DataFrame 保存 DMG::

        dmg = DmgFile()
        dmg.save_from_segments("output.dmg", df)

    用法 - 加载 DMG 查看结构::

        dmg = DmgFile("input.dmg")
        print(dmg.info)                 # {file, layers, total_entities, ...}
        print(dmg.layer_names)          # ['L0', 'L1', 'L2']
        for layer in dmg:               # 遍历图层和实体
            for entity in layer.entities:
                print(type(entity).__name__, entity.geometry)

    继承自 ``DmfFile``，可使用所有图层/实体操作方法。
    """

    # ------------------------------------------------------------------
    # 生命周期
    # ------------------------------------------------------------------

    def load(self, file_path: str) -> "DmgFile":
        """
        加载 .dmg 文件（继承 DmfFile.load）

        Returns:
            self（支持链式调用）

        Raises:
            DmfLoadError: 文件不存在或加载失败
        """
        super().load(file_path)
        return self

    def save_from_drill_db(
        self,
        file_path: str,
        drill_db: Any,
    ) -> None:
        """
        从钻孔数据库对象保存为 DMG 文件

        直接通过 save_as_dmg 保存为 DMG 格式。
        接受 ``DmdFile`` 实例。

        Args:
            file_path: 目标 .dmg 文件路径
            drill_db: DmdFile 实例

        Usage::

            from dimine_python_sdk.lib.io import DmdFile
            dmd = DmdFile("钻孔数据.dmd")
            DmgFile().save_from_drill_db("output.dmg", dmd)
        """
        if not isinstance(drill_db, DmdFile):
            raise DmgError("drill_db 必须是 DmdFile 实例")

        try:
            drill_db.save_as_dmg(file_path)
        except Exception as exc:
            raise DmgError(f"保存 .dmg 失败: {file_path}") from exc

        self.file_path = str(file_path)

    def save_from_segments(
        self,
        file_path: str,
        df: pd.DataFrame,
    ) -> None:
        """
        从分段 DataFrame 构建并保存为 DMG 文件

        输入 DataFrame 每行代表某个钻孔下某一段的完整信息，包含起点坐标、
        终点坐标、样品信息及元素品位等。

        必需列：
            BHID、start_x、start_y、start_z、end_x、end_y、end_z、
            勘探线、工程类型、ROCK-TYPE、SAMPLE-ID、SAMFROM、SAMTO、样长。

        可选列：
            任意大写元素品位列（如 CU、TFE 等），类型为浮点型。

        Args:
            file_path: 目标 .dmg 文件路径
            df: 钻孔分段 DataFrame

        Usage::

            DmgFile().save_from_segments(
                "output.dmg",
                df,
            )
        """
        try:
            handle = NativeDatabaseHandle.create_local()
            _build_dmg_from_segments(handle, df)
            handle.save(str(file_path))
        except Exception as exc:
            raise DmgError(f"保存 .dmg 失败: {file_path}") from exc

        self.file_path = str(file_path)

    def to_segments_dataframe(self) -> pd.DataFrame:
        """
        从已加载的 DMG 文件中提取分段 DataFrame

        读取"钻孔轨迹"图层中的线段实体，提取起点/终点坐标及属性，
        重构为与 ``save_from_segments`` 输入一致的 DataFrame。

        Returns:
            分段 DataFrame，列包含：
            BHID、start_x、start_y、start_z、end_x、end_y、end_z、
            勘探线、工程类型、ROCK-TYPE、SAMPLE-ID、SAMFROM、SAMTO、
            样长，以及检测到的大写元素品位列。

        Raises:
            DmgError: 文件未加载，或"钻孔轨迹"图层不存在/为空

        Usage::

            dmg = DmgFile("input.dmg")
            df = dmg.to_segments_dataframe()
            print(df.head())
        """
        if not self.file_path:
            raise DmgError("请先加载 DMG 文件")

        # 查找"钻孔轨迹"图层
        traces_layer = None
        for layer in self.layers:
            if layer.name == _DMG_LAYER_SAMPLES:
                traces_layer = layer
                break

        if traces_layer is None:
            raise DmgError(f"未找到'{_DMG_LAYER_SAMPLES}'图层")
        if not traces_layer.entities:
            raise DmgError(f"'{_DMG_LAYER_SAMPLES}'图层中没有实体")

        rows: list[dict[str, Any]] = []
        all_prop_names: set[str] = set()

        for entity in traces_layer.entities:
            # 只处理线段实体
            if not isinstance(entity, Line):
                continue

            geom = entity.geometry
            if geom is None or geom.shape[0] < 2:
                continue

            # 提取属性
            props: dict[str, Any] = {}
            for p in entity.properties:
                props[p.name] = p.value
                all_prop_names.add(p.name)

            row: dict[str, Any] = {
                "BHID": str(props.get("BHID", "")),
                "start_x": float(geom[0][0]),
                "start_y": float(geom[0][1]),
                "start_z": float(geom[0][2]),
                "end_x": float(geom[1][0]),
                "end_y": float(geom[1][1]),
                "end_z": float(geom[1][2]),
                "勘探线": str(props.get("勘探线号", "")),
                "工程类型": str(props.get("工程类型", "")),
                "ROCK-TYPE": str(props.get("ROCK-TYPE", "")),
                "SAMPLE-ID": props.get("SAMPLE-ID"),
                "SAMFROM": props.get("SAMFROM"),
                "SAMTO": props.get("SAMTO"),
                "样长": props.get("样长"),
            }

            # 提取元素品位列（大写字母列，不在固定列中）
            element_cols = _detect_element_columns(
                pd.DataFrame([props])
            )
            for col in element_cols:
                row[col] = props.get(col)

            rows.append(row)

        if not rows:
            raise DmgError(f"'{_DMG_LAYER_SAMPLES}'图层中没有有效的线段实体")

        return pd.DataFrame(rows)

    def save_from_drill_db_dataframe(
        self,
        file_path: str,
        drill_db: Any,
    ) -> None:
        """
        从钻孔数据库对象构建分段 DataFrame 并保存为 DMG 文件

        将 ``DmdFile`` 中的孔口表、测斜表、样品表、
        岩性表合并，通过测斜数据计算每个样品段的三维起点/终点坐标，再调用
        ``save_from_segments`` 保存为 DMG。

        Args:
            file_path: 目标 .dmg 文件路径
            drill_db: ``DmdFile`` 实例

        Usage::

            from dimine_python_sdk.lib.io import DmdFile
            dmd = DmdFile("钻孔数据.dmd")
            DmgFile().save_from_drill_db_dataframe("output.dmg", dmd)
        """
        if not isinstance(drill_db, DmdFile):
            raise DmgError("drill_db 必须是 DmdFile 实例")

        try:
            segment_df = _drill_db_to_segment_df(drill_db)
            self.save_from_segments(file_path, segment_df)
        except Exception as exc:
            raise DmgError(
                f"从钻孔数据库构建 DMG 分段失败: {file_path}"
            ) from exc

    # ------------------------------------------------------------------
    # 属性
    # ------------------------------------------------------------------

    @property
    def info(self) -> Dict[str, Any]:
        """DMG 文件摘要信息"""
        base = super().info
        # 统计各类实体数量
        point_count = 0
        line_count = 0
        for layer in self.layers:
            for entity in layer.entities:
                if isinstance(entity, Point):
                    point_count += 1
                elif isinstance(entity, Line):
                    line_count += 1
        base["point_entities"] = point_count
        base["line_entities"] = line_count
        return base

    def __repr__(self) -> str:
        return (
            f"DmgFile('{self.file_path or ''}', "
            f"layers={self.layer_count}, "
            f"entities={sum(layer.entity_count for layer in self.layers)})"
        )

`info` `property`

DMG 文件摘要信息

`load(file_path)`

加载 .dmg 文件（继承 DmfFile.load）

Returns:

Type	Description
`'DmgFile'`	self（支持链式调用）

Raises:

Type	Description
`DmfLoadError`	文件不存在或加载失败

Source code in dimine_python_sdk\lib\io\dmg_file.py

def load(self, file_path: str) -> "DmgFile":
    """
    加载 .dmg 文件（继承 DmfFile.load）

    Returns:
        self（支持链式调用）

    Raises:
        DmfLoadError: 文件不存在或加载失败
    """
    super().load(file_path)
    return self

`save_from_drill_db(file_path, drill_db)`

从钻孔数据库对象保存为 DMG 文件

直接通过 save_as_dmg 保存为 DMG 格式。接受 DmdFile 实例。

Parameters:

Name	Type	Description	Default
`file_path`	`str`	目标 .dmg 文件路径	required
`drill_db`	`Any`	DmdFile 实例	required

Usage::

from dimine_python_sdk.lib.io import DmdFile
dmd = DmdFile("钻孔数据.dmd")
DmgFile().save_from_drill_db("output.dmg", dmd)

Source code in dimine_python_sdk\lib\io\dmg_file.py

def save_from_drill_db(
    self,
    file_path: str,
    drill_db: Any,
) -> None:
    """
    从钻孔数据库对象保存为 DMG 文件

    直接通过 save_as_dmg 保存为 DMG 格式。
    接受 ``DmdFile`` 实例。

    Args:
        file_path: 目标 .dmg 文件路径
        drill_db: DmdFile 实例

    Usage::

        from dimine_python_sdk.lib.io import DmdFile
        dmd = DmdFile("钻孔数据.dmd")
        DmgFile().save_from_drill_db("output.dmg", dmd)
    """
    if not isinstance(drill_db, DmdFile):
        raise DmgError("drill_db 必须是 DmdFile 实例")

    try:
        drill_db.save_as_dmg(file_path)
    except Exception as exc:
        raise DmgError(f"保存 .dmg 失败: {file_path}") from exc

    self.file_path = str(file_path)

`save_from_drill_db_dataframe(file_path, drill_db)`

从钻孔数据库对象构建分段 DataFrame 并保存为 DMG 文件

将 DmdFile 中的孔口表、测斜表、样品表、岩性表合并，通过测斜数据计算每个样品段的三维起点/终点坐标，再调用 save_from_segments 保存为 DMG。

Parameters:

Name	Type	Description	Default
`file_path`	`str`	目标 .dmg 文件路径	required
`drill_db`	`Any`	`DmdFile` 实例	required

Usage::

from dimine_python_sdk.lib.io import DmdFile
dmd = DmdFile("钻孔数据.dmd")
DmgFile().save_from_drill_db_dataframe("output.dmg", dmd)

Source code in dimine_python_sdk\lib\io\dmg_file.py

def save_from_drill_db_dataframe(
    self,
    file_path: str,
    drill_db: Any,
) -> None:
    """
    从钻孔数据库对象构建分段 DataFrame 并保存为 DMG 文件

    将 ``DmdFile`` 中的孔口表、测斜表、样品表、
    岩性表合并，通过测斜数据计算每个样品段的三维起点/终点坐标，再调用
    ``save_from_segments`` 保存为 DMG。

    Args:
        file_path: 目标 .dmg 文件路径
        drill_db: ``DmdFile`` 实例

    Usage::

        from dimine_python_sdk.lib.io import DmdFile
        dmd = DmdFile("钻孔数据.dmd")
        DmgFile().save_from_drill_db_dataframe("output.dmg", dmd)
    """
    if not isinstance(drill_db, DmdFile):
        raise DmgError("drill_db 必须是 DmdFile 实例")

    try:
        segment_df = _drill_db_to_segment_df(drill_db)
        self.save_from_segments(file_path, segment_df)
    except Exception as exc:
        raise DmgError(
            f"从钻孔数据库构建 DMG 分段失败: {file_path}"
        ) from exc

`save_from_segments(file_path, df)`

从分段 DataFrame 构建并保存为 DMG 文件

输入 DataFrame 每行代表某个钻孔下某一段的完整信息，包含起点坐标、终点坐标、样品信息及元素品位等。

必需列： BHID、start_x、start_y、start_z、end_x、end_y、end_z、勘探线、工程类型、ROCK-TYPE、SAMPLE-ID、SAMFROM、SAMTO、样长。

可选列：任意大写元素品位列（如 CU、TFE 等），类型为浮点型。

Parameters:

Name	Type	Description	Default
`file_path`	`str`	目标 .dmg 文件路径	required
`df`	`DataFrame`	钻孔分段 DataFrame	required

Usage::

DmgFile().save_from_segments(
    "output.dmg",
    df,
)

Source code in dimine_python_sdk\lib\io\dmg_file.py

def save_from_segments(
    self,
    file_path: str,
    df: pd.DataFrame,
) -> None:
    """
    从分段 DataFrame 构建并保存为 DMG 文件

    输入 DataFrame 每行代表某个钻孔下某一段的完整信息，包含起点坐标、
    终点坐标、样品信息及元素品位等。

    必需列：
        BHID、start_x、start_y、start_z、end_x、end_y、end_z、
        勘探线、工程类型、ROCK-TYPE、SAMPLE-ID、SAMFROM、SAMTO、样长。

    可选列：
        任意大写元素品位列（如 CU、TFE 等），类型为浮点型。

    Args:
        file_path: 目标 .dmg 文件路径
        df: 钻孔分段 DataFrame

    Usage::

        DmgFile().save_from_segments(
            "output.dmg",
            df,
        )
    """
    try:
        handle = NativeDatabaseHandle.create_local()
        _build_dmg_from_segments(handle, df)
        handle.save(str(file_path))
    except Exception as exc:
        raise DmgError(f"保存 .dmg 失败: {file_path}") from exc

    self.file_path = str(file_path)

`to_segments_dataframe()`

从已加载的 DMG 文件中提取分段 DataFrame

读取"钻孔轨迹"图层中的线段实体，提取起点/终点坐标及属性，重构为与 save_from_segments 输入一致的 DataFrame。

Returns:

Type	Description
`DataFrame`	分段 DataFrame，列包含：
`DataFrame`	BHID、start_x、start_y、start_z、end_x、end_y、end_z、
`DataFrame`	勘探线、工程类型、ROCK-TYPE、SAMPLE-ID、SAMFROM、SAMTO、
`DataFrame`	样长，以及检测到的大写元素品位列。

Raises:

Type	Description
`DmgError`	文件未加载，或"钻孔轨迹"图层不存在/为空

Usage::

dmg = DmgFile("input.dmg")
df = dmg.to_segments_dataframe()
print(df.head())

Source code in dimine_python_sdk\lib\io\dmg_file.py

def to_segments_dataframe(self) -> pd.DataFrame:
    """
    从已加载的 DMG 文件中提取分段 DataFrame

    读取"钻孔轨迹"图层中的线段实体，提取起点/终点坐标及属性，
    重构为与 ``save_from_segments`` 输入一致的 DataFrame。

    Returns:
        分段 DataFrame，列包含：
        BHID、start_x、start_y、start_z、end_x、end_y、end_z、
        勘探线、工程类型、ROCK-TYPE、SAMPLE-ID、SAMFROM、SAMTO、
        样长，以及检测到的大写元素品位列。

    Raises:
        DmgError: 文件未加载，或"钻孔轨迹"图层不存在/为空

    Usage::

        dmg = DmgFile("input.dmg")
        df = dmg.to_segments_dataframe()
        print(df.head())
    """
    if not self.file_path:
        raise DmgError("请先加载 DMG 文件")

    # 查找"钻孔轨迹"图层
    traces_layer = None
    for layer in self.layers:
        if layer.name == _DMG_LAYER_SAMPLES:
            traces_layer = layer
            break

    if traces_layer is None:
        raise DmgError(f"未找到'{_DMG_LAYER_SAMPLES}'图层")
    if not traces_layer.entities:
        raise DmgError(f"'{_DMG_LAYER_SAMPLES}'图层中没有实体")

    rows: list[dict[str, Any]] = []
    all_prop_names: set[str] = set()

    for entity in traces_layer.entities:
        # 只处理线段实体
        if not isinstance(entity, Line):
            continue

        geom = entity.geometry
        if geom is None or geom.shape[0] < 2:
            continue

        # 提取属性
        props: dict[str, Any] = {}
        for p in entity.properties:
            props[p.name] = p.value
            all_prop_names.add(p.name)

        row: dict[str, Any] = {
            "BHID": str(props.get("BHID", "")),
            "start_x": float(geom[0][0]),
            "start_y": float(geom[0][1]),
            "start_z": float(geom[0][2]),
            "end_x": float(geom[1][0]),
            "end_y": float(geom[1][1]),
            "end_z": float(geom[1][2]),
            "勘探线": str(props.get("勘探线号", "")),
            "工程类型": str(props.get("工程类型", "")),
            "ROCK-TYPE": str(props.get("ROCK-TYPE", "")),
            "SAMPLE-ID": props.get("SAMPLE-ID"),
            "SAMFROM": props.get("SAMFROM"),
            "SAMTO": props.get("SAMTO"),
            "样长": props.get("样长"),
        }

        # 提取元素品位列（大写字母列，不在固定列中）
        element_cols = _detect_element_columns(
            pd.DataFrame([props])
        )
        for col in element_cols:
            row[col] = props.get(col)

        rows.append(row)

    if not rows:
        raise DmgError(f"'{_DMG_LAYER_SAMPLES}'图层中没有有效的线段实体")

    return pd.DataFrame(rows)

`DmgValidationError`

Bases: DmgError

DMG 数据校验失败（缺少必需列、无效数据等）

Source code in dimine_python_sdk\lib\io\dmg_file.py

class DmgValidationError(DmgError):
    """DMG 数据校验失败（缺少必需列、无效数据等）"""

    pass

`dmg_conn(file_path=None)`

DMG 文件连接上下文管理器

Source code in dimine_python_sdk\lib\io\dmg_file.py

@contextmanager
def dmg_conn(file_path: Optional[str] = None) -> Generator[DmgFile, None, None]:
    """DMG 文件连接上下文管理器"""
    dmg = DmgFile(file_path)
    try:
        yield dmg
    finally:
        dmg.close()

dmg_file

DmgError

DmgFile

info property

load(file_path)

save_from_drill_db(file_path, drill_db)

save_from_drill_db_dataframe(file_path, drill_db)

save_from_segments(file_path, df)

to_segments_dataframe()

DmgValidationError

dmg_conn(file_path=None)

`DmgError`

`DmgFile`

`info` `property`

`load(file_path)`

`save_from_drill_db(file_path, drill_db)`

`save_from_drill_db_dataframe(file_path, drill_db)`

`save_from_segments(file_path, df)`

`to_segments_dataframe()`

`DmgValidationError`

`dmg_conn(file_path=None)`