dmd_file

DMD 钻孔数据库文件操作

提供 DmdFile 类直接打开、修改、保存 .dmd 钻孔数据库文件。 遵循 io 模块统一接口模式：load / save / close / 上下文管理器。

禁止 import DmPyBindInterface；通过 lib.native.prospecting 间接使用 C++ 扩展。

DmdError

Bases: RuntimeError

DMD 操作异常基类

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

class DmdError(RuntimeError):
    """DMD 操作异常基类"""
    pass

DmdFile

DMD 钻孔数据库文件管理器。

DMD 文件是 Dimine 钻孔数据库的主引用文件（单行 CDataTable）， 包含孔口表、测斜表、岩性表、样品表及扩展表的列名映射和 伴随 .dmt 文件路径。实际数据存储在伴随的 .dmt 子表文件中。

用法::

# 从文件加载
dmd = DmdFile("钻孔数据.dmd")
print(dmd.collar.head())
dmd.survey = new_survey_df
dmd.save()

# 上下文管理器
with dmd_conn("钻孔数据.dmd") as dmd:
    print(dmd.info)
    for name in dmd.table_names:
        print(name, len(dmd[name]))

# 创建空文件并保存
dmd = DmdFile()
dmd.collar = collar_df
dmd.survey = survey_df
dmd.save("output.dmd")

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

class DmdFile:
    """
    DMD 钻孔数据库文件管理器。

    DMD 文件是 Dimine 钻孔数据库的主引用文件（单行 CDataTable），
    包含孔口表、测斜表、岩性表、样品表及扩展表的列名映射和
    伴随 .dmt 文件路径。实际数据存储在伴随的 .dmt 子表文件中。

    用法::

        # 从文件加载
        dmd = DmdFile("钻孔数据.dmd")
        print(dmd.collar.head())
        dmd.survey = new_survey_df
        dmd.save()

        # 上下文管理器
        with dmd_conn("钻孔数据.dmd") as dmd:
            print(dmd.info)
            for name in dmd.table_names:
                print(name, len(dmd[name]))

        # 创建空文件并保存
        dmd = DmdFile()
        dmd.collar = collar_df
        dmd.survey = survey_df
        dmd.save("output.dmd")
    """

    def __init__(self, file_path: Optional[Union[str, Path]] = None):
        self._loaded: bool = False
        self._ref_dict: dict[str, Any] = {}
        self._extend_tables: list[NativeExtendTable] = []
        self._file_path: Optional[str] = None
        self._collar_df: Any = None
        self._survey_df: Any = None
        self._lithology_df: Any = None
        self._sample_df: Any = None

        if file_path is not None:
            self.load(file_path)
        else:
            self._loaded = True  # 空构造即"已打开"，可直接设表保存

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

    def load(self, file_path: Union[str, Path]) -> "DmdFile":
        """加载 .dmd 文件及其伴随 .dmt 子表。

        临时创建 DrillSession 读取全部表数据到缓存后立即关闭，
        不持有持久化的 C++ session。

        Returns:
            self，支持链式调用

        Raises:
            DmdLoadError: 文件不存在或加载失败
        """
        path = Path(file_path)
        if not path.exists():
            raise DmdLoadError(f"文件不存在: {file_path}")

        # 1. 读取引用字典（纯 Python，无需 C++ session）
        try:
            self._ref_dict = _load_dmd_reference(str(path))
        except NativeProspectingError as exc:
            raise DmdLoadError(f"加载 .dmd 失败: {file_path}") from exc
        except Exception as exc:
            raise DmdLoadError(f"加载 .dmd 异常: {file_path}") from exc

        # 2. 临时 session 读取全部表数据后立即关闭
        try:
            session = DrillSession.load(str(path))
        except NativeProspectingError as exc:
            raise DmdLoadError(f"加载 .dmd 失败: {file_path}") from exc
        except Exception as exc:
            raise DmdLoadError(f"加载 .dmd 异常: {file_path}") from exc

        try:
            self._collar_df = session.collar_table()
            self._survey_df = session.survey_table()
            self._lithology_df = session.lithology_table()
            self._sample_df = session.sample_table()
            # 扩展表：在 session 存活期间加载（session 持有 dmd 路径）
            self._extend_tables = session.extend_tables()
        finally:
            session.close()

        # 2.5 检查各表列名是否为标准版
        is_std, warnings = _check_dmd_tables_standard(
            self._collar_df, self._survey_df,
            self._lithology_df, self._sample_df,
        )
        if not is_std:
            logger.warning(
                "DMD 文件为非标准版（表列名与标准定义不一致）: %s\n  详情: %s",
                file_path, "; ".join(warnings),
            )

        self._file_path = str(path)
        self._loaded = True
        return self

    def _invalidate_cache(self) -> None:
        """使缓存的 DataFrame 失效"""
        self._collar_df = None
        self._survey_df = None
        self._lithology_df = None
        self._sample_df = None

    def save(self, output_path: Optional[Union[str, Path]] = None) -> None:
        """保存钻孔数据库（.dmd 引用文件 + 伴随 .dmt 子表）。

        Args:
            output_path: 保存路径。为 None 时使用加载时的路径。

        Raises:
            DmdSaveError: 未指定保存路径或保存失败
        """
        save_path = output_path or self._file_path
        if not save_path:
            raise DmdSaveError("未指定保存路径，请传入 output_path 或先调用 load()")

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

        if not self._loaded:
            raise DmdSaveError("无数据可保存，请先加载文件或设置表数据")

        try:
            save_drill_database(
                self.collar,
                self.survey,
                self.lithology,
                str(save_path),
                sample_df=self.sample,
                extend_tables=self._extend_tables if self._extend_tables else None,
            )
        except Exception as exc:
            raise DmdSaveError(f"保存 .dmd 失败: {save_path}") from exc

        self._file_path = str(save_path)

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

        通过临时 .dmd 文件中转：先将当前表数据保存为临时 .dmd，
        再用临时 DrillSession 加载并转换为 .dmg。

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

        Raises:
            DmdSaveError: 保存失败
        """
        self._check_open()
        save_path = _normalize_dmg_path(output_path)
        self._save_as_dmg_via_temp_dmd(save_path)

    def _save_as_dmg_via_temp_dmd(self, output_path: str) -> None:
        """通过临时 .dmd 文件间接保存为 .dmg（用于空 session 场景）"""
        import shutil
        import tempfile
        import os
        tmp_dir = tempfile.mkdtemp(prefix="dimine_drill_")
        try:
            tmp_dmd = os.path.join(tmp_dir, "_temp.dmd")
            save_drill_database(
                collar_df=self.collar,
                survey_df=self.survey,
                lithology_df=self.lithology,
                save_path=tmp_dmd,
                sample_df=self.sample,
                extend_tables=self._extend_tables if self._extend_tables else None,
            )
            tmp_session = DrillSession.load(tmp_dmd)
            try:
                tmp_session.save_as_dmg(output_path)
            finally:
                tmp_session.close()
        finally:
            shutil.rmtree(tmp_dir, ignore_errors=True)

    def close(self) -> None:
        """释放资源，解除文件占用"""
        self._loaded = False
        self._ref_dict = {}
        self._extend_tables = []
        self._file_path = None
        self._invalidate_cache()

    # ------------------------------------------------------------------
    # 内部 helpers
    # ------------------------------------------------------------------

    def _check_open(self) -> None:
        """检查 DmdFile 是否已打开"""
        if not self._loaded:
            raise DmdError("DmdFile 已关闭或未初始化，请先调用 load() 加载钻孔数据库")

    # ------------------------------------------------------------------
    # 核心表属性（直接操作缓存 DataFrame，纯 Python 对象）
    # ------------------------------------------------------------------

    @property
    def collar(self) -> "pd.DataFrame":
        """钻孔孔口表（pd.DataFrame）"""
        self._check_open()
        return self._collar_df

    @collar.setter
    def collar(self, df: "pd.DataFrame") -> None:
        """设置钻孔孔口表（校验列名，更新缓存）"""
        self._check_open()
        _validate_dmd_columns(df, COLLAR_COLUMN_MAP, "孔口表", required=list(_COLLAR_REQUIRED))
        self._collar_df = df

    @property
    def survey(self) -> "pd.DataFrame":
        """钻孔测斜表（pd.DataFrame）"""
        self._check_open()
        return self._survey_df

    @survey.setter
    def survey(self, df: "pd.DataFrame") -> None:
        """设置钻孔测斜表（校验列名，更新缓存）"""
        self._check_open()
        _validate_dmd_columns(df, SURVEY_COLUMN_MAP, "测斜表", required=list(_SURVEY_REQUIRED))
        self._survey_df = df

    @property
    def lithology(self) -> "pd.DataFrame":
        """钻孔岩性表（pd.DataFrame）"""
        self._check_open()
        return self._lithology_df

    @lithology.setter
    def lithology(self, df: "pd.DataFrame") -> None:
        """设置钻孔岩性表（校验列名，更新缓存）"""
        self._check_open()
        _validate_dmd_columns(df, LITHOLOGY_COLUMN_MAP, "岩性表", required=list(_LITHOLOGY_REQUIRED))
        self._lithology_df = df

    @property
    def sample(self) -> "pd.DataFrame":
        """钻孔样品表（pd.DataFrame）"""
        self._check_open()
        return self._sample_df

    @sample.setter
    def sample(self, df: "pd.DataFrame") -> None:
        """设置钻孔样品表（校验列名，更新缓存）"""
        self._check_open()
        _validate_dmd_columns(df, SAMPLE_COLUMN_MAP, "样品表", required=list(_SAMPLE_REQUIRED))
        self._sample_df = df

    # ------------------------------------------------------------------
    # 扩展表
    # ------------------------------------------------------------------

    @property
    def extend_tables(self) -> list[NativeExtendTable]:
        """扩展表列表。

        每项为 NativeExtendTable：
        - df: pd.DataFrame           扩展表数据（保留 DMT 原始列名）
        - dmt_filename: str          .dmt 文件名（如 "分层表.dmt"）
        - field_list: str            字段元数据（4 个逗号分隔值，用于 roundtrip）
        - table_name: str            表名（优先 display_title，其次 dmt 文件名）
        - hole_id_column: str|None   钻孔编号对应列名（field_list[0]）
        - from_column: str|None      起始深度对应列名（field_list[1]）
        - to_column: str|None        结束深度对应列名（field_list[2]）
        - display_title: str|None    标题名称（field_list[3]）
        """
        return self._extend_tables

    @extend_tables.setter
    def extend_tables(self, tables: list[NativeExtendTable | dict[str, Any]]) -> None:
        """设置扩展表列表。

        每项可为 NativeExtendTable 或 dict（向后兼容）。
        dict 必须包含 'df' 和 'dmt_filename'。
        NativeExtendTable 的 field_list ↔ 元数据双向推导由
        :class:`NativeExtendTable.__post_init__` 自动完成。
        """
        converted: list[NativeExtendTable] = []
        for t in tables:
            if isinstance(t, NativeExtendTable):
                converted.append(t)
            elif isinstance(t, dict):
                converted.append(self._dict_to_extend_table(t))
            else:
                raise DmdError(
                    f"扩展表每项必须为 NativeExtendTable 或 dict，收到: {type(t).__name__}"
                )
        self._extend_tables = converted

    @staticmethod
    def _dict_to_extend_table(d: dict[str, Any]) -> NativeExtendTable:
        """将 dict 转换为 NativeExtendTable（向后兼容）。

        NativeExtendTable.__post_init__ 会自动完成
        field_list ↔ 元数据的双向推导。
        """
        if "df" not in d or "dmt_filename" not in d:
            raise DmdError("扩展表 dict 必须包含 'df' 和 'dmt_filename'")

        return NativeExtendTable(
            df=d["df"],
            dmt_filename=d["dmt_filename"],
            field_list=d.get("field_list", ""),
            hole_id_column=d.get("hole_id_column"),
            from_column=d.get("from_column"),
            to_column=d.get("to_column"),
            display_title=d.get("display_title"),
            table_name=d.get("table_name", ""),
        )

    # ------------------------------------------------------------------
    # 元数据属性
    # ------------------------------------------------------------------

    @property
    def file_path(self) -> Optional[str]:
        """当前文件路径"""
        return self._file_path

    @property
    def ref_dict(self) -> dict:
        """钻孔数据库引用字典（.dmd 文件内容，只读）"""
        self._check_open()
        return self._ref_dict

    @property
    def name(self) -> str:
        """文件名（不含扩展名）"""
        return Path(self._file_path).stem if self._file_path else ""

    @property
    def table_names(self) -> list[str]:
        """所有表名列表（含扩展表）"""
        self._check_open()
        names = ["collar", "survey", "lithology", "sample"]
        for ext in self._extend_tables:
            if ext.table_name:
                names.append(ext.table_name)
            # display_title 与 table_name 不同时一并加入
            if ext.display_title and ext.display_title != ext.table_name:
                names.append(ext.display_title)
        return names

    @property
    def info(self) -> dict[str, Any]:
        """DMD 文件摘要信息"""
        if not self._loaded:
            return {
                "file": self._file_path or "",
                "name": self.name,
                "loaded": False,
            }

        def _table_info(df):
            if df is None:
                return {"columns": [], "rows": 0}
            return {"columns": list(df.columns), "rows": len(df)}

        return {
            "file": self._file_path or "",
            "name": self.name,
            "loaded": True,
            "tables": {
                "collar": _table_info(self.collar),
                "survey": _table_info(self.survey),
                "lithology": _table_info(self.lithology),
                "sample": _table_info(self.sample),
            },
            "extend_table_count": len(self._extend_tables),
            "extend_table_names": [e.table_name for e in self._extend_tables],
            "extend_tables": [
                {
                    "table_name": e.table_name,
                    "dmt_filename": e.dmt_filename,
                    "display_title": e.display_title,
                    "hole_id_column": e.hole_id_column,
                    "from_column": e.from_column,
                    "to_column": e.to_column,
                    "rows": len(e.df) if e.df is not None else 0,
                    "columns": list(e.df.columns) if e.df is not None else [],
                }
                for e in self._extend_tables
            ],
        }

    # ------------------------------------------------------------------
    # Pythonic 访问
    # ------------------------------------------------------------------

    def __getitem__(self, table_name: str) -> "pd.DataFrame":
        """统一表访问：dmd['collar'], dmd['survey'], dmd['lithology'], dmd['sample']"""
        self._check_open()
        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

        # 查找扩展表
        for ext in self._extend_tables:
            if ext.table_name == table_name:
                return ext.df
        # 也尝试按 display_title 匹配
        for ext in self._extend_tables:
            if ext.display_title == table_name:
                return ext.df

        raise KeyError(
            f"未知表: {table_name}。可用表: {self.table_names}"
        )

    def __bool__(self) -> bool:
        """已加载返回 True，已关闭返回 False"""
        return self._loaded

    def __len__(self) -> int:
        """返回表总数（含扩展表）"""
        return len(self.table_names)

    def __contains__(self, table_name: str) -> bool:
        """检查表名是否存在"""
        return table_name in self.table_names

    def __iter__(self):
        """按表名迭代"""
        return iter(self.table_names)

    def __repr__(self) -> str:
        if not self._loaded:
            return f"DmdFile('{self._file_path or ''}', closed)"
        n_tables = len(self._extend_tables) + 4  # 4 base tables
        return f"DmdFile('{self._file_path or ''}', loaded, tables={n_tables})"

    def __enter__(self) -> "DmdFile":
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self.close()

collar property writable

钻孔孔口表（pd.DataFrame）

extend_tables property writable

扩展表列表。

每项为 NativeExtendTable： - df: pd.DataFrame 扩展表数据（保留 DMT 原始列名） - dmt_filename: str .dmt 文件名（如 "分层表.dmt"） - field_list: str 字段元数据（4 个逗号分隔值，用于 roundtrip） - table_name: str 表名（优先 display_title，其次 dmt 文件名） - hole_id_column: str|None 钻孔编号对应列名（field_list[0]） - from_column: str|None 起始深度对应列名（field_list[1]） - to_column: str|None 结束深度对应列名（field_list[2]） - display_title: str|None 标题名称（field_list[3]）

file_path property

当前文件路径

info property

DMD 文件摘要信息

lithology property writable

钻孔岩性表（pd.DataFrame）

name property

文件名（不含扩展名）

ref_dict property

钻孔数据库引用字典（.dmd 文件内容，只读）

sample property writable

钻孔样品表（pd.DataFrame）

survey property writable

钻孔测斜表（pd.DataFrame）

table_names property

所有表名列表（含扩展表）

__bool__()

已加载返回 True，已关闭返回 False

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

def __bool__(self) -> bool:
    """已加载返回 True，已关闭返回 False"""
    return self._loaded

__contains__(table_name)

检查表名是否存在

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

def __contains__(self, table_name: str) -> bool:
    """检查表名是否存在"""
    return table_name in self.table_names

__getitem__(table_name)

统一表访问：dmd['collar'], dmd['survey'], dmd['lithology'], dmd['sample']

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

def __getitem__(self, table_name: str) -> "pd.DataFrame":
    """统一表访问：dmd['collar'], dmd['survey'], dmd['lithology'], dmd['sample']"""
    self._check_open()
    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

    # 查找扩展表
    for ext in self._extend_tables:
        if ext.table_name == table_name:
            return ext.df
    # 也尝试按 display_title 匹配
    for ext in self._extend_tables:
        if ext.display_title == table_name:
            return ext.df

    raise KeyError(
        f"未知表: {table_name}。可用表: {self.table_names}"
    )

__iter__()

按表名迭代

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

def __iter__(self):
    """按表名迭代"""
    return iter(self.table_names)

__len__()

返回表总数（含扩展表）

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

def __len__(self) -> int:
    """返回表总数（含扩展表）"""
    return len(self.table_names)

close()

释放资源，解除文件占用

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

def close(self) -> None:
    """释放资源，解除文件占用"""
    self._loaded = False
    self._ref_dict = {}
    self._extend_tables = []
    self._file_path = None
    self._invalidate_cache()

load(file_path)

加载 .dmd 文件及其伴随 .dmt 子表。

临时创建 DrillSession 读取全部表数据到缓存后立即关闭， 不持有持久化的 C++ session。

Returns:

Type	Description
'DmdFile'	self，支持链式调用

Raises:

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

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

def load(self, file_path: Union[str, Path]) -> "DmdFile":
    """加载 .dmd 文件及其伴随 .dmt 子表。

    临时创建 DrillSession 读取全部表数据到缓存后立即关闭，
    不持有持久化的 C++ session。

    Returns:
        self，支持链式调用

    Raises:
        DmdLoadError: 文件不存在或加载失败
    """
    path = Path(file_path)
    if not path.exists():
        raise DmdLoadError(f"文件不存在: {file_path}")

    # 1. 读取引用字典（纯 Python，无需 C++ session）
    try:
        self._ref_dict = _load_dmd_reference(str(path))
    except NativeProspectingError as exc:
        raise DmdLoadError(f"加载 .dmd 失败: {file_path}") from exc
    except Exception as exc:
        raise DmdLoadError(f"加载 .dmd 异常: {file_path}") from exc

    # 2. 临时 session 读取全部表数据后立即关闭
    try:
        session = DrillSession.load(str(path))
    except NativeProspectingError as exc:
        raise DmdLoadError(f"加载 .dmd 失败: {file_path}") from exc
    except Exception as exc:
        raise DmdLoadError(f"加载 .dmd 异常: {file_path}") from exc

    try:
        self._collar_df = session.collar_table()
        self._survey_df = session.survey_table()
        self._lithology_df = session.lithology_table()
        self._sample_df = session.sample_table()
        # 扩展表：在 session 存活期间加载（session 持有 dmd 路径）
        self._extend_tables = session.extend_tables()
    finally:
        session.close()

    # 2.5 检查各表列名是否为标准版
    is_std, warnings = _check_dmd_tables_standard(
        self._collar_df, self._survey_df,
        self._lithology_df, self._sample_df,
    )
    if not is_std:
        logger.warning(
            "DMD 文件为非标准版（表列名与标准定义不一致）: %s\n  详情: %s",
            file_path, "; ".join(warnings),
        )

    self._file_path = str(path)
    self._loaded = True
    return self

save(output_path=None)

保存钻孔数据库（.dmd 引用文件 + 伴随 .dmt 子表）。

Parameters:

Name	Type	Description	Default
output_path	Optional[Union[str, Path]]	保存路径。为 None 时使用加载时的路径。	None

Raises:

Type	Description
DmdSaveError	未指定保存路径或保存失败

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

def save(self, output_path: Optional[Union[str, Path]] = None) -> None:
    """保存钻孔数据库（.dmd 引用文件 + 伴随 .dmt 子表）。

    Args:
        output_path: 保存路径。为 None 时使用加载时的路径。

    Raises:
        DmdSaveError: 未指定保存路径或保存失败
    """
    save_path = output_path or self._file_path
    if not save_path:
        raise DmdSaveError("未指定保存路径，请传入 output_path 或先调用 load()")

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

    if not self._loaded:
        raise DmdSaveError("无数据可保存，请先加载文件或设置表数据")

    try:
        save_drill_database(
            self.collar,
            self.survey,
            self.lithology,
            str(save_path),
            sample_df=self.sample,
            extend_tables=self._extend_tables if self._extend_tables else None,
        )
    except Exception as exc:
        raise DmdSaveError(f"保存 .dmd 失败: {save_path}") from exc

    self._file_path = str(save_path)

save_as_dmg(output_path)

保存为 .dmg 可视化格式。

通过临时 .dmd 文件中转：先将当前表数据保存为临时 .dmd， 再用临时 DrillSession 加载并转换为 .dmg。

Parameters:

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

Raises:

Type	Description
DmdSaveError	保存失败

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

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

    通过临时 .dmd 文件中转：先将当前表数据保存为临时 .dmd，
    再用临时 DrillSession 加载并转换为 .dmg。

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

    Raises:
        DmdSaveError: 保存失败
    """
    self._check_open()
    save_path = _normalize_dmg_path(output_path)
    self._save_as_dmg_via_temp_dmd(save_path)

DmdLoadError

Bases: DmdError

加载 .dmd 文件失败

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

class DmdLoadError(DmdError):
    """加载 .dmd 文件失败"""
    pass

DmdSaveError

Bases: DmdError

保存 .dmd 文件失败

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

class DmdSaveError(DmdError):
    """保存 .dmd 文件失败"""
    pass

dmd_conn(file_path=None)

DMD 文件连接上下文管理器

Usage::

with dmd_conn("钻孔数据.dmd") as dmd:
    print(dmd.collar.head())

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

@contextmanager
def dmd_conn(file_path: Optional[Union[str, Path]] = None) -> Generator[DmdFile, None, None]:
    """DMD 文件连接上下文管理器

    Usage::

        with dmd_conn("钻孔数据.dmd") as dmd:
            print(dmd.collar.head())
    """
    dmd = DmdFile(file_path)
    try:
        yield dmd
    finally:
        dmd.close()