dmo_file

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

DMO 文件本质上是 DMT 数据表文件，但拥有固定表头（炮孔数据库标准字段）。 本模块的 :class:DmoFile 不继承 :class:DmtFile，改为组合：仅在读取 / 写回 dmo 文件时调用 DmtFile / native 接口，且不持有、不暴露 DataFrame。额外提供：

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

对外输入/输出为类型化模型（DataFrame 仅在 load/save 内部作为临时物）：

• :class:DmoHole —— 炮孔结构信息（单孔，装药信息平铺；含装药分段）
• :class:DmoNetwork —— 连线信息（内存图模型：nodes/edges 引用 DmoHole 对象， 延时/起爆孔按编号字符串键控）

DmoFile.holes / DmoFile.network 属性可读可写：读取文件时一次构建并缓存， 就地修改模型后 save() 重建 DataFrame 写回。底层 binary 列的（反）序列化由 内部函数承载（以下均为模块内部实现，不对外开放；外部请使用类型化 API）：

• :class:ChargeSegment / :class:ChargeResult —— 装药分段 / 单孔装药结果模型
• _encode_charge_structure / _decode_charge_structure —— 与 「装药结构」列的原生二进制互转（格式见 _decode_charge_structure）
• :class:BlastLine —— 单条连线记录（终点坐标 + 显示属性）
• _encode_blast_lines / _decode_blast_lines —— 与「连线信息」列 的原生二进制互转（格式见 _decode_blast_lines）
• _delay_hole_records / _encode_delay_hole —— 「孔内分段起爆」列 的原生二进制生成（只写不读，格式见 _encode_delay_hole）
• read_charge_structures / write_charge_structures —— 整列读写， 仅供内部过渡使用，不对外公开

注意：native import 均为方法内懒加载，模块本身不依赖 DmPyBindInterface 即可导入。

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

BlastLine

Bases: ParamModel

单条炮孔连线（从本孔孔口指向终点）。

对应「连线信息」二进制里的一条记录（7 个 double）。起点恒为本孔孔口坐标 （C++ 的 dPtCente），二进制不存储起点；end_x/y/z 为目标孔孔口坐标。

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

class BlastLine(ParamModel):
    """单条炮孔连线（从本孔孔口指向终点）。

    对应「连线信息」二进制里的一条记录（7 个 double）。起点恒为本孔孔口坐标
    （C++ 的 ``dPtCente``），二进制不存储起点；``end_x/y/z`` 为目标孔孔口坐标。
    """

    end_x: float = Field(default=0.0, description="终点X（目标孔孔口坐标）")
    end_y: float = Field(default=0.0, description="终点Y（目标孔孔口坐标）")
    end_z: float = Field(default=0.0, description="终点Z（目标孔孔口坐标）")
    arrow_size: float = Field(default=1.0, description="箭头尺寸")
    delay: float = Field(default=0.0, description="延时雷管 延时 ms")
    text_height: float = Field(default=1.0, description="文字高度")
    line_width: float = Field(default=1.0, description="线宽")

ChargeResult

Bases: ParamModel

单孔装药结果（由装药设计计算得到，可写回 DMO 各装药列）

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

class ChargeResult(ParamModel):
    """单孔装药结果（由装药设计计算得到，可写回 DMO 各装药列）"""

    hole_number: str
    depth: float
    charge_length: float = Field(..., description="装药总长度 m")
    charge_weight: float = Field(..., description="装药总量 kg")
    stemming_length: float = Field(..., description="填塞长度 m")
    spacer_length: float = Field(..., description="间隔总长度 m")
    segments: List[ChargeSegment] = Field(..., description="分段结构")
    specific_charge_area: float = Field(..., description="单耗面积（负担面积）")
    warnings: List[str] = Field(default_factory=list, description="红色警告（如填塞不足）")

ChargeSegment

Bases: ParamModel

装药分段（填塞物 / 炸药 / 间隔器 / 空气/ 起爆药包）

对应「装药结构」二进制里的一个分段：段名、长度、雷管延时。 二进制格式不包含重量，反序列化时 weight 恒为 0.0。

雷管延时（delay）语义：惰性段（填塞物 / 间隔器 / 空气）不参与起爆， 真实 Dimine 文件里为负（-1）；炸药 / 起爆药包等为 >= 0 的起爆延时。 其中「孔内起爆时间」由孔口到第一段炸药段的雷管延时之和推导 （见 :func:first_explosive_relay）。

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

class ChargeSegment(ParamModel):
    """装药分段（填塞物 / 炸药 / 间隔器 / 空气/ 起爆药包）

    对应「装药结构」二进制里的一个分段：段名、长度、雷管延时。
    二进制格式不包含重量，反序列化时 ``weight`` 恒为 0.0。

    雷管延时（``delay``）语义：惰性段（填塞物 / 间隔器 / 空气）不参与起爆，
    真实 Dimine 文件里为负（-1）；炸药 / 起爆药包等为 ``>= 0`` 的起爆延时。
    其中「孔内起爆时间」由孔口到第一段炸药段的雷管延时之和推导
    （见 :func:`first_explosive_relay`）。
    """

    fill_type: str = Field(..., description="填塞物 / 炸药 / 间隔器 / 空气 / 起爆药包")
    length: float = Field(..., ge=0, description="长度 m")
    weight: float = Field(default=0.0, ge=0, description="重量 kg（炸药段才非零）")
    delay: int = Field(
        default=0,
        description="雷管延时 ms（>=0 参与起爆；惰性段不参与起爆，真实文件用 -1）",
    )

DmoBlastLineError

Bases: DmoError

连线信息序列化 / 反序列化失败（「连线信息」列 hex 值解析或构造异常）

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

class DmoBlastLineError(DmoError):
    """连线信息序列化 / 反序列化失败（「连线信息」列 hex 值解析或构造异常）"""

    pass

DmoChargeStructureError

Bases: DmoError

装药结构序列化 / 反序列化失败（「装药结构」列 hex 值解析或构造异常）

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

class DmoChargeStructureError(DmoError):
    """装药结构序列化 / 反序列化失败（「装药结构」列 hex 值解析或构造异常）"""

    pass

DmoError

Bases: RuntimeError

DMO 操作异常基类

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

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

    pass

DmoFile

DMO 炮孔数据库文件管理器（纯模型容器）。

DMO 是 Dimine 炮孔数据库（爆破设计）使用的数据表文件，本质上是带有 固定表头的 DMT 文件。本类持有 :attr:holes（炮孔结构）与 :attr:network （连线信息）类型化模型，读取文件时一次构建并缓存；DataFrame 仅在 load/save 内部作为临时物，不对外暴露。「孔内分段起爆」列只写不读： 保存时按装药分段 + 孔口/孔底坐标 + 地表起爆时间自动推导 （见 :func:_delay_hole_records）。

用法::

# 从文件加载（读取后 holes/network 立即可用）
dmo = DmoFile("炮孔数据库.dmo")
dmo.network.nodes           # 炮孔节点（DmoHole 对象）
dmo.holes[0].depth          # 孔参数

# 修改后保存（就地修改模型）
dmo.holes[0].diameter = 250.0
dmo.save("output.dmo")

# 从模型构建并保存
dmo = DmoFile()
dmo.holes = [DmoHole(number="A-01", ...)]
dmo.network = DmoNetwork(nodes=[...], edges=[...], delays=...)
dmo.save("new.dmo")

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

class DmoFile:
    """
    DMO 炮孔数据库文件管理器（纯模型容器）。

    DMO 是 Dimine 炮孔数据库（爆破设计）使用的数据表文件，本质上是带有
    固定表头的 DMT 文件。本类持有 :attr:`holes`（炮孔结构）与 :attr:`network`
    （连线信息）类型化模型，读取文件时一次构建并缓存；DataFrame 仅在
    load/save 内部作为临时物，不对外暴露。「孔内分段起爆」列只写不读：
    保存时按装药分段 + 孔口/孔底坐标 + 地表起爆时间自动推导
    （见 :func:`_delay_hole_records`）。

    用法::

        # 从文件加载（读取后 holes/network 立即可用）
        dmo = DmoFile("炮孔数据库.dmo")
        dmo.network.nodes           # 炮孔节点（DmoHole 对象）
        dmo.holes[0].depth          # 孔参数

        # 修改后保存（就地修改模型）
        dmo.holes[0].diameter = 250.0
        dmo.save("output.dmo")

        # 从模型构建并保存
        dmo = DmoFile()
        dmo.holes = [DmoHole(number="A-01", ...)]
        dmo.network = DmoNetwork(nodes=[...], edges=[...], delays=...)
        dmo.save("new.dmo")
    """

    def __init__(
        self,
        file_path: Optional[Union[str, Path]] = None,
    ):
        self._holes: List[DmoHole] = []
        self._network: DmoNetwork = DmoNetwork()
        self._header_warnings: List[str] = []
        self.file_path: Optional[str] = None
        if file_path is not None:
            self.load(file_path)

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

    def load(self, file_path: Union[str, Path]) -> "DmoFile":
        """读取 .dmo 文件，构建 holes / network 模型并缓存。

        DataFrame 仅在此处作为临时物解析后丢弃；读取后 :attr:`holes` /
        :attr:`network` 立即可用。

        Returns:
            self（支持链式调用）

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

        from dimine_python_sdk.lib.native.data_table import NativeFieldType

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

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

    def recompute_delays(
        self, burn_velocity: Optional[float] = None
    ) -> "DmoFile":
        """沿当前连线网络重算 :attr:`network.delays`（地表起爆时间）。

        即「设计过 / 修改过连线信息后重新生成地表起爆时间」的显式入口：自各起爆点 /
        根孔向外逐孔传播（父孔 + 孔间延时雷管 + 孔口距离/燃速，数码电子雷管即
        ``network.count_line_burn=False`` 时不计连线燃烧时长），未达节点保留原值。
        无连线时无操作。

        Args:
            burn_velocity: 导爆管燃速 m/s；为空时取 Dimine 全局
                ``HoleParam.tube_burn_velocity``（缺 DIMINE_HOME 回退 2000）

        Returns:
            self（支持链式调用）

        Raises:
            DmoError: 燃速非正
        """
        if burn_velocity is None:
            burn_velocity = _default_burn_velocity()
        self._network.compute_delays(burn_velocity=float(burn_velocity))
        return self

    def save(
        self,
        file_path: Optional[Union[str, Path]] = None,
        *,
        burn_velocity: Optional[float] = None,
    ) -> None:
        """把当前 holes / network 模型保存为 .dmo 文件。

        保存时从模型重建 DataFrame（只含标准列），按固定表头还原字段类型；
        「孔内分段起爆」列不落入模型，保存时按装药分段与地表起爆时间自动推导。

        只要模型持有了连线信息（``network.edges`` 非空，即设计过/修改过连线网络），
        保存前会沿网络重算 ``delays``（地表起爆时间），保证与连线拓扑 + 孔间延时雷管
        + 连线燃烧时长（孔口距离/燃速，数码电子雷管即 ``network.count_line_burn=False``
        时不计）一致；无连线时 ``delays`` 原样透传。

        Args:
            file_path: 目标路径，为空时使用实例中已记录的路径
            burn_velocity: 导爆管燃速 m/s（重算用）；为空时取全局
                ``HoleParam.tube_burn_velocity``（缺 DIMINE_HOME 回退 2000）

        Raises:
            DmoSaveError: 未指定路径或无炮孔数据时抛出
            DmoError: 有连线但燃速非正
        """
        save_path = file_path or self.file_path
        if not save_path:
            raise DmoSaveError("未指定保存路径")

        if not self._holes:
            raise DmoSaveError("无数据可保存")

        save_path = _normalize_dmo_path(save_path)

        if self._network.edges:
            self.recompute_delays(burn_velocity)

        from dimine_python_sdk.lib.native._base import NativeDataTableError

        df = _empty_dmo_df()
        _apply_holes(df, self._holes)
        _apply_network(df, self._network)
        try:
            handle = _build_native_handle(df)
            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)

    @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 摘要信息（基于模型）"""
        return {
            "file": self.file_path or "",
            "hole_count": len(self._holes),
            "edge_count": len(self._network.edges),
            "initiating_hole": self._network.initiating_hole,
            "header_warnings": list(self._header_warnings),
        }

    # ------------------------------------------------------------------
    # 炮孔结构信息（类型化模型）
    # ------------------------------------------------------------------

    @property
    def holes(self) -> List[DmoHole]:
        """炮孔结构信息：每孔一个 :class:`DmoHole`（含装药信息）。

        读取文件时构建并缓存；就地修改后调用 :meth:`save` 生效。
        """
        return self._holes

    @holes.setter
    def holes(self, holes: Iterable[DmoHole]) -> None:
        """写回炮孔结构信息（含装药信息），保存时重建为 DMO 表行。"""
        self._holes = list(holes)

    # ------------------------------------------------------------------
    # 连线信息（类型化模型）
    # ------------------------------------------------------------------

    @property
    def network(self) -> DmoNetwork:
        """连线信息（内存图模型：节点为 DmoHole 对象，边引用节点对象）。

        读取文件时构建并缓存；就地修改（connect / add_hole 等）后调用
        :meth:`save` 生效。
        """
        return self._network

    @network.setter
    def network(self, net: DmoNetwork) -> None:
        """写回连线信息（整体拓扑 + 延时 + 起爆孔），保存时重建为连线二进制。"""
        self._network = net

    # ------------------------------------------------------------------
    # 生命周期管理
    # ------------------------------------------------------------------

    def close(self) -> None:
        """释放模型与文件路径"""
        self._holes = []
        self._network = DmoNetwork()
        self._header_warnings = []
        self.file_path = None

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

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

    def __repr__(self) -> str:
        return (
            f"DmoFile('{self.file_path or ''}', "
            f"holes={len(self._holes)}, edges={len(self._network.edges)})"
        )

header_warnings property

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

holes property writable

炮孔结构信息：每孔一个 :class:DmoHole（含装药信息）。

读取文件时构建并缓存；就地修改后调用 :meth:save 生效。

info property

DMO 摘要信息（基于模型）

network property writable

连线信息（内存图模型：节点为 DmoHole 对象，边引用节点对象）。

读取文件时构建并缓存；就地修改（connect / add_hole 等）后调用 :meth:save 生效。

schema property

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

close()

释放模型与文件路径

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

def close(self) -> None:
    """释放模型与文件路径"""
    self._holes = []
    self._network = DmoNetwork()
    self._header_warnings = []
    self.file_path = None

load(file_path)

读取 .dmo 文件，构建 holes / network 模型并缓存。

DataFrame 仅在此处作为临时物解析后丢弃；读取后 :attr:holes / :attr:network 立即可用。

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 文件，构建 holes / network 模型并缓存。

    DataFrame 仅在此处作为临时物解析后丢弃；读取后 :attr:`holes` /
    :attr:`network` 立即可用。

    Returns:
        self（支持链式调用）

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

    from dimine_python_sdk.lib.native.data_table import NativeFieldType

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

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

recompute_delays(burn_velocity=None)

沿当前连线网络重算 :attr:network.delays（地表起爆时间）。

即「设计过 / 修改过连线信息后重新生成地表起爆时间」的显式入口：自各起爆点 / 根孔向外逐孔传播（父孔 + 孔间延时雷管 + 孔口距离/燃速，数码电子雷管即 network.count_line_burn=False 时不计连线燃烧时长），未达节点保留原值。 无连线时无操作。

Parameters:

Name	Type	Description	Default
burn_velocity	Optional[float]	导爆管燃速 m/s；为空时取 Dimine 全局 HoleParam.tube_burn_velocity（缺 DIMINE_HOME 回退 2000）	None

Returns:

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

Raises:

Type	Description
DmoError	燃速非正

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

def recompute_delays(
    self, burn_velocity: Optional[float] = None
) -> "DmoFile":
    """沿当前连线网络重算 :attr:`network.delays`（地表起爆时间）。

    即「设计过 / 修改过连线信息后重新生成地表起爆时间」的显式入口：自各起爆点 /
    根孔向外逐孔传播（父孔 + 孔间延时雷管 + 孔口距离/燃速，数码电子雷管即
    ``network.count_line_burn=False`` 时不计连线燃烧时长），未达节点保留原值。
    无连线时无操作。

    Args:
        burn_velocity: 导爆管燃速 m/s；为空时取 Dimine 全局
            ``HoleParam.tube_burn_velocity``（缺 DIMINE_HOME 回退 2000）

    Returns:
        self（支持链式调用）

    Raises:
        DmoError: 燃速非正
    """
    if burn_velocity is None:
        burn_velocity = _default_burn_velocity()
    self._network.compute_delays(burn_velocity=float(burn_velocity))
    return self

save(file_path=None, *, burn_velocity=None)

把当前 holes / network 模型保存为 .dmo 文件。

保存时从模型重建 DataFrame（只含标准列），按固定表头还原字段类型； 「孔内分段起爆」列不落入模型，保存时按装药分段与地表起爆时间自动推导。

只要模型持有了连线信息（network.edges 非空，即设计过/修改过连线网络）， 保存前会沿网络重算 delays（地表起爆时间），保证与连线拓扑 + 孔间延时雷管 + 连线燃烧时长（孔口距离/燃速，数码电子雷管即 network.count_line_burn=False 时不计）一致；无连线时 delays 原样透传。

Parameters:

Name	Type	Description	Default
file_path	Optional[Union[str, Path]]	目标路径，为空时使用实例中已记录的路径	None
burn_velocity	Optional[float]	导爆管燃速 m/s（重算用）；为空时取全局 HoleParam.tube_burn_velocity（缺 DIMINE_HOME 回退 2000）	None

Raises:

Type	Description
DmoSaveError	未指定路径或无炮孔数据时抛出
DmoError	有连线但燃速非正

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

def save(
    self,
    file_path: Optional[Union[str, Path]] = None,
    *,
    burn_velocity: Optional[float] = None,
) -> None:
    """把当前 holes / network 模型保存为 .dmo 文件。

    保存时从模型重建 DataFrame（只含标准列），按固定表头还原字段类型；
    「孔内分段起爆」列不落入模型，保存时按装药分段与地表起爆时间自动推导。

    只要模型持有了连线信息（``network.edges`` 非空，即设计过/修改过连线网络），
    保存前会沿网络重算 ``delays``（地表起爆时间），保证与连线拓扑 + 孔间延时雷管
    + 连线燃烧时长（孔口距离/燃速，数码电子雷管即 ``network.count_line_burn=False``
    时不计）一致；无连线时 ``delays`` 原样透传。

    Args:
        file_path: 目标路径，为空时使用实例中已记录的路径
        burn_velocity: 导爆管燃速 m/s（重算用）；为空时取全局
            ``HoleParam.tube_burn_velocity``（缺 DIMINE_HOME 回退 2000）

    Raises:
        DmoSaveError: 未指定路径或无炮孔数据时抛出
        DmoError: 有连线但燃速非正
    """
    save_path = file_path or self.file_path
    if not save_path:
        raise DmoSaveError("未指定保存路径")

    if not self._holes:
        raise DmoSaveError("无数据可保存")

    save_path = _normalize_dmo_path(save_path)

    if self._network.edges:
        self.recompute_delays(burn_velocity)

    from dimine_python_sdk.lib.native._base import NativeDataTableError

    df = _empty_dmo_df()
    _apply_holes(df, self._holes)
    _apply_network(df, self._network)
    try:
        handle = _build_native_handle(df)
        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)

DmoHole

Bases: ParamModel

炮孔结构信息（单孔，DMO 行模型）。

与 DMO 列一一对应，装药信息平铺（不嵌套）：只保留 :class:ChargeResult 的 segments（装药结构），填塞 / 间隔长度等其余字段可由 segments 计算 （见 :meth:stemming_length / :meth:spacer_length）。

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

class DmoHole(ParamModel):
    """炮孔结构信息（单孔，DMO 行模型）。

    与 DMO 列一一对应，装药信息平铺（不嵌套）：只保留 :class:`ChargeResult`
    的 ``segments``（装药结构），填塞 / 间隔长度等其余字段可由 segments 计算
    （见 :meth:`stemming_length` / :meth:`spacer_length`）。
    """

    number: str = Field(..., description="炮孔编号")
    handle: int = Field(default=0, description="炮孔Handle（本地生成递增，同步 GUI 后回填）")
    region: str = Field(default="", description="爆破区域")
    virtual: bool = Field(default=False, description="虚拟炮孔 / 补孔标记（bool；文件里以 是/否 存储）")
    is_initiating: bool = Field(default=False, description="起爆点炮孔（bool；文件里以 是/否 存储）")
    mouth_x: float = Field(default=0.0, description="孔口坐标X")
    mouth_y: float = Field(default=0.0, description="孔口坐标Y")
    mouth_z: float = Field(default=0.0, description="孔口坐标Z")
    bottom_x: float = Field(default=0.0, description="孔底坐标X")
    bottom_y: float = Field(default=0.0, description="孔底坐标Y")
    bottom_z: float = Field(default=0.0, description="孔底坐标Z")
    diameter: float = Field(default=0.0, description="孔径 mm")
    depth: float = Field(default=0.0, description="孔深 m")
    azimuth: float = Field(default=0.0, description="方位角 °")
    inclination: float = Field(default=-90.0, description="倾角 °")
    charge_shrink_ratio: float = Field(
        default=1.0, description="装药半径相对孔径的收缩比例（不耦合装药用，1.0=耦合）"
    )
    pile: int = Field(default=0, description="所属爆堆")
    # ---- 装药信息（平铺）----
    charge_weight: float = Field(default=0.0, description="装药重量 kg")
    charge_length: float = Field(default=0.0, description="装药长度 m")
    specific_charge_area: float = Field(default=0.0, description="单耗面积（负担面积）m²")
    segments: List[ChargeSegment] = Field(
        default_factory=list, description="装药结构分段（自孔口→孔底）"
    )

    @property
    def stemming_length(self) -> float:
        """填塞长度 m（由 segments 计算）"""
        return round(sum(s.length for s in self.segments if s.fill_type == _FILL_STEMMING), 3)

    @property
    def spacer_length(self) -> float:
        """间隔总长度 m（由 segments 计算）"""
        return round(sum(s.length for s in self.segments if s.fill_type == _FILL_SPACER), 3)

spacer_length property

间隔总长度 m（由 segments 计算）

stemming_length property

填塞长度 m（由 segments 计算）

DmoLoadError

Bases: DmoError

加载 .dmo 文件失败

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

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

    pass

DmoNetwork

Bases: ParamModel

连线信息（内存图模型，单一数据源）。

节点为 :class:DmoHole 对象；连线拓扑即 :attr:edges 的键，每条连线的 线属性（延时雷管延时 + 箭头/字高/线宽）即其值 :class:BlastLine——没有 「有边无边属性」的第二份状态，读文件时按「连线信息」列填充、保存时原样 写回，杜绝纯 load→save 把延时雷管改写成时间差。

DMO 时间列语义如下（已用真实 Dimine 文件核对）：

• 「起爆延时」:只有起爆点孔有意义 —— 从按下开关到燃烧到达该起爆点的延时， 其余孔此列恒为 0。各起爆点即各自「地表起爆时间」的锚，保存时取 delays[起爆点] 自动写回，模型无需单独存储。
• 「地表起爆时间」:从按下总开关到燃烧到达该炮孔的延时，逐孔沿连线网络 累积（父孔地表起爆时间 + 孔间延时雷管 + 连线燃烧时长 = 孔口距离/燃速； 数码电子雷管逐孔独立编程、无导爆管传爆，不计连线燃烧时长，见 :attr:count_line_burn）。 模型以 delays（孔号 → 绝对 ms）保存；根孔（各起爆点等无父孔者）没有 父孔，其值即「起爆延时」列，作为传播锚点。只要模型持有了连线信息 （edges 非空，即设计过/修改过连线网络），保存时即沿网络重算 delays（见 :meth:DmoNetwork.compute_delays，燃速缺省取全局 HoleParam.tube_burn_velocity）；仅无连线时 delays 才原样透传。
• 「孔内起爆时间」:= 「地表起爆时间」+ 孔口到第一段炸药段的雷管延时之和 （见 :func:first_explosive_relay）。恒可推导，模型不存储，保存时 由 delays + 装药结构自动计算。
• 「孔内分段起爆」:binary 列，只写不读（模型不存储）。保存时按装药分段
  • 孔口/孔底坐标 + delays 推导每个起爆分段中点的坐标与起爆时间 （见 :func:_delay_hole_records）。

键/值语义：

• nodes / edges 「连线信息」binary 列（解码为对象图）
• edges { (源孔编号, 目标孔编号): BlastLine }， BlastLine.delay = 该连线安装的延时雷管延时 （真实文件里如 17ms），其余字段为显示属性。
• delays 「地表起爆时间」列（孔号 → 绝对 ms；各起爆点另受 「起爆延时」列回填约束，见 :meth:DmoFile.load）。 有连线信息时保存前会被沿网络重算（见 :meth:DmoNetwork.compute_delays）；无连线时透传。
• initiating_hole / initiating_holes 「起爆点炮孔」列。一个起爆网络可有 多个起爆点（导爆管/数码电子均可）：initiating_holes 收录全部（兼容列存编号或「是」标记的写法）， initiating_hole 为兼容旧 API 的主起爆点（= 列表首项）。
• count_line_burn 「是否计入连线燃烧时长」不由文件直接存储：load 时按 「地表起爆时间」列与连线拓扑反推（导爆管含 travel； 数码电子纯 relay，见 :func:_infer_count_line_burn）， 无连线信息时保持默认 True。

兼容性：构造时 nodes 可传编号字符串（自动转为仅含编号的桩节点）； edges 可传 dict（推荐，值 :class:BlastLine），也兼容旧的 (a, b) 编号对列表（自动补默认 :class:BlastLine）。读回后 edges 恒为 dict、键为编号对。

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

class DmoNetwork(ParamModel):
    """连线信息（内存图模型，单一数据源）。

    节点为 :class:`DmoHole` 对象；连线拓扑即 :attr:`edges` 的键，每条连线的
    线属性（延时雷管延时 + 箭头/字高/线宽）即其值 :class:`BlastLine`——没有
    「有边无边属性」的第二份状态，读文件时按「连线信息」列填充、保存时原样
    写回，杜绝纯 load→save 把延时雷管改写成时间差。

    DMO 时间列语义如下（已用真实 Dimine 文件核对）：

      - 「起爆延时」:只有起爆点孔有意义 —— 从按下开关到燃烧到达该起爆点的延时，
        其余孔此列恒为 0。各起爆点即各自「地表起爆时间」的锚，保存时取
        ``delays[起爆点]`` 自动写回，模型无需单独存储。
      - 「地表起爆时间」:从按下总开关到燃烧到达该炮孔的延时，逐孔沿连线网络
        累积（父孔地表起爆时间 + 孔间延时雷管 + 连线燃烧时长 = 孔口距离/燃速；
        数码电子雷管逐孔独立编程、无导爆管传爆，不计连线燃烧时长，见
        :attr:`count_line_burn`）。
        模型以 ``delays``（孔号 → 绝对 ms）保存；根孔（各起爆点等无父孔者）没有
        父孔，其值即「起爆延时」列，作为传播锚点。**只要模型持有了连线信息**
        （``edges`` 非空，即设计过/修改过连线网络），保存时即沿网络重算
        ``delays``（见 :meth:`DmoNetwork.compute_delays`，燃速缺省取全局
        ``HoleParam.tube_burn_velocity``）；仅无连线时 ``delays`` 才原样透传。
      - 「孔内起爆时间」:= 「地表起爆时间」+ 孔口到第一段炸药段的雷管延时之和
        （见 :func:`first_explosive_relay`）。恒可推导，模型**不存储**，保存时
        由 ``delays`` + 装药结构自动计算。
      - 「孔内分段起爆」:binary 列，只写不读（模型不存储）。保存时按装药分段
        + 孔口/孔底坐标 + ``delays`` 推导每个起爆分段中点的坐标与起爆时间
        （见 :func:`_delay_hole_records`）。

    键/值语义：

      - ``nodes`` / ``edges``    「连线信息」binary 列（解码为对象图）
      - ``edges``                ``{ (源孔编号, 目标孔编号): BlastLine }``，
                                 ``BlastLine.delay`` = 该连线安装的延时雷管延时
                                 （真实文件里如 17ms），其余字段为显示属性。
      - ``delays``               「地表起爆时间」列（孔号 → 绝对 ms；各起爆点另受
                                 「起爆延时」列回填约束，见 :meth:`DmoFile.load`）。
                                 有连线信息时保存前会被沿网络重算（见
                                 :meth:`DmoNetwork.compute_delays`）；无连线时透传。
      - ``initiating_hole`` / ``initiating_holes``  「起爆点炮孔」列。一个起爆网络可有
                                 多个起爆点（导爆管/数码电子均可）：``initiating_holes``
                                 收录全部（兼容列存编号或「是」标记的写法），
                                 ``initiating_hole`` 为兼容旧 API 的主起爆点（= 列表首项）。
      - ``count_line_burn``      「是否计入连线燃烧时长」不由文件直接存储：load 时按
                                 「地表起爆时间」列与连线拓扑反推（导爆管含 travel；
                                 数码电子纯 relay，见 :func:`_infer_count_line_burn`），
                                 无连线信息时保持默认 True。

    兼容性：构造时 ``nodes`` 可传编号字符串（自动转为仅含编号的桩节点）；
    ``edges`` 可传 ``dict``（推荐，值 :class:`BlastLine`），也兼容旧的
    ``(a, b)`` 编号对列表（自动补默认 :class:`BlastLine`）。读回后
    ``edges`` 恒为 dict、键为编号对。
    """

    nodes: List[DmoHole] = Field(default_factory=list, description="炮孔节点（DmoHole 对象）")
    edges: Dict[Tuple[str, str], BlastLine] = Field(
        default_factory=dict,
        description=(
            "连线拓扑（单一数据源）：键 = (源孔编号, 目标孔编号)，值 = BlastLine——"
            "该连线的延时雷管延时（delay）+ 箭头/字高/线宽显示属性。读文件由「连线信息」"
            "列解码填充，保存时原样写回。"
        ),
    )
    delays: Dict[str, int] = Field(
        default_factory=dict,
        description=(
            "炮孔编号 -> 地表起爆时间 ms（无连线信息时为权威数据源；"
            "有连线时 save 前沿网络重算，根孔仍锚定此处值）"
        ),
    )
    initiating_hole: str = Field(default="", description="起爆点孔编号（兼容旧 API；多起爆点取第一个）")
    initiating_holes: List[str] = Field(
        default_factory=list,
        description=(
            "起爆点孔编号列表（导爆管/数码电子均可有多个起爆点）。各起爆点即起爆网络的一个"
            "根/首爆点：导爆管各自锚定自身「起爆延时」，数码电子锚 0。"
        ),
    )
    count_line_burn: bool = Field(
        default=True,
        description=(
            "「地表起爆时间」沿连线累积是否计入连线燃烧时长（孔口距离/导爆管燃速）。"
            "导爆管雷管为 True；数码电子雷管逐孔独立编程、无导爆管传爆，为 False。"
        ),
    )

    # ------------------------------------------------------------------
    # 校验与兼容（只收不吐：str 输入 → 桩节点 / 默认 BlastLine）
    # ------------------------------------------------------------------

    @field_validator("nodes", mode="before")
    @classmethod
    def _coerce_nodes(cls, v: Any) -> Any:
        """nodes 兼容 str -> DmoHole(number=...) 桩节点"""
        if not isinstance(v, (list, tuple)):
            return v
        return [DmoHole(number=s) if isinstance(s, str) else s for s in v]

    @field_validator("edges", mode="before")
    @classmethod
    def _coerce_edges(cls, v: Any) -> Any:
        """edges 归一化为 ``{ (源号, 目标号): BlastLine }``。

        兼容三种输入：dict（值 BlastLine 或 None→默认）、旧式 ``(a, b)`` 编号对
        列表（补默认 BlastLine）。键接受编号字符串或 DmoHole 对象。
        """
        def _num(x: Any) -> str:
            return x.number if isinstance(x, DmoHole) else str(x)

        if isinstance(v, dict):
            out: Dict[Tuple[str, str], BlastLine] = {}
            for (a, b), line in v.items():
                na, nb = _num(a), _num(b)
                out[(na, nb)] = line if isinstance(line, BlastLine) else BlastLine()
            return out
        if isinstance(v, (list, tuple)):
            out = {}
            for item in v:
                a, b = item[0], item[1]
                line = item[2] if len(item) > 2 and isinstance(item[2], BlastLine) else None
                out[(_num(a), _num(b))] = line or BlastLine()
            return out
        return v

    @model_validator(mode="after")
    def _normalize_graph(self) -> "DmoNetwork":
        """按编号去重节点、把缺失边端点补成桩节点。

        保证图不变量「每个边端点编号 ∈ nodes（有唯一规范节点对象）」。
        并同步起爆点字段：``initiating_hole``（兼容旧 API 的主起爆点）与
        ``initiating_holes``（全部起爆点，可有多个）互相补齐。
        """
        by_number: Dict[str, DmoHole] = {}
        nodes: List[DmoHole] = []
        for h in self.nodes:
            if h.number in by_number:
                continue
            by_number[h.number] = h
            nodes.append(h)

        for (a, b) in self.edges:
            if a not in by_number:
                ca = DmoHole(number=a)
                by_number[a] = ca
                nodes.append(ca)
            if b not in by_number:
                cb = DmoHole(number=b)
                by_number[b] = cb
                nodes.append(cb)

        self.nodes = nodes

        # 起爆点归一：列表与主起爆点互补；主起爆点恒为列表首项（去重、仅留节点内编号）
        inits = [n for n in self.initiating_holes if n in by_number]
        if self.initiating_hole and self.initiating_hole not in inits:
            inits.insert(0, self.initiating_hole)
        self.initiating_holes = inits
        self.initiating_hole = inits[0] if inits else ""
        return self

    # ------------------------------------------------------------------
    # 图 API
    # ------------------------------------------------------------------

    @property
    def numbers(self) -> List[str]:
        """炮孔编号列表"""
        return [h.number for h in self.nodes]

    @property
    def edge_pairs(self) -> List[Tuple[str, str]]:
        """连线编号对（edges dict 的键，保持插入顺序）"""
        return list(self.edges)

    def get(self, number: str) -> Optional[DmoHole]:
        """按编号取炮孔节点；不存在返回 None"""
        for h in self.nodes:
            if h.number == number:
                return h
        return None

    def has(self, number: str) -> bool:
        """节点是否存在"""
        return self.get(number) is not None

    def __contains__(self, number: str) -> bool:
        return self.has(number)

    def __len__(self) -> int:
        """节点数（注意：不是连线数，连线数用 len(edges)）"""
        return len(self.nodes)

    def __iter__(self):
        """迭代炮孔节点（DmoHole）"""
        return iter(self.nodes)

    def add_hole(self, hole: DmoHole) -> None:
        """添加节点；同编号已存在时替换（upsert）"""
        self.nodes = [h for h in self.nodes if h.number != hole.number]
        self.nodes.append(hole)

    def remove_hole(self, number: str) -> None:
        """移除节点及其关联连线（入边 / 出边一并删除）与起爆点标记"""
        self.nodes = [h for h in self.nodes if h.number != number]
        self.edges = {
            k: v for k, v in self.edges.items()
            if k[0] != number and k[1] != number
        }
        self.initiating_holes = [n for n in self.initiating_holes if n != number]
        self.initiating_hole = self.initiating_holes[0] if self.initiating_holes else ""

    def connect(
        self,
        a: Union[str, DmoHole],
        b: Union[str, DmoHole],
        *,
        line: Optional[BlastLine] = None,
    ) -> None:
        """连接两个节点（接受编号或 DmoHole 对象）；缺节点抛 DmoError。

        ``line`` 为该连线的线属性（延时雷管延时 + 显示属性），缺省给默认
        BlastLine。已存在同向连线时覆盖其线属性。
        """
        na, nb = self._resolve(a), self._resolve(b)
        if na is None or nb is None:
            raise DmoError("connect 的节点不存在，请先 add_hole")
        self.edges[(na.number, nb.number)] = line or BlastLine()

    def disconnect(self, a: Union[str, DmoHole], b: Union[str, DmoHole]) -> None:
        """移除单条连线（按编号匹配，两端可传编号或对象）"""
        na = a.number if isinstance(a, DmoHole) else a
        nb = b.number if isinstance(b, DmoHole) else b
        self.edges.pop((na, nb), None)

    def line_of(self, a: Union[str, DmoHole], b: Union[str, DmoHole]) -> Optional[BlastLine]:
        """某条连线的线属性（BlastLine）；无边返回 None"""
        na = a.number if isinstance(a, DmoHole) else a
        nb = b.number if isinstance(b, DmoHole) else b
        return self.edges.get((na, nb))

    def outgoing(self, number: str) -> List[DmoHole]:
        """以 number 为起点的连线目标节点（出边）"""
        return [
            self.get(t) for (f, t) in self.edges
            if f == number and self.get(t) is not None
        ]

    def incoming(self, number: str) -> List[DmoHole]:
        """指向 number 的连线源节点（入边）"""
        return [
            self.get(f) for (f, t) in self.edges
            if t == number and self.get(f) is not None
        ]

    def neighbors(self, number: str) -> List[DmoHole]:
        """与 number 相连的所有节点（入边 + 出边，不去重）"""
        return self.outgoing(number) + self.incoming(number)

    def degree(self, number: str) -> int:
        """节点度数（入边 + 出边条数）"""
        return len(self.outgoing(number)) + len(self.incoming(number))

    def _resolve(self, ref: Union[str, DmoHole]) -> Optional[DmoHole]:
        """编号 / 对象 -> 规范节点对象（按编号定位，不在图中返回 None）"""
        return self.get(ref.number) if isinstance(ref, DmoHole) else self.get(ref)

    def compute_delays(self, *, burn_velocity: float) -> Dict[str, int]:
        """沿连线网络自根孔向外传播，重算每孔「地表起爆时间」（就地改写 delays）。

        「地表起爆时间」= 从按下总开关到燃烧到达该炮孔的延时，逐孔沿连线网络累积：

           子孔 = 父孔地表起爆时间 + 孔间延时雷管（BlastLine.delay）
                  + 连线燃烧时长（孔口距离 / 燃速）

        数码电子雷管（:attr:`count_line_burn` 为 False）逐孔独立编程、无导爆管
        传爆：**不计连线燃烧时长**，且**无起爆点概念**——首爆点 = 入度 0 的孔
        （可有多个），其地表起爆时间锚定 **0**（不用「起爆延时」/既有 delays 锚），
        沿网络只累加孔间延时雷管：子孔 = 父孔 + 延时雷管。

        用于「改了连线延时雷管再保存」与「新建网络保存」两种场景（触发见
        :meth:`DmoFile.save`，手动刷新见 :meth:`DmoFile.recompute_delays`）。

        导爆管：根孔（无入边节点，及各起爆点）没有父孔，各自锚定 ``delays`` 里已有的
        值（=「起爆延时」列，代表从总开关到该孔的真实起点时刻），无则 0；
        只从锚点向外传播重算可达后代，未达节点（如分离片段）保留原值。连线往返
        燃烧时间相同，边按无向处理即可，不依赖存储方向。

        Args:
            burn_velocity: 导爆管燃速 m/s（> 0；仅 :attr:`count_line_burn` 时要求）

        Returns:
            新 delays 字典（孔号 → 地表起爆时间 ms），已写回 :attr:`delays`
        """
        if burn_velocity <= 0 and self.count_line_burn:
            raise DmoError(f"导爆管燃速必须 > 0: {burn_velocity!r}")
        if not self.nodes or not self.edges:
            return dict(self.delays)

        by_number: Dict[str, DmoHole] = {h.number: h for h in self.nodes}
        in_degree: Dict[str, int] = {}
        adj: Dict[str, List[str]] = {}
        for (src, dst) in self.edges:
            in_degree[dst] = in_degree.get(dst, 0) + 1
            adj.setdefault(src, []).append(dst)
            adj.setdefault(dst, []).append(src)

        # 导爆管：入度为 0 的节点为根；标记的起爆点（可有多个）即便有入边也强制作为传播源。
        # 数码电子雷管无起爆点概念，不加此强制。
        roots = [h.number for h in self.nodes if in_degree.get(h.number, 0) == 0]
        if self.count_line_burn:
            for init in self.initiating_holes or ([self.initiating_hole]
                                                  if self.initiating_hole else []):
                if init in by_number and init not in roots:
                    roots.append(init)
        if not roots:
            return dict(self.delays)

        # 导爆管：根孔锚定 delays 里已有的值（=「起爆延时」列）；无则 0。
        # 数码电子：首爆点（可能有多个）地表起爆时间锚定 0，沿网络只累加延时雷管。
        if self.count_line_burn:
            values: Dict[str, float] = {r: float(self.delays.get(r, 0)) for r in roots}
        else:
            values = {r: 0.0 for r in roots}
        stack: List[str] = list(roots)
        while stack:
            u = stack.pop()
            uh = by_number.get(u)
            base = values[u]
            for v in adj.get(u, ()):
                if v in values:
                    continue
                vh = by_number.get(v)
                line = self.edges.get((u, v)) or self.edges.get((v, u))
                relay = float(line.delay) if line is not None else 0.0
                travel = 0.0
                # 数码电子雷管无导爆管传爆，不计连线燃烧时长
                if self.count_line_burn and uh is not None and vh is not None:
                    dx = uh.mouth_x - vh.mouth_x
                    dy = uh.mouth_y - vh.mouth_y
                    dz = uh.mouth_z - vh.mouth_z
                    travel = (dx * dx + dy * dy + dz * dz) ** 0.5 / burn_velocity * 1000.0
                values[v] = base + relay + travel
                stack.append(v)

        result: Dict[str, int] = {}
        for h in self.nodes:
            result[h.number] = (
                _round_ms(values[h.number]) if h.number in values
                else self.delays.get(h.number, 0)
            )
        self.delays = result
        return result

edge_pairs property

连线编号对（edges dict 的键，保持插入顺序）

numbers property

炮孔编号列表

__iter__()

迭代炮孔节点（DmoHole）

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

def __iter__(self):
    """迭代炮孔节点（DmoHole）"""
    return iter(self.nodes)

__len__()

节点数（注意：不是连线数，连线数用 len(edges)）

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

def __len__(self) -> int:
    """节点数（注意：不是连线数，连线数用 len(edges)）"""
    return len(self.nodes)

add_hole(hole)

添加节点；同编号已存在时替换（upsert）

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

def add_hole(self, hole: DmoHole) -> None:
    """添加节点；同编号已存在时替换（upsert）"""
    self.nodes = [h for h in self.nodes if h.number != hole.number]
    self.nodes.append(hole)

compute_delays(*, burn_velocity)

沿连线网络自根孔向外传播，重算每孔「地表起爆时间」（就地改写 delays）。

「地表起爆时间」= 从按下总开关到燃烧到达该炮孔的延时，逐孔沿连线网络累积：

子孔 = 父孔地表起爆时间 + 孔间延时雷管（BlastLine.delay） + 连线燃烧时长（孔口距离 / 燃速）

数码电子雷管（:attr:count_line_burn 为 False）逐孔独立编程、无导爆管 传爆：不计连线燃烧时长，且无起爆点概念——首爆点 = 入度 0 的孔 （可有多个），其地表起爆时间锚定 0（不用「起爆延时」/既有 delays 锚）， 沿网络只累加孔间延时雷管：子孔 = 父孔 + 延时雷管。

用于「改了连线延时雷管再保存」与「新建网络保存」两种场景（触发见 :meth:DmoFile.save，手动刷新见 :meth:DmoFile.recompute_delays）。

导爆管：根孔（无入边节点，及各起爆点）没有父孔，各自锚定 delays 里已有的 值（=「起爆延时」列，代表从总开关到该孔的真实起点时刻），无则 0； 只从锚点向外传播重算可达后代，未达节点（如分离片段）保留原值。连线往返 燃烧时间相同，边按无向处理即可，不依赖存储方向。

Parameters:

Name	Type	Description	Default
burn_velocity	float	导爆管燃速 m/s（> 0；仅 :attr:count_line_burn 时要求）	required

Returns:

Type	Description
Dict[str, int]	新 delays 字典（孔号 → 地表起爆时间 ms），已写回 :attr:delays

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

def compute_delays(self, *, burn_velocity: float) -> Dict[str, int]:
    """沿连线网络自根孔向外传播，重算每孔「地表起爆时间」（就地改写 delays）。

    「地表起爆时间」= 从按下总开关到燃烧到达该炮孔的延时，逐孔沿连线网络累积：

       子孔 = 父孔地表起爆时间 + 孔间延时雷管（BlastLine.delay）
              + 连线燃烧时长（孔口距离 / 燃速）

    数码电子雷管（:attr:`count_line_burn` 为 False）逐孔独立编程、无导爆管
    传爆：**不计连线燃烧时长**，且**无起爆点概念**——首爆点 = 入度 0 的孔
    （可有多个），其地表起爆时间锚定 **0**（不用「起爆延时」/既有 delays 锚），
    沿网络只累加孔间延时雷管：子孔 = 父孔 + 延时雷管。

    用于「改了连线延时雷管再保存」与「新建网络保存」两种场景（触发见
    :meth:`DmoFile.save`，手动刷新见 :meth:`DmoFile.recompute_delays`）。

    导爆管：根孔（无入边节点，及各起爆点）没有父孔，各自锚定 ``delays`` 里已有的
    值（=「起爆延时」列，代表从总开关到该孔的真实起点时刻），无则 0；
    只从锚点向外传播重算可达后代，未达节点（如分离片段）保留原值。连线往返
    燃烧时间相同，边按无向处理即可，不依赖存储方向。

    Args:
        burn_velocity: 导爆管燃速 m/s（> 0；仅 :attr:`count_line_burn` 时要求）

    Returns:
        新 delays 字典（孔号 → 地表起爆时间 ms），已写回 :attr:`delays`
    """
    if burn_velocity <= 0 and self.count_line_burn:
        raise DmoError(f"导爆管燃速必须 > 0: {burn_velocity!r}")
    if not self.nodes or not self.edges:
        return dict(self.delays)

    by_number: Dict[str, DmoHole] = {h.number: h for h in self.nodes}
    in_degree: Dict[str, int] = {}
    adj: Dict[str, List[str]] = {}
    for (src, dst) in self.edges:
        in_degree[dst] = in_degree.get(dst, 0) + 1
        adj.setdefault(src, []).append(dst)
        adj.setdefault(dst, []).append(src)

    # 导爆管：入度为 0 的节点为根；标记的起爆点（可有多个）即便有入边也强制作为传播源。
    # 数码电子雷管无起爆点概念，不加此强制。
    roots = [h.number for h in self.nodes if in_degree.get(h.number, 0) == 0]
    if self.count_line_burn:
        for init in self.initiating_holes or ([self.initiating_hole]
                                              if self.initiating_hole else []):
            if init in by_number and init not in roots:
                roots.append(init)
    if not roots:
        return dict(self.delays)

    # 导爆管：根孔锚定 delays 里已有的值（=「起爆延时」列）；无则 0。
    # 数码电子：首爆点（可能有多个）地表起爆时间锚定 0，沿网络只累加延时雷管。
    if self.count_line_burn:
        values: Dict[str, float] = {r: float(self.delays.get(r, 0)) for r in roots}
    else:
        values = {r: 0.0 for r in roots}
    stack: List[str] = list(roots)
    while stack:
        u = stack.pop()
        uh = by_number.get(u)
        base = values[u]
        for v in adj.get(u, ()):
            if v in values:
                continue
            vh = by_number.get(v)
            line = self.edges.get((u, v)) or self.edges.get((v, u))
            relay = float(line.delay) if line is not None else 0.0
            travel = 0.0
            # 数码电子雷管无导爆管传爆，不计连线燃烧时长
            if self.count_line_burn and uh is not None and vh is not None:
                dx = uh.mouth_x - vh.mouth_x
                dy = uh.mouth_y - vh.mouth_y
                dz = uh.mouth_z - vh.mouth_z
                travel = (dx * dx + dy * dy + dz * dz) ** 0.5 / burn_velocity * 1000.0
            values[v] = base + relay + travel
            stack.append(v)

    result: Dict[str, int] = {}
    for h in self.nodes:
        result[h.number] = (
            _round_ms(values[h.number]) if h.number in values
            else self.delays.get(h.number, 0)
        )
    self.delays = result
    return result

connect(a, b, *, line=None)

连接两个节点（接受编号或 DmoHole 对象）；缺节点抛 DmoError。

line 为该连线的线属性（延时雷管延时 + 显示属性），缺省给默认 BlastLine。已存在同向连线时覆盖其线属性。

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

def connect(
    self,
    a: Union[str, DmoHole],
    b: Union[str, DmoHole],
    *,
    line: Optional[BlastLine] = None,
) -> None:
    """连接两个节点（接受编号或 DmoHole 对象）；缺节点抛 DmoError。

    ``line`` 为该连线的线属性（延时雷管延时 + 显示属性），缺省给默认
    BlastLine。已存在同向连线时覆盖其线属性。
    """
    na, nb = self._resolve(a), self._resolve(b)
    if na is None or nb is None:
        raise DmoError("connect 的节点不存在，请先 add_hole")
    self.edges[(na.number, nb.number)] = line or BlastLine()

degree(number)

节点度数（入边 + 出边条数）

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

def degree(self, number: str) -> int:
    """节点度数（入边 + 出边条数）"""
    return len(self.outgoing(number)) + len(self.incoming(number))

disconnect(a, b)

移除单条连线（按编号匹配，两端可传编号或对象）

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

def disconnect(self, a: Union[str, DmoHole], b: Union[str, DmoHole]) -> None:
    """移除单条连线（按编号匹配，两端可传编号或对象）"""
    na = a.number if isinstance(a, DmoHole) else a
    nb = b.number if isinstance(b, DmoHole) else b
    self.edges.pop((na, nb), None)

get(number)

按编号取炮孔节点；不存在返回 None

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

def get(self, number: str) -> Optional[DmoHole]:
    """按编号取炮孔节点；不存在返回 None"""
    for h in self.nodes:
        if h.number == number:
            return h
    return None

has(number)

节点是否存在

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

def has(self, number: str) -> bool:
    """节点是否存在"""
    return self.get(number) is not None

incoming(number)

指向 number 的连线源节点（入边）

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

def incoming(self, number: str) -> List[DmoHole]:
    """指向 number 的连线源节点（入边）"""
    return [
        self.get(f) for (f, t) in self.edges
        if t == number and self.get(f) is not None
    ]

line_of(a, b)

某条连线的线属性（BlastLine）；无边返回 None

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

def line_of(self, a: Union[str, DmoHole], b: Union[str, DmoHole]) -> Optional[BlastLine]:
    """某条连线的线属性（BlastLine）；无边返回 None"""
    na = a.number if isinstance(a, DmoHole) else a
    nb = b.number if isinstance(b, DmoHole) else b
    return self.edges.get((na, nb))

neighbors(number)

与 number 相连的所有节点（入边 + 出边，不去重）

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

def neighbors(self, number: str) -> List[DmoHole]:
    """与 number 相连的所有节点（入边 + 出边，不去重）"""
    return self.outgoing(number) + self.incoming(number)

outgoing(number)

以 number 为起点的连线目标节点（出边）

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

def outgoing(self, number: str) -> List[DmoHole]:
    """以 number 为起点的连线目标节点（出边）"""
    return [
        self.get(t) for (f, t) in self.edges
        if f == number and self.get(t) is not None
    ]

remove_hole(number)

移除节点及其关联连线（入边 / 出边一并删除）与起爆点标记

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

def remove_hole(self, number: str) -> None:
    """移除节点及其关联连线（入边 / 出边一并删除）与起爆点标记"""
    self.nodes = [h for h in self.nodes if h.number != number]
    self.edges = {
        k: v for k, v in self.edges.items()
        if k[0] != number and k[1] != number
    }
    self.initiating_holes = [n for n in self.initiating_holes if n != number]
    self.initiating_hole = self.initiating_holes[0] if self.initiating_holes else ""

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.network.edges)   # 连线拓扑（读取后立即可用）

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.network.edges)   # 连线拓扑（读取后立即可用）
    """
    dmo = DmoFile(file_path)
    try:
        yield dmo
    finally:
        dmo.close()

first_explosive_relay(segments)

孔口→孔底走到第一段炸药段的雷管延时之和（含该段）ms。

惰性分段（填塞物 / 间隔器 / 空气）不参与起爆，其雷管延时（真实文件为 -1）不计入；第一段炸药之上的起爆药包等先爆载体若带正延时一并累计； 遇到第一段炸药段即停止。用于推导「孔内起爆时间」（= 地表起爆时间 + 本函数返回值）。无炸药段返回 0。

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

def first_explosive_relay(segments: Iterable[ChargeSegment]) -> int:
    """孔口→孔底走到第一段炸药段的雷管延时之和（含该段）ms。

    惰性分段（填塞物 / 间隔器 / 空气）不参与起爆，其雷管延时（真实文件为
    -1）不计入；第一段炸药之上的起爆药包等先爆载体若带正延时一并累计；
    遇到第一段炸药段即停止。用于推导「孔内起爆时间」（= 地表起爆时间 +
    本函数返回值）。无炸药段返回 0。
    """
    total = 0
    for seg in segments:
        d = int(seg.delay or 0)
        if d > 0:
            total += d
        if seg.fill_type not in _INERT_FILLS and seg.fill_type != _FILL_PRIMER:
            break  # 遇到第一段炸药段（含真实炸药名如「2#岩石炸药」）
    return total

read_charge_structures(source, *, hole_number_col='炮孔编号', column='装药结构')

读取 DMO 数据表「装药结构」列，反序列化每个炮孔的装药分段。

.. note:: 内部过渡 API，不对外公开；请改用 :attr:DmoFile.holes（类型化模型）。

Parameters:

Name	Type	Description	Default
source	Any	pd.DataFrame / DmoFile（有 .data）/ 文件路径	required
hole_number_col	str	炮孔编号列名（输出行标识）	'炮孔编号'
column	str	「装药结构」列名	'装药结构'

Returns:

Type	Description
'pd.DataFrame'	每孔一行的 DataFrame：炮孔编号、段数、装药总长度、装药结构(list[ChargeSegment])

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

def read_charge_structures(
    source: Any,
    *,
    hole_number_col: str = "炮孔编号",
    column: str = "装药结构",
) -> "pd.DataFrame":
    """读取 DMO 数据表「装药结构」列，反序列化每个炮孔的装药分段。

    .. note::
        内部过渡 API，不对外公开；请改用 :attr:`DmoFile.holes`（类型化模型）。

    Args:
        source: pd.DataFrame / DmoFile（有 ``.data``）/ 文件路径
        hole_number_col: 炮孔编号列名（输出行标识）
        column: 「装药结构」列名

    Returns:
        每孔一行的 DataFrame：炮孔编号、段数、装药总长度、装药结构(list[ChargeSegment])
    """
    import pandas as pd

    df = _as_dataframe(source, column)
    records = []
    for _, row in df.iterrows():
        raw = row[column]
        segments = [] if pd.isna(raw) else _decode_charge_structure(raw)
        records.append({
            hole_number_col: row[hole_number_col] if hole_number_col in df.columns else None,
            "段数": len(segments),
            "装药总长度": round(sum(s.length for s in segments), 3),
            column: segments,
        })
    return pd.DataFrame(records)

write_charge_structures(source, structures, *, hole_number_col='炮孔编号', column='装药结构')

将装药分段编码为原始 bytes 写回 DMO 数据表「装药结构」列。

structures 三种形态：

• 单个 list[ChargeSegment]：所有炮孔共用（覆盖「大部分相似」场景）
• dict[炮孔编号 -> 分段列表]：按孔号匹配，缺孔写空结构
• 按行顺序的 list[list[ChargeSegment]]：长度必须与行数一致

.. note:: 内部过渡 API，不对外公开；请改用 :attr:DmoFile.holes（类型化模型）。

Parameters:

Name	Type	Description	Default
source	Any	DmoFile（有 .data）或 pd.DataFrame；DmoFile 后续由调用方 .save()	required
structures	Union[Dict[str, List[ChargeSegment]], List[List[ChargeSegment]], List[ChargeSegment]]	装药分段（见上）	required
hole_number_col	str	炮孔编号列名	'炮孔编号'
column	str	「装药结构」列名	'装药结构'

Raises:

Type	Description
DmoChargeStructureError	缺列 / structures 形态不合法 / 长度不匹配

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

def write_charge_structures(
    source: Any,
    structures: Union[Dict[str, List[ChargeSegment]], List[List[ChargeSegment]], List[ChargeSegment]],
    *,
    hole_number_col: str = "炮孔编号",
    column: str = "装药结构",
) -> None:
    """将装药分段编码为原始 ``bytes`` 写回 DMO 数据表「装药结构」列。

    ``structures`` 三种形态：

      - 单个 ``list[ChargeSegment]``：所有炮孔共用（覆盖「大部分相似」场景）
      - ``dict[炮孔编号 -> 分段列表]``：按孔号匹配，缺孔写空结构
      - 按行顺序的 ``list[list[ChargeSegment]]``：长度必须与行数一致

    .. note::
        内部过渡 API，不对外公开；请改用 :attr:`DmoFile.holes`（类型化模型）。

    Args:
        source: DmoFile（有 ``.data``）或 pd.DataFrame；DmoFile 后续由调用方 ``.save()``
        structures: 装药分段（见上）
        hole_number_col: 炮孔编号列名
        column: 「装药结构」列名

    Raises:
        DmoChargeStructureError: 缺列 / structures 形态不合法 / 长度不匹配
    """
    df = _as_dataframe(source, column)
    if hole_number_col not in df.columns:
        raise DmoChargeStructureError(f"数据表缺少炮孔编号列: {hole_number_col}")

    # 归一化 structures → 行 -> 分段
    if isinstance(structures, dict):
        by_hole = structures

        def pick(row, pos):  # noqa: ANN001
            return by_hole.get(row[hole_number_col], [])

    elif len(structures) > 0 and _is_segment(structures[0]):
        template = list(structures)

        def pick(row, pos):  # noqa: ANN001
            return template

    else:
        aligned = list(structures)
        if len(aligned) != len(df):
            raise DmoChargeStructureError(
                f"按行对齐的 structures 长度 {len(aligned)} 与行数 {len(df)} 不一致"
            )

        def pick(row, pos):  # noqa: ANN001
            return aligned[pos]

    for pos, (_, row) in enumerate(df.iterrows()):
        # 写入原始 bytes（binary 字段经 Set_Value(bytes) 原样存储），而非 hex 字符串
        df.at[row.name, column] = _encode_charge_structure(pick(row, pos))