Skip to content

datatable

数据表(DmDataTable / DmDataTableRecord)与行代理(TableRow)封装

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

DmDataTable

数据表 Python 二次封装类

Source code in dimine_python_sdk\lib\db\datatable.py
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
class DmDataTable:
    """
    数据表 Python 二次封装类
    """

    def __init__(self, file_path: Optional[str] = None):
        self._obj = Dm.CDataTable()
        if file_path is not None:
            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.dmt")
        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_field(self, field_name: str, field_type: Union[int, str, type, TABLE_FIELD_TYPE]) -> bool | int:
        """
        添加字段到数据表

        Args:
            field_name: 字段名
            field_type: 字段类型,支持多种格式:
                - int: 枚举值(0-11)
                - str: 类型别名('string', 'int', 'float', 'double' 等)
                - type: Python 类型(str, int, float, bool)
                - TABLE_FIELD_TYPE: 枚举对象

        Returns:
            成功返回 True/0,失败返回 False/错误码
        """
        type_map = {
            str: TABLE_FIELD_TYPE.FIELD_STRING_TYPE.value,
            int: TABLE_FIELD_TYPE.FIELD_INT_TYPE.value,
            float: TABLE_FIELD_TYPE.FIELD_DOUBLE_TYPE.value,
            bool: TABLE_FIELD_TYPE.FIELD_BYTE_TYPE.value,
        }
        alias_map = {
            'string': TABLE_FIELD_TYPE.FIELD_STRING_TYPE.value,
            'int': TABLE_FIELD_TYPE.FIELD_INT_TYPE.value,
            'long': TABLE_FIELD_TYPE.FIELD_LONG_TYPE.value,
            'float': TABLE_FIELD_TYPE.FIELD_FLOAT_TYPE.value,
            'double': TABLE_FIELD_TYPE.FIELD_DOUBLE_TYPE.value,
            'byte': TABLE_FIELD_TYPE.FIELD_BYTE_TYPE.value,
            'short': TABLE_FIELD_TYPE.FIELD_SHORT_TYPE.value,
        }

        if isinstance(field_type, type) and field_type in type_map:
            field_type = type_map[field_type]
        elif isinstance(field_type, str) and field_type.lower() in alias_map:
            field_type = alias_map[field_type.lower()]
        elif isinstance(field_type, TABLE_FIELD_TYPE):
            field_type = field_type.value
        elif not isinstance(field_type, int):
            raise TypeError(f"不支持的字段类型: {field_type}")

        return self._obj.Add_Field(field_name, field_type)

    def add_record(self) -> DmDataTableRecord:
        """添加一条新记录,返回该记录对象"""
        self._obj.Add_Record()
        return self.get_record_from_index(self.get_record_count() - 1)

    def del_record(self, nindex: int) -> bool | int:
        """删除记录"""
        return self._obj.Del_Record(nindex)

    def set_file_name(self, field_name: str):
        """设置文件名"""
        return self._obj.SetFileName(field_name)

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

    def clear_records(self) -> None:
        """清空所有记录"""
        while self.get_record_count() > 0:
            self.del_record(0)

    def _infer_field_type_from_series(self, series) -> TABLE_FIELD_TYPE:
        """根据 pandas Series 的 dtype 推断字段类型"""
        try:
            import pandas as pd
        except ImportError:
            return TABLE_FIELD_TYPE.FIELD_STRING_TYPE

        if pd.api.types.is_integer_dtype(series):
            return TABLE_FIELD_TYPE.FIELD_INT_TYPE
        elif pd.api.types.is_float_dtype(series):
            return TABLE_FIELD_TYPE.FIELD_DOUBLE_TYPE
        elif pd.api.types.is_bool_dtype(series):
            return TABLE_FIELD_TYPE.FIELD_BYTE_TYPE
        else:
            return TABLE_FIELD_TYPE.FIELD_STRING_TYPE

    def insert_from_dataframe(
        self,
        df: "pd.DataFrame",
        column_mapping: Optional[dict[str, str]] = None,
        exclude_columns: Optional[list[str]] = None,
        field_types: Optional[dict[str, str]] = {},
    ) -> None:
        """
        从 DataFrame 插入数据

        Args:
            df: 要插入的 DataFrame
            column_mapping: 列名映射 {DataFrame列名: C++字段名},为 None 时直接使用原列名
            exclude_columns: 要跳过的列名列表(这些字段会创建但不写入记录值)

        Example:
            >>> dt.insert_from_dataframe(df)
            >>> dt.insert_from_dataframe(df, column_mapping={"工程号": "BHID"})
            >>> dt.insert_from_dataframe(df, exclude_columns=["TABLE_NAME", "DRILL_NAME"])
        """
        try:
            import pandas as pd
        except ImportError as exc:
            raise ImportError("使用 insert_from_dataframe() 需要安装 pandas: uv add pandas") from exc

        exclude = set(exclude_columns) if exclude_columns else set()

        # 如果表没有字段,根据 DataFrame 列自动创建(使用映射后的字段名)
        if self.get_field_count() == 0:
            for col in df.columns:
                field_name = column_mapping.get(col, col) if column_mapping else col
                field_type = field_types.get(field_name) or self._infer_field_type_from_series(df[col])
                self.add_field(field_name, field_type)

        for _, row in df.iterrows():
            record = self.add_record()
            for col in df.columns:
                if col in exclude:
                    continue
                field_name = column_mapping.get(col, col) if column_mapping else col
                if pd.notna(row[col]):
                    value = row[col]
                    if isinstance(value, bool):
                        value = int(value)
                    elif not isinstance(value, (str, int, float)):
                        value = str(value)
                    record.set_value(field_name, value)

    # ---------------------- 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_dataframe(cls, df: "pd.DataFrame", file_path: Optional[str] = None) -> "DmDataTable":
        """
        从 pandas DataFrame 直接创建 DmDataTable,并可一步保存为 .dmt 文件

        Args:
            df: 源 DataFrame
            file_path: 若提供,则自动保存到该路径(.dmt 文件)

        Returns:
            DmDataTable 实例

        Example:
            >>> DmDataTable.from_dataframe(df, "E:/output/my_table.dmt")
        """
        try:
            import pandas as pd
        except ImportError as exc:
            raise ImportError("使用 from_dataframe() 需要安装 pandas: uv add pandas") from exc

        if not isinstance(df, pd.DataFrame):
            raise TypeError("df 必须为 pandas DataFrame")

        instance = cls.__new__(cls)
        instance._obj = Dm.CDataTable()
        instance.insert_from_dataframe(df)

        if file_path is not None:
            instance.set_file_name(file_path)
            instance.save()

        return instance

    @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
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
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
439
440
441
442
def __iter__(self):
    """逐行迭代,每次返回 TableRow"""
    for i in range(len(self)):
        yield self[i]

__len__()

返回记录总数

Source code in dimine_python_sdk\lib\db\datatable.py
415
416
417
def __len__(self) -> int:
    """返回记录总数"""
    return self.get_record_count()

add_field(field_name, field_type)

添加字段到数据表

Parameters:

Name Type Description Default
field_name str

字段名

required
field_type Union[int, str, type, TABLE_FIELD_TYPE]

字段类型,支持多种格式: - int: 枚举值(0-11) - str: 类型别名('string', 'int', 'float', 'double' 等) - type: Python 类型(str, int, float, bool) - TABLE_FIELD_TYPE: 枚举对象

required

Returns:

Type Description
bool | int

成功返回 True/0,失败返回 False/错误码

Source code in dimine_python_sdk\lib\db\datatable.py
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def add_field(self, field_name: str, field_type: Union[int, str, type, TABLE_FIELD_TYPE]) -> bool | int:
    """
    添加字段到数据表

    Args:
        field_name: 字段名
        field_type: 字段类型,支持多种格式:
            - int: 枚举值(0-11)
            - str: 类型别名('string', 'int', 'float', 'double' 等)
            - type: Python 类型(str, int, float, bool)
            - TABLE_FIELD_TYPE: 枚举对象

    Returns:
        成功返回 True/0,失败返回 False/错误码
    """
    type_map = {
        str: TABLE_FIELD_TYPE.FIELD_STRING_TYPE.value,
        int: TABLE_FIELD_TYPE.FIELD_INT_TYPE.value,
        float: TABLE_FIELD_TYPE.FIELD_DOUBLE_TYPE.value,
        bool: TABLE_FIELD_TYPE.FIELD_BYTE_TYPE.value,
    }
    alias_map = {
        'string': TABLE_FIELD_TYPE.FIELD_STRING_TYPE.value,
        'int': TABLE_FIELD_TYPE.FIELD_INT_TYPE.value,
        'long': TABLE_FIELD_TYPE.FIELD_LONG_TYPE.value,
        'float': TABLE_FIELD_TYPE.FIELD_FLOAT_TYPE.value,
        'double': TABLE_FIELD_TYPE.FIELD_DOUBLE_TYPE.value,
        'byte': TABLE_FIELD_TYPE.FIELD_BYTE_TYPE.value,
        'short': TABLE_FIELD_TYPE.FIELD_SHORT_TYPE.value,
    }

    if isinstance(field_type, type) and field_type in type_map:
        field_type = type_map[field_type]
    elif isinstance(field_type, str) and field_type.lower() in alias_map:
        field_type = alias_map[field_type.lower()]
    elif isinstance(field_type, TABLE_FIELD_TYPE):
        field_type = field_type.value
    elif not isinstance(field_type, int):
        raise TypeError(f"不支持的字段类型: {field_type}")

    return self._obj.Add_Field(field_name, field_type)

add_record()

添加一条新记录,返回该记录对象

Source code in dimine_python_sdk\lib\db\datatable.py
327
328
329
330
def add_record(self) -> DmDataTableRecord:
    """添加一条新记录,返回该记录对象"""
    self._obj.Add_Record()
    return self.get_record_from_index(self.get_record_count() - 1)

clear_records()

清空所有记录

Source code in dimine_python_sdk\lib\db\datatable.py
344
345
346
347
def clear_records(self) -> None:
    """清空所有记录"""
    while self.get_record_count() > 0:
        self.del_record(0)

del_record(nindex)

删除记录

Source code in dimine_python_sdk\lib\db\datatable.py
332
333
334
def del_record(self, nindex: int) -> bool | int:
    """删除记录"""
    return self._obj.Del_Record(nindex)

from_dataframe(df, file_path=None) classmethod

从 pandas DataFrame 直接创建 DmDataTable,并可一步保存为 .dmt 文件

Parameters:

Name Type Description Default
df DataFrame

源 DataFrame

required
file_path Optional[str]

若提供,则自动保存到该路径(.dmt 文件)

None

Returns:

Type Description
DmDataTable

DmDataTable 实例

Example

DmDataTable.from_dataframe(df, "E:/output/my_table.dmt")

Source code in dimine_python_sdk\lib\db\datatable.py
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
@classmethod
def from_dataframe(cls, df: "pd.DataFrame", file_path: Optional[str] = None) -> "DmDataTable":
    """
    从 pandas DataFrame 直接创建 DmDataTable,并可一步保存为 .dmt 文件

    Args:
        df: 源 DataFrame
        file_path: 若提供,则自动保存到该路径(.dmt 文件)

    Returns:
        DmDataTable 实例

    Example:
        >>> DmDataTable.from_dataframe(df, "E:/output/my_table.dmt")
    """
    try:
        import pandas as pd
    except ImportError as exc:
        raise ImportError("使用 from_dataframe() 需要安装 pandas: uv add pandas") from exc

    if not isinstance(df, pd.DataFrame):
        raise TypeError("df 必须为 pandas DataFrame")

    instance = cls.__new__(cls)
    instance._obj = Dm.CDataTable()
    instance.insert_from_dataframe(df)

    if file_path is not None:
        instance.set_file_name(file_path)
        instance.save()

    return instance

get_field_count()

获取字段总数

Source code in dimine_python_sdk\lib\db\datatable.py
207
208
209
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
241
242
243
244
245
246
247
248
249
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
223
224
225
226
227
228
229
230
231
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
233
234
235
236
237
238
239
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
251
252
253
254
255
256
257
258
259
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
211
212
213
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
272
273
274
275
276
277
278
279
280
281
282
283
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
215
216
217
218
219
220
221
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
261
262
263
264
265
266
267
268
269
270
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)

insert_from_dataframe(df, column_mapping=None, exclude_columns=None, field_types={})

从 DataFrame 插入数据

Parameters:

Name Type Description Default
df DataFrame

要插入的 DataFrame

required
column_mapping Optional[dict[str, str]]

列名映射 {DataFrame列名: C++字段名},为 None 时直接使用原列名

None
exclude_columns Optional[list[str]]

要跳过的列名列表(这些字段会创建但不写入记录值)

None
Example

dt.insert_from_dataframe(df) dt.insert_from_dataframe(df, column_mapping={"工程号": "BHID"}) dt.insert_from_dataframe(df, exclude_columns=["TABLE_NAME", "DRILL_NAME"])

Source code in dimine_python_sdk\lib\db\datatable.py
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def insert_from_dataframe(
    self,
    df: "pd.DataFrame",
    column_mapping: Optional[dict[str, str]] = None,
    exclude_columns: Optional[list[str]] = None,
    field_types: Optional[dict[str, str]] = {},
) -> None:
    """
    从 DataFrame 插入数据

    Args:
        df: 要插入的 DataFrame
        column_mapping: 列名映射 {DataFrame列名: C++字段名},为 None 时直接使用原列名
        exclude_columns: 要跳过的列名列表(这些字段会创建但不写入记录值)

    Example:
        >>> dt.insert_from_dataframe(df)
        >>> dt.insert_from_dataframe(df, column_mapping={"工程号": "BHID"})
        >>> dt.insert_from_dataframe(df, exclude_columns=["TABLE_NAME", "DRILL_NAME"])
    """
    try:
        import pandas as pd
    except ImportError as exc:
        raise ImportError("使用 insert_from_dataframe() 需要安装 pandas: uv add pandas") from exc

    exclude = set(exclude_columns) if exclude_columns else set()

    # 如果表没有字段,根据 DataFrame 列自动创建(使用映射后的字段名)
    if self.get_field_count() == 0:
        for col in df.columns:
            field_name = column_mapping.get(col, col) if column_mapping else col
            field_type = field_types.get(field_name) or self._infer_field_type_from_series(df[col])
            self.add_field(field_name, field_type)

    for _, row in df.iterrows():
        record = self.add_record()
        for col in df.columns:
            if col in exclude:
                continue
            field_name = column_mapping.get(col, col) if column_mapping else col
            if pd.notna(row[col]):
                value = row[col]
                if isinstance(value, bool):
                    value = int(value)
                elif not isinstance(value, (str, int, float)):
                    value = str(value)
                record.set_value(field_name, value)

load(file_path)

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

example:

dt = DmDataTable()
dt.load("E:/dassistant/tests/123.dmt")
print("字段总数:",dt.get_field_count())
Source code in dimine_python_sdk\lib\db\datatable.py
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
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.dmt")
    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
340
341
342
def save(self) -> bool | int:
    """保存数据表"""
    return self._obj.Save()

set_file_name(field_name)

设置文件名

Source code in dimine_python_sdk\lib\db\datatable.py
336
337
338
def set_file_name(self, field_name: str):
    """设置文件名"""
    return self._obj.SetFileName(field_name)

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
472
473
474
475
476
477
478
479
480
481
482
483
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
460
461
462
463
464
465
466
467
468
469
470
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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
150
151
152
153
154
155
156
157
158
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
140
141
142
143
144
145
146
147
148
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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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("字段值仅支持字符串或数字类型")

TABLE_FIELD_TYPE

Bases: Enum

表字段类型枚举(对应底层 C++ 枚举)

Source code in dimine_python_sdk\lib\db\datatable.py
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class TABLE_FIELD_TYPE(Enum):
    """表字段类型枚举(对应底层 C++ 枚举)"""
    FIELD_NONE_TYPE = 0       # 未知
    FIELD_BYTE_TYPE = 1       # 字节型
    FIELD_SHORT_TYPE = 2      # 短整型
    FIELD_INT_TYPE = 3        # 整型
    FIELD_LONG_TYPE = 4       # 长整型
    FIELD_FLOAT_TYPE = 5      # 浮点型
    FIELD_DOUBLE_TYPE = 6     # 双精度型
    FIELD_STRING_TYPE = 7     # 字符串
    FIELD_COLOR_TYPE = 8      # 颜色型
    FIELD_DATE_TYPE = 9       # 日期型
    FIELD_BINARY_TYPE = 10    # 二进制
    FIELD_INT64_TYPE = 11     # 64位整型
    FIELD_TYPE_NUM = 12

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
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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
111
112
113
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
107
108
109
def items(self):
    """返回 (字段名, 字段值) 迭代器"""
    return self._data.items()

keys()

返回字段名迭代器

Source code in dimine_python_sdk\lib\db\datatable.py
 99
100
101
def keys(self):
    """返回字段名迭代器"""
    return self._data.keys()

to_dict()

转换为原生 dict

Source code in dimine_python_sdk\lib\db\datatable.py
115
116
117
def to_dict(self) -> dict[str, Any]:
    """转换为原生 dict"""
    return dict(self._data)

values()

返回字段值迭代器

Source code in dimine_python_sdk\lib\db\datatable.py
103
104
105
def values(self):
    """返回字段值迭代器"""
    return self._data.values()

data_table_conn(file_path)

数据表连接上下文管理器

Source code in dimine_python_sdk\lib\db\datatable.py
534
535
536
537
538
539
540
541
542
543
@contextmanager
def data_table_conn(file_path: str) -> Generator[DmDataTable, None, None]:
    """
    数据表连接上下文管理器
    """
    db = DmDataTable(file_path)
    try:
        yield db
    finally:
        db.save()