datatable

数据表（DmDataTable / DmDataTableRecord）与行代理（TableRow）封装

将原本散落在 lib.types.entity 与 lib.types._row 中的数据表业务类迁移至此， 以厘清 types 与 db 模块边界： types 仅承载基础几何与实体类型，db 承载数据库/数据表等业务封装。

DmDataTable

数据表 Python 二次封装类

Source code in dimine_python_sdk\lib\db\datatable.py

class DmDataTable:
    """
    数据表 Python 二次封装类
    """

    def __init__(self, file_path):
        self._obj = Dm.CDataTable()
        self.load(file_path)

    def load(self, file_path: str) -> bool | int:
        """
        加载数据表文件
        :param file_path: UTF-8 编码的数据表文件路径
        :return: 加载结果（成功返回 True/0，失败返回 False/错误码）

        example:
        ```python
        dt = DmDataTable()
        dt.load("E:/dassistant/tests/123.dmd")
        print("字段总数:",dt.get_field_count())
        ```
        """
        if not isinstance(file_path, str) or not file_path.strip():
            raise ValueError("文件路径不能为空字符串")
        return self._obj.Load(file_path)

    def get_field_count(self) -> int:
        """获取字段总数"""
        return self._obj.Get_Field_Count()

    def get_record_count(self) -> int:
        """获取记录总数"""
        return self._obj.Get_Record_Count()

    def get_record_data(self, record_index: int) -> dict:
        """
        通过索引获取单条记录的完整数据
        :param record_index: 记录索引（从0开始）
        :return: 记录数据字典，键为字段名，值为字段值
        """
        return self._obj.Get_Record_Data(record_index)

    def get_field_id(self, field_name: str) -> int:
        """
        通过字段名获取字段ID
        :param field_name: UTF-8 编码的字段名
        :return: 字段ID
        """
        if not isinstance(field_name, str) or not field_name.strip():
            raise ValueError("字段名不能为空字符串")
        return self._obj.Get_Field_Id(field_name)

    def get_field_name(self, field_index: int) -> str:
        """
        通过索引获取字段名
        :param field_index: 字段索引（从0开始）
        :return: UTF-8 编码的字段名
        """
        return self._obj.Get_Field_Name(field_index)

    def get_field_display_name(self, field_index: int) -> str:
        """
        通过索引获取字段显示名
        :param field_index: 字段索引（从0开始）
        :return: UTF-8 编码的字段显示名
        """
        if field_index > self.get_field_count() - 1:
            raise IndexError("字段索引超出范围")
        return self._obj.Get_Field_DisplayName(field_index)

    def get_field_type(self, field_index: int) -> int:
        """
        通过索引获取字段类型
        :param field_index: 字段索引（从0开始）
        :return: 字段类型枚举值
        """
        if field_index > self.get_field_count() - 1:
            raise IndexError("字段索引超出范围")
        return self._obj.Get_Field_Type(field_index)

    def get_record_from_index(self, record_index: int) -> DmDataTableRecord:
        """
        通过索引获取原始记录对象
        :param record_index: 记录索引（从0开始）
        :return: 原始记录对象
        """
        if record_index > self.get_record_count() - 1:
            raise IndexError("记录索引超出范围")
        cpp_result = self._obj.Get_Record_From_Index(record_index)
        return DmDataTableRecord._from_obj(cpp_result)

    def get_record_count_by_condition(self, field_name: str, field_value: str) -> int:
        """
        根据字段名和值筛选记录数量
        :param field_name: UTF-8 编码的字段名
        :param field_value: UTF-8 编码的字段值
        :return: 符合条件的记录数量
        """
        if not isinstance(field_name, str) or not field_name.strip():
            raise ValueError("字段名不能为空字符串")
        if not isinstance(field_value, str):
            raise ValueError("字段值必须为字符串")
        return self._obj.GetRecordCountByCondition(field_name, field_value)

    def add_record(self) -> bool | int:
        """添加一条新记录"""
        return self._obj.Add_Record()

    def save(self) -> bool | int:
        """保存数据表"""
        return self._obj.Save()

    # ---------------------- DataFrame 化扩展 ----------------------

    def __len__(self) -> int:
        """返回记录总数"""
        return self.get_record_count()

    def __getitem__(self, key: Union[str, int, slice]):
        """
        支持列访问、行访问与切片

        - str   : 按字段名取整列  -> list[Any]
        - int   : 按索引取单行    -> TableRow
        - slice : 切片取子集      -> list[TableRow]
        """
        if isinstance(key, str):
            return [row[key] for row in self]
        if isinstance(key, int):
            if key < 0:
                key += len(self)
            if key < 0 or key >= len(self):
                raise IndexError(f"记录索引 {key} 超出范围 [0, {len(self)})")
            return self._get_row(key)
        if isinstance(key, slice):
            return [self[i] for i in range(*key.indices(len(self)))]
        raise TypeError(f"不支持 key 类型: {type(key)}")

    def __iter__(self):
        """逐行迭代，每次返回 TableRow"""
        for i in range(len(self)):
            yield self[i]

    def __repr__(self) -> str:
        return (
            f"DmDataTable(records={len(self)}, fields={self.get_field_count()}, "
            f"columns={self.columns!r})"
        )

    def _get_row(self, index: int) -> TableRow:
        """通过索引获取 TableRow 行代理"""
        data = self.get_record_data(index)
        return TableRow(data, index)

    @property
    def columns(self) -> list[str]:
        """返回所有字段名列表"""
        return [self.get_field_name(i) for i in range(self.get_field_count())]

    def to_pandas(self) -> "pd.DataFrame":
        """
        转换为 pandas DataFrame

        需要安装 pandas: uv add pandas
        """
        try:
            import pandas as pd
        except ImportError as exc:
            raise ImportError("使用 to_pandas() 需要安装 pandas: uv add pandas") from exc
        return pd.DataFrame([row.to_dict() for row in self])

    def to_numpy(self, column: Optional[str] = None):
        """
        转换为 numpy 数组

        Args:
            column: 指定字段名时返回该列的一维数组；
                    为 None 时返回二维数组（所有记录的所有字段值）
        """
        import numpy as np
        if column is not None:
            return np.array(self[column])
        return np.array([list(row.values()) for row in self])

    @classmethod
    def _from_obj(cls, obj):
        """
        原生dmDataTable对象创建封装实例
        :param obj: 原生dmDataTable对象（bind11封装后）
        :return: DmDataTable实例
        """
        instance = cls.__new__(cls)
        instance._obj = obj
        return instance

columns property

返回所有字段名列表

__getitem__(key)

支持列访问、行访问与切片

• str : 按字段名取整列 -> list[Any]
• int : 按索引取单行 -> TableRow
• slice : 切片取子集 -> list[TableRow]

Source code in dimine_python_sdk\lib\db\datatable.py

def __getitem__(self, key: Union[str, int, slice]):
    """
    支持列访问、行访问与切片

    - str   : 按字段名取整列  -> list[Any]
    - int   : 按索引取单行    -> TableRow
    - slice : 切片取子集      -> list[TableRow]
    """
    if isinstance(key, str):
        return [row[key] for row in self]
    if isinstance(key, int):
        if key < 0:
            key += len(self)
        if key < 0 or key >= len(self):
            raise IndexError(f"记录索引 {key} 超出范围 [0, {len(self)})")
        return self._get_row(key)
    if isinstance(key, slice):
        return [self[i] for i in range(*key.indices(len(self)))]
    raise TypeError(f"不支持 key 类型: {type(key)}")

__iter__()

逐行迭代，每次返回 TableRow

Source code in dimine_python_sdk\lib\db\datatable.py

def __iter__(self):
    """逐行迭代，每次返回 TableRow"""
    for i in range(len(self)):
        yield self[i]

__len__()

返回记录总数

Source code in dimine_python_sdk\lib\db\datatable.py

def __len__(self) -> int:
    """返回记录总数"""
    return self.get_record_count()

add_record()

添加一条新记录

Source code in dimine_python_sdk\lib\db\datatable.py

def add_record(self) -> bool | int:
    """添加一条新记录"""
    return self._obj.Add_Record()

get_field_count()

获取字段总数

Source code in dimine_python_sdk\lib\db\datatable.py

def get_field_count(self) -> int:
    """获取字段总数"""
    return self._obj.Get_Field_Count()

get_field_display_name(field_index)

通过索引获取字段显示名 :param field_index: 字段索引（从0开始） :return: UTF-8 编码的字段显示名

Source code in dimine_python_sdk\lib\db\datatable.py

def get_field_display_name(self, field_index: int) -> str:
    """
    通过索引获取字段显示名
    :param field_index: 字段索引（从0开始）
    :return: UTF-8 编码的字段显示名
    """
    if field_index > self.get_field_count() - 1:
        raise IndexError("字段索引超出范围")
    return self._obj.Get_Field_DisplayName(field_index)

get_field_id(field_name)

通过字段名获取字段ID :param field_name: UTF-8 编码的字段名 :return: 字段ID

Source code in dimine_python_sdk\lib\db\datatable.py

def get_field_id(self, field_name: str) -> int:
    """
    通过字段名获取字段ID
    :param field_name: UTF-8 编码的字段名
    :return: 字段ID
    """
    if not isinstance(field_name, str) or not field_name.strip():
        raise ValueError("字段名不能为空字符串")
    return self._obj.Get_Field_Id(field_name)

get_field_name(field_index)

通过索引获取字段名 :param field_index: 字段索引（从0开始） :return: UTF-8 编码的字段名

Source code in dimine_python_sdk\lib\db\datatable.py

def get_field_name(self, field_index: int) -> str:
    """
    通过索引获取字段名
    :param field_index: 字段索引（从0开始）
    :return: UTF-8 编码的字段名
    """
    return self._obj.Get_Field_Name(field_index)

get_field_type(field_index)

通过索引获取字段类型 :param field_index: 字段索引（从0开始） :return: 字段类型枚举值

Source code in dimine_python_sdk\lib\db\datatable.py

def get_field_type(self, field_index: int) -> int:
    """
    通过索引获取字段类型
    :param field_index: 字段索引（从0开始）
    :return: 字段类型枚举值
    """
    if field_index > self.get_field_count() - 1:
        raise IndexError("字段索引超出范围")
    return self._obj.Get_Field_Type(field_index)

get_record_count()

获取记录总数

Source code in dimine_python_sdk\lib\db\datatable.py

def get_record_count(self) -> int:
    """获取记录总数"""
    return self._obj.Get_Record_Count()

get_record_count_by_condition(field_name, field_value)

根据字段名和值筛选记录数量 :param field_name: UTF-8 编码的字段名 :param field_value: UTF-8 编码的字段值 :return: 符合条件的记录数量

Source code in dimine_python_sdk\lib\db\datatable.py

def get_record_count_by_condition(self, field_name: str, field_value: str) -> int:
    """
    根据字段名和值筛选记录数量
    :param field_name: UTF-8 编码的字段名
    :param field_value: UTF-8 编码的字段值
    :return: 符合条件的记录数量
    """
    if not isinstance(field_name, str) or not field_name.strip():
        raise ValueError("字段名不能为空字符串")
    if not isinstance(field_value, str):
        raise ValueError("字段值必须为字符串")
    return self._obj.GetRecordCountByCondition(field_name, field_value)

get_record_data(record_index)

通过索引获取单条记录的完整数据 :param record_index: 记录索引（从0开始） :return: 记录数据字典，键为字段名，值为字段值

Source code in dimine_python_sdk\lib\db\datatable.py

def get_record_data(self, record_index: int) -> dict:
    """
    通过索引获取单条记录的完整数据
    :param record_index: 记录索引（从0开始）
    :return: 记录数据字典，键为字段名，值为字段值
    """
    return self._obj.Get_Record_Data(record_index)

get_record_from_index(record_index)

通过索引获取原始记录对象 :param record_index: 记录索引（从0开始） :return: 原始记录对象

Source code in dimine_python_sdk\lib\db\datatable.py

def get_record_from_index(self, record_index: int) -> DmDataTableRecord:
    """
    通过索引获取原始记录对象
    :param record_index: 记录索引（从0开始）
    :return: 原始记录对象
    """
    if record_index > self.get_record_count() - 1:
        raise IndexError("记录索引超出范围")
    cpp_result = self._obj.Get_Record_From_Index(record_index)
    return DmDataTableRecord._from_obj(cpp_result)

load(file_path)

加载数据表文件 :param file_path: UTF-8 编码的数据表文件路径 :return: 加载结果（成功返回 True/0，失败返回 False/错误码）

example:

dt = DmDataTable()
dt.load("E:/dassistant/tests/123.dmd")
print("字段总数:",dt.get_field_count())

Source code in dimine_python_sdk\lib\db\datatable.py

def load(self, file_path: str) -> bool | int:
    """
    加载数据表文件
    :param file_path: UTF-8 编码的数据表文件路径
    :return: 加载结果（成功返回 True/0，失败返回 False/错误码）

    example:
    ```python
    dt = DmDataTable()
    dt.load("E:/dassistant/tests/123.dmd")
    print("字段总数:",dt.get_field_count())
    ```
    """
    if not isinstance(file_path, str) or not file_path.strip():
        raise ValueError("文件路径不能为空字符串")
    return self._obj.Load(file_path)

save()

保存数据表

Source code in dimine_python_sdk\lib\db\datatable.py

def save(self) -> bool | int:
    """保存数据表"""
    return self._obj.Save()

to_numpy(column=None)

转换为 numpy 数组

Parameters:

Name	Type	Description	Default
column	Optional[str]	指定字段名时返回该列的一维数组； 为 None 时返回二维数组（所有记录的所有字段值）	None

Source code in dimine_python_sdk\lib\db\datatable.py

def to_numpy(self, column: Optional[str] = None):
    """
    转换为 numpy 数组

    Args:
        column: 指定字段名时返回该列的一维数组；
                为 None 时返回二维数组（所有记录的所有字段值）
    """
    import numpy as np
    if column is not None:
        return np.array(self[column])
    return np.array([list(row.values()) for row in self])

to_pandas()

转换为 pandas DataFrame

需要安装 pandas: uv add pandas

Source code in dimine_python_sdk\lib\db\datatable.py

def to_pandas(self) -> "pd.DataFrame":
    """
    转换为 pandas DataFrame

    需要安装 pandas: uv add pandas
    """
    try:
        import pandas as pd
    except ImportError as exc:
        raise ImportError("使用 to_pandas() 需要安装 pandas: uv add pandas") from exc
    return pd.DataFrame([row.to_dict() for row in self])

DmDataTableRecord

数据表记录类

Source code in dimine_python_sdk\lib\db\datatable.py

class DmDataTableRecord:
    """
    数据表记录类
    """

    @classmethod
    def _from_obj(cls, obj):
        """
        原生dmDataTableRecord对象创建封装实例
        :param obj: 原生dmDataTableRecord对象（bind11封装后）
        :return: DmDataTableRecord实例
        """
        instance = cls.__new__(cls)
        instance._obj = obj
        return instance

    def as_string(self, field_name: str) -> str:
        """
        获取字段值（字符串形式）
        :param field_name: UTF-8 编码的字段名
        :return: 字段值字符串
        """
        if not isinstance(field_name, str) or not field_name.strip():
            raise ValueError("字段名不能为空字符串")
        return self._obj.asString(field_name)

    def as_double(self, field_name: str) -> float:
        """
        获取字段值（浮点形式）
        :param field_name: UTF-8 编码的字段名
        :return: 字段值浮点数
        """
        if not isinstance(field_name, str) or not field_name.strip():
            raise ValueError("字段名不能为空字符串")
        return self._obj.asDouble(field_name)

    def set_value(self, field_name: str, value: str | float) -> None:
        """
        设置字段值（支持字符串/浮点数）
        :param field_name: UTF-8 编码的字段名
        :param value: 字段值（字符串或浮点数）
        """
        if not isinstance(field_name, str) or not field_name.strip():
            raise ValueError("字段名不能为空字符串")
        if isinstance(value, str):
            self._obj.Set_Value(field_name, value)
        elif isinstance(value, (int, float)):
            self._obj.Set_Value(field_name, float(value))
        else:
            raise TypeError("字段值仅支持字符串或数字类型")

as_double(field_name)

获取字段值（浮点形式） :param field_name: UTF-8 编码的字段名 :return: 字段值浮点数

Source code in dimine_python_sdk\lib\db\datatable.py

def as_double(self, field_name: str) -> float:
    """
    获取字段值（浮点形式）
    :param field_name: UTF-8 编码的字段名
    :return: 字段值浮点数
    """
    if not isinstance(field_name, str) or not field_name.strip():
        raise ValueError("字段名不能为空字符串")
    return self._obj.asDouble(field_name)

as_string(field_name)

获取字段值（字符串形式） :param field_name: UTF-8 编码的字段名 :return: 字段值字符串

Source code in dimine_python_sdk\lib\db\datatable.py

def as_string(self, field_name: str) -> str:
    """
    获取字段值（字符串形式）
    :param field_name: UTF-8 编码的字段名
    :return: 字段值字符串
    """
    if not isinstance(field_name, str) or not field_name.strip():
        raise ValueError("字段名不能为空字符串")
    return self._obj.asString(field_name)

set_value(field_name, value)

设置字段值（支持字符串/浮点数） :param field_name: UTF-8 编码的字段名 :param value: 字段值（字符串或浮点数）

Source code in dimine_python_sdk\lib\db\datatable.py

def set_value(self, field_name: str, value: str | float) -> None:
    """
    设置字段值（支持字符串/浮点数）
    :param field_name: UTF-8 编码的字段名
    :param value: 字段值（字符串或浮点数）
    """
    if not isinstance(field_name, str) or not field_name.strip():
        raise ValueError("字段名不能为空字符串")
    if isinstance(value, str):
        self._obj.Set_Value(field_name, value)
    elif isinstance(value, (int, float)):
        self._obj.Set_Value(field_name, float(value))
    else:
        raise TypeError("字段值仅支持字符串或数字类型")

TableRow

轻量级行数据代理，支持 dict 式与属性式双访问

Example

row = TableRow({"HoleID": "ZK001", "Depth": 500.0}, index=0) row["HoleID"] 'ZK001' row.Depth 500.0 row.index 0

Source code in dimine_python_sdk\lib\db\datatable.py

class TableRow:
    """
    轻量级行数据代理，支持 dict 式与属性式双访问

    Example:
        >>> row = TableRow({"HoleID": "ZK001", "Depth": 500.0}, index=0)
        >>> row["HoleID"]
        'ZK001'
        >>> row.Depth
        500.0
        >>> row.index
        0
    """

    def __init__(self, data: dict[str, Any], index: int):
        self._data = data
        self._index = index

    @property
    def index(self) -> int:
        """行索引"""
        return self._index

    def __getitem__(self, key: str) -> Any:
        return self._data[key]

    def __getattr__(self, name: str) -> Any:
        if name.startswith("_"):
            raise AttributeError(f"'{type(self).__name__}' 对象没有属性 '{name}'")
        try:
            return self._data[name]
        except KeyError:
            raise AttributeError(f"'{type(self).__name__}' 对象没有字段 '{name}'")

    def __repr__(self) -> str:
        items = ", ".join(f"{k}={v!r}" for k, v in list(self._data.items())[:5])
        if len(self._data) > 5:
            items += ", ..."
        return f"TableRow(index={self._index}, {items})"

    def __contains__(self, key: str) -> bool:
        return key in self._data

    def __len__(self) -> int:
        return len(self._data)

    def __iter__(self):
        return iter(self._data)

    def keys(self):
        """返回字段名迭代器"""
        return self._data.keys()

    def values(self):
        """返回字段值迭代器"""
        return self._data.values()

    def items(self):
        """返回 (字段名, 字段值) 迭代器"""
        return self._data.items()

    def get(self, key: str, default: Any = None) -> Any:
        """获取字段值，不存在时返回 default"""
        return self._data.get(key, default)

    def to_dict(self) -> dict[str, Any]:
        """转换为原生 dict"""
        return dict(self._data)

index property

行索引

get(key, default=None)

获取字段值，不存在时返回 default

Source code in dimine_python_sdk\lib\db\datatable.py

def get(self, key: str, default: Any = None) -> Any:
    """获取字段值，不存在时返回 default"""
    return self._data.get(key, default)

items()

返回 (字段名, 字段值) 迭代器

Source code in dimine_python_sdk\lib\db\datatable.py

def items(self):
    """返回 (字段名, 字段值) 迭代器"""
    return self._data.items()

keys()

返回字段名迭代器

Source code in dimine_python_sdk\lib\db\datatable.py

def keys(self):
    """返回字段名迭代器"""
    return self._data.keys()

to_dict()

转换为原生 dict

Source code in dimine_python_sdk\lib\db\datatable.py

def to_dict(self) -> dict[str, Any]:
    """转换为原生 dict"""
    return dict(self._data)

values()

返回字段值迭代器

Source code in dimine_python_sdk\lib\db\datatable.py

def values(self):
    """返回字段值迭代器"""
    return self._data.values()

data_table_conn(file_path)

数据表连接上下文管理器

Source code in dimine_python_sdk\lib\db\datatable.py

@contextmanager
def data_table_conn(file_path: str) -> Generator[DmDataTable, None, None]:
    """
    数据表连接上下文管理器
    """
    db = DmDataTable(file_path)
    try:
        yield db
    finally:
        db.save()