dmo_file

DMO 炮孔数据库文件业务层：DmoFile、dmo_conn 与相关异常

DMO 文件本质上是 DMT 数据表文件，但拥有固定表头（炮孔数据库标准字段）。 本模块基于 :class:DmtFile，额外提供：

• 固定表头 schema（字段名 + 字段类型，新版 26 列）
• 加载时宽松校验（缺列 / 多列 / 类型不符仅记录警告，不抛异常）
• 保存时按 schema 还原字段类型，避免 pandas dtype 推断丢失 LONG/BINARY 等类型

允许类型：pd.DataFrame、Path、str 禁止类型：Dm* 禁止 import DmPyBindInterface

DmoError

Bases: RuntimeError

DMO 操作异常基类

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

class DmoError(RuntimeError):
    """DMO 操作异常基类"""

    pass

DmoFile

Bases: DmtFile

DMO 炮孔数据库文件管理器。

DMO 是 Dimine 炮孔数据库（爆破设计）使用的数据表文件，本质上是带有 固定表头的 DMT 文件。本类继承 :class:DmtFile，加载时对表头做宽松校验， 保存时按 :data:DMO_FIELD_SCHEMA 还原字段类型。

用法::

# 从文件加载
dmo = DmoFile("炮孔数据库-data.dmt")
print(dmo.columns)
print(dmo.data.head())
print(dmo.header_warnings)  # 非标准表头告警（如有）

# 修改后保存
dmo.data.loc[0, "孔径"] = 250.0
dmo.save("output.dmo")

# 从 DataFrame 创建并保存
DmoFile(data=df).save("new.dmo")

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

class DmoFile(DmtFile):
    """
    DMO 炮孔数据库文件管理器。

    DMO 是 Dimine 炮孔数据库（爆破设计）使用的数据表文件，本质上是带有
    固定表头的 DMT 文件。本类继承 :class:`DmtFile`，加载时对表头做宽松校验，
    保存时按 :data:`DMO_FIELD_SCHEMA` 还原字段类型。

    用法::

        # 从文件加载
        dmo = DmoFile("炮孔数据库-data.dmt")
        print(dmo.columns)
        print(dmo.data.head())
        print(dmo.header_warnings)  # 非标准表头告警（如有）

        # 修改后保存
        dmo.data.loc[0, "孔径"] = 250.0
        dmo.save("output.dmo")

        # 从 DataFrame 创建并保存
        DmoFile(data=df).save("new.dmo")
    """

    def __init__(
        self,
        file_path: Optional[Union[str, Path]] = None,
        data: Optional["pd.DataFrame"] = None,
    ):
        self._header_warnings: List[str] = []
        super().__init__(file_path=file_path, data=data)

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

    def load(self, file_path: Union[str, Path]) -> "DmoFile":
        """加载 .dmo 文件到 DataFrame，并对表头做宽松校验。

        Returns:
            self（支持链式调用）

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

        try:
            handle = NativeDataTableHandle.create()
            handle.load(str(path))
            self._df = handle.to_dataframe()
            actual_types = {
                d.name: NativeFieldType(d.type).name.lower()
                for d in handle.field_definitions()
            }
            self._header_warnings = self._validate_header(
                list(self._df.columns), actual_types
            )
        except NativeDataTableError as exc:
            raise DmoLoadError(f"加载 .dmo 失败: {file_path}") from exc
        except Exception as exc:
            raise DmoLoadError(f"加载 .dmo 异常: {file_path}") from exc

        self.file_path = str(path)
        for warning in self._header_warnings:
            logger.warning("DMO 文件表头非标准: %s: %s", file_path, warning)
        return self

    def save(self, file_path: Optional[Union[str, Path]] = None) -> None:
        """将当前 DataFrame 保存为 .dmo 文件（按固定表头还原字段类型）。

        Args:
            file_path: 目标路径，为空时使用实例中已记录的路径

        Raises:
            DmoSaveError: 未指定路径或无数据时抛出
        """
        save_path = file_path or self.file_path
        if not save_path:
            raise DmoSaveError("未指定保存路径")

        if self._df is None:
            raise DmoSaveError("无数据可保存")

        save_path = _normalize_dmo_path(save_path)

        try:
            handle = self._build_native_handle()
            handle.save(str(save_path))
        except NativeDataTableError as exc:
            raise DmoSaveError(f"保存 .dmo 失败: {save_path}") from exc
        except Exception as exc:
            raise DmoSaveError(f"保存 .dmo 异常: {save_path}") from exc

        self.file_path = str(save_path)

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

    def _build_native_handle(self) -> NativeDataTableHandle:
        """按固定表头构建 native 数据表句柄。

        固定表头内的列使用 schema 中的类型；非标准列回退为按 dtype 推断。
        写入后字段顺序与 DataFrame 列顺序一致（保持加载时的兼容性）。
        """
        df = self._df
        handle = NativeDataTableHandle.create()
        for col in df.columns:
            ftype = DMO_FIELD_TYPES.get(col) or _infer_field_type(df[col])
            handle.add_field(col, ftype)
        handle.insert_from_dataframe(df)
        return handle

    @staticmethod
    def _validate_header(
        columns: list[str], actual_types: dict[str, str]
    ) -> list[str]:
        """宽松校验表头，返回告警信息列表（不抛异常）。

        - 缺少标准列、含非标准列、字段类型与 schema 不符时各记一条告警
        """
        warnings: list[str] = []
        standard = set(DMO_FIELD_NAMES)
        actual = set(columns)

        missing = [c for c in DMO_FIELD_NAMES if c not in actual]
        if missing:
            warnings.append(f"缺少标准列: {missing}")

        extra = [c for c in columns if c not in standard]
        if extra:
            warnings.append(f"含非标准列: {extra}")

        for name, ftype in DMO_FIELD_SCHEMA:
            if name in actual_types and actual_types[name] != ftype:
                warnings.append(
                    f"字段 '{name}' 类型不符: 期望 {ftype}, 实际 {actual_types[name]}"
                )

        return warnings

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

    @property
    def header_warnings(self) -> list[str]:
        """加载时表头校验产生的告警列表（无告警时为空）"""
        return list(self._header_warnings)

    @property
    def schema(self) -> list[dict[str, str]]:
        """固定表头 schema（``[{"name": ..., "type": ...}, ...]``）"""
        return [{"name": name, "type": ftype} for name, ftype in DMO_FIELD_SCHEMA]

    @property
    def info(self) -> dict[str, Any]:
        """DMO 文件摘要信息（在 DmtFile.info 基础上附加表头告警）"""
        info = super().info
        info["header_warnings"] = self._header_warnings
        return info

    def __repr__(self) -> str:
        rows, cols = self.shape
        return f"DmoFile('{self.file_path or ''}', rows={rows}, cols={cols})"

header_warnings property

加载时表头校验产生的告警列表（无告警时为空）

info property

DMO 文件摘要信息（在 DmtFile.info 基础上附加表头告警）

schema property

固定表头 schema（[{"name": ..., "type": ...}, ...]）

load(file_path)

加载 .dmo 文件到 DataFrame，并对表头做宽松校验。

Returns:

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

Raises:

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

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

def load(self, file_path: Union[str, Path]) -> "DmoFile":
    """加载 .dmo 文件到 DataFrame，并对表头做宽松校验。

    Returns:
        self（支持链式调用）

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

    try:
        handle = NativeDataTableHandle.create()
        handle.load(str(path))
        self._df = handle.to_dataframe()
        actual_types = {
            d.name: NativeFieldType(d.type).name.lower()
            for d in handle.field_definitions()
        }
        self._header_warnings = self._validate_header(
            list(self._df.columns), actual_types
        )
    except NativeDataTableError as exc:
        raise DmoLoadError(f"加载 .dmo 失败: {file_path}") from exc
    except Exception as exc:
        raise DmoLoadError(f"加载 .dmo 异常: {file_path}") from exc

    self.file_path = str(path)
    for warning in self._header_warnings:
        logger.warning("DMO 文件表头非标准: %s: %s", file_path, warning)
    return self

save(file_path=None)

将当前 DataFrame 保存为 .dmo 文件（按固定表头还原字段类型）。

Parameters:

Name	Type	Description	Default
file_path	Optional[Union[str, Path]]	目标路径，为空时使用实例中已记录的路径	None

Raises:

Type	Description
DmoSaveError	未指定路径或无数据时抛出

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

def save(self, file_path: Optional[Union[str, Path]] = None) -> None:
    """将当前 DataFrame 保存为 .dmo 文件（按固定表头还原字段类型）。

    Args:
        file_path: 目标路径，为空时使用实例中已记录的路径

    Raises:
        DmoSaveError: 未指定路径或无数据时抛出
    """
    save_path = file_path or self.file_path
    if not save_path:
        raise DmoSaveError("未指定保存路径")

    if self._df is None:
        raise DmoSaveError("无数据可保存")

    save_path = _normalize_dmo_path(save_path)

    try:
        handle = self._build_native_handle()
        handle.save(str(save_path))
    except NativeDataTableError as exc:
        raise DmoSaveError(f"保存 .dmo 失败: {save_path}") from exc
    except Exception as exc:
        raise DmoSaveError(f"保存 .dmo 异常: {save_path}") from exc

    self.file_path = str(save_path)

DmoLoadError

Bases: DmoError

加载 .dmo 文件失败

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

class DmoLoadError(DmoError):
    """加载 .dmo 文件失败"""

    pass

DmoSaveError

Bases: DmoError

保存 .dmo 文件失败

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

class DmoSaveError(DmoError):
    """保存 .dmo 文件失败"""

    pass

dmo_conn(file_path=None)

DMO 文件连接上下文管理器

Usage::

with dmo_conn("炮孔数据库.dmo") as dmo:
    print(dmo.columns)

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

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

    Usage::

        with dmo_conn("炮孔数据库.dmo") as dmo:
            print(dmo.columns)
    """
    dmo = DmoFile(file_path)
    try:
        yield dmo
    finally:
        dmo.close()