data_table

Native Layer v2 数据表模块

CDataTable <-> pd.DataFrame 转换，对外不暴露 C++ 对象。

NativeDataTableHandle

数据表不透明句柄，内部持有 C++ CDataTable 对象

Source code in dimine_python_sdk\lib\native\data_table.py

class NativeDataTableHandle:
    """数据表不透明句柄，内部持有 C++ CDataTable 对象"""

    def __init__(self, cpp_obj: Any):
        self._cpp_obj = cpp_obj

    @classmethod
    def create(cls) -> "NativeDataTableHandle":
        """创建空数据表"""
        return cls(Dm.CDataTable())

    @classmethod
    def from_dataframe(cls, df: "pd.DataFrame") -> "NativeDataTableHandle":
        """从 DataFrame 创建数据表"""
        require_pandas()
        import pandas as pd

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

        handle = cls.create()
        if len(df.columns) == 0:
            return handle

        for col in df.columns:
            handle.add_field(col, _infer_field_type_from_series(df[col]))

        for _, row in df.iterrows():
            record = handle.add_record()
            for col in df.columns:
                value = row[col]
                if pd.isna(value):
                    continue
                if isinstance(value, bool):
                    value = int(value)
                elif isinstance(value, bytes):
                    pass  # 原样写入（binary 字段）
                elif not isinstance(value, (str, int, float)):
                    value = str(value)
                record[col] = value

        return handle

    # ----------------------------------------------------------------------
    # 基本操作
    # ----------------------------------------------------------------------

    def load(self, file_path: str) -> None:
        """加载数据表文件"""
        result = self._cpp_obj.Load(file_path)
        if result is False:
            raise NativeDataTableError(f"加载数据表失败: {file_path}")

    def save(self, file_path: str | None = None) -> None:
        """保存数据表"""
        if file_path is not None:
            self._cpp_obj.SetFileName(file_path)
        self._cpp_obj.Save()
        #raise NativeDataTableError("保存数据表失败")

    def field_names(self) -> list[str]:
        """返回字段名列表"""
        count = self._cpp_obj.Get_Field_Count()
        return [self._cpp_obj.Get_Field_Name(i) for i in range(count)]

    def field_types(self) -> list[int]:
        """返回字段类型编码列表"""
        count = self._cpp_obj.Get_Field_Count()
        return [self._cpp_obj.Get_Field_Type(i) for i in range(count)]

    def field_definitions(self) -> list[NativeFieldDef]:
        """返回字段定义列表"""
        count = self._cpp_obj.Get_Field_Count()
        defs = []
        for i in range(count):
            name = self._cpp_obj.Get_Field_Name(i)
            ftype = self._cpp_obj.Get_Field_Type(i)
            defs.append(NativeFieldDef(name=name, type=ftype))
        return defs

    def record_count(self) -> int:
        """返回记录总数"""
        if self._cpp_obj is None:
            return 0
        return self._cpp_obj.Get_Record_Count()

    def get_record(self, index: int) -> dict:
        """通过索引获取单条记录字典。

        BINARY 字段通过 ``CDataTableRecord.asBytes`` 还原为原始 ``bytes``
        （``Get_Record_Data`` 对 binary 走 ``asDouble`` 会把字节打烂）。
        """
        if index < 0 or index >= self.record_count():
            raise IndexError(f"记录索引 {index} 超出范围 [0, {self.record_count()})")
        data = self._cpp_obj.Get_Record_Data(index)
        binary_fields = [
            name
            for name, ftype in zip(self.field_names(), self.field_types())
            if ftype == NativeFieldType.BINARY
        ]
        if binary_fields:
            record = self._cpp_obj.Get_Record_From_Index(index)
            as_bytes = getattr(record, "asBytes", None)
            if as_bytes is not None:
                for name in binary_fields:
                    data[name] = as_bytes(name)
        return data

    def add_field(self, name: str, field_type: int | str) -> None:
        """添加字段"""
        code = _resolve_field_type(field_type)
        self._cpp_obj.Add_Field(name, code)

    def add_record(self) -> "_RecordProxy":
        """添加一条新记录，返回记录代理"""
        record = self._cpp_obj.Add_Record()
        return _RecordProxy(record)

    def clear_records(self) -> None:
        """清空所有记录"""
        while self.record_count() > 0:
            self._cpp_obj.Del_Record(0)

    def to_dataframe(self) -> "pd.DataFrame":
        """转换为 pandas DataFrame"""
        require_pandas()
        import pandas as pd

        rows = [self.get_record(i) for i in range(self.record_count())]
        return pd.DataFrame(rows, columns=self.field_names())

    def insert_from_dataframe(
        self,
        df: "pd.DataFrame",
        column_mapping: dict[str, str] | None = None,
        exclude_columns: list[str] | None = None,
        field_types: dict[str, str] | None = None,
    ) -> None:
        """从 DataFrame 批量插入数据"""
        require_pandas()
        import pandas as pd

        column_mapping = column_mapping or {}
        exclude = set(exclude_columns or [])
        field_types = field_types or {}

        if self.record_count() == 0 and self._cpp_obj.Get_Field_Count() == 0:
            for col in df.columns:
                field_name = column_mapping.get(col, col)
                ftype = field_types.get(field_name)
                if ftype is None:
                    ftype = _infer_field_type_from_series(df[col])
                self.add_field(field_name, ftype)

        for _, row in df.iterrows():
            record = self.add_record()
            for col in df.columns:
                field_name = column_mapping.get(col, col)
                if field_name in exclude:
                    continue
                value = row[col]
                if pd.isna(value):
                    continue
                if isinstance(value, bool):
                    value = int(value)
                elif isinstance(value, bytes):
                    pass  # 原样写入（binary 字段）
                elif not isinstance(value, (str, int, float)):
                    value = str(value)
                record[field_name] = value

    def __len__(self) -> int:
        return self.record_count()

    def __iter__(self):
        for i in range(self.record_count()):
            yield self.get_record(i)

    def __getitem__(self, key: str | int | slice):
        if isinstance(key, str):
            return [row[key] for row in self]
        if isinstance(key, int):
            return self.get_record(key)
        if isinstance(key, slice):
            return [self.get_record(i) for i in range(*key.indices(len(self)))]
        raise TypeError(f"不支持的 key 类型: {type(key)}")

add_field(name, field_type)

添加字段

Source code in dimine_python_sdk\lib\native\data_table.py

def add_field(self, name: str, field_type: int | str) -> None:
    """添加字段"""
    code = _resolve_field_type(field_type)
    self._cpp_obj.Add_Field(name, code)

add_record()

添加一条新记录，返回记录代理

Source code in dimine_python_sdk\lib\native\data_table.py

def add_record(self) -> "_RecordProxy":
    """添加一条新记录，返回记录代理"""
    record = self._cpp_obj.Add_Record()
    return _RecordProxy(record)

clear_records()

清空所有记录

Source code in dimine_python_sdk\lib\native\data_table.py

def clear_records(self) -> None:
    """清空所有记录"""
    while self.record_count() > 0:
        self._cpp_obj.Del_Record(0)

create() classmethod

创建空数据表

Source code in dimine_python_sdk\lib\native\data_table.py

@classmethod
def create(cls) -> "NativeDataTableHandle":
    """创建空数据表"""
    return cls(Dm.CDataTable())

field_definitions()

返回字段定义列表

Source code in dimine_python_sdk\lib\native\data_table.py

def field_definitions(self) -> list[NativeFieldDef]:
    """返回字段定义列表"""
    count = self._cpp_obj.Get_Field_Count()
    defs = []
    for i in range(count):
        name = self._cpp_obj.Get_Field_Name(i)
        ftype = self._cpp_obj.Get_Field_Type(i)
        defs.append(NativeFieldDef(name=name, type=ftype))
    return defs

field_names()

返回字段名列表

Source code in dimine_python_sdk\lib\native\data_table.py

def field_names(self) -> list[str]:
    """返回字段名列表"""
    count = self._cpp_obj.Get_Field_Count()
    return [self._cpp_obj.Get_Field_Name(i) for i in range(count)]

field_types()

返回字段类型编码列表

Source code in dimine_python_sdk\lib\native\data_table.py

def field_types(self) -> list[int]:
    """返回字段类型编码列表"""
    count = self._cpp_obj.Get_Field_Count()
    return [self._cpp_obj.Get_Field_Type(i) for i in range(count)]

from_dataframe(df) classmethod

从 DataFrame 创建数据表

Source code in dimine_python_sdk\lib\native\data_table.py

@classmethod
def from_dataframe(cls, df: "pd.DataFrame") -> "NativeDataTableHandle":
    """从 DataFrame 创建数据表"""
    require_pandas()
    import pandas as pd

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

    handle = cls.create()
    if len(df.columns) == 0:
        return handle

    for col in df.columns:
        handle.add_field(col, _infer_field_type_from_series(df[col]))

    for _, row in df.iterrows():
        record = handle.add_record()
        for col in df.columns:
            value = row[col]
            if pd.isna(value):
                continue
            if isinstance(value, bool):
                value = int(value)
            elif isinstance(value, bytes):
                pass  # 原样写入（binary 字段）
            elif not isinstance(value, (str, int, float)):
                value = str(value)
            record[col] = value

    return handle

get_record(index)

通过索引获取单条记录字典。

BINARY 字段通过 CDataTableRecord.asBytes 还原为原始 bytes （Get_Record_Data 对 binary 走 asDouble 会把字节打烂）。

Source code in dimine_python_sdk\lib\native\data_table.py

def get_record(self, index: int) -> dict:
    """通过索引获取单条记录字典。

    BINARY 字段通过 ``CDataTableRecord.asBytes`` 还原为原始 ``bytes``
    （``Get_Record_Data`` 对 binary 走 ``asDouble`` 会把字节打烂）。
    """
    if index < 0 or index >= self.record_count():
        raise IndexError(f"记录索引 {index} 超出范围 [0, {self.record_count()})")
    data = self._cpp_obj.Get_Record_Data(index)
    binary_fields = [
        name
        for name, ftype in zip(self.field_names(), self.field_types())
        if ftype == NativeFieldType.BINARY
    ]
    if binary_fields:
        record = self._cpp_obj.Get_Record_From_Index(index)
        as_bytes = getattr(record, "asBytes", None)
        if as_bytes is not None:
            for name in binary_fields:
                data[name] = as_bytes(name)
    return data

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

从 DataFrame 批量插入数据

Source code in dimine_python_sdk\lib\native\data_table.py

def insert_from_dataframe(
    self,
    df: "pd.DataFrame",
    column_mapping: dict[str, str] | None = None,
    exclude_columns: list[str] | None = None,
    field_types: dict[str, str] | None = None,
) -> None:
    """从 DataFrame 批量插入数据"""
    require_pandas()
    import pandas as pd

    column_mapping = column_mapping or {}
    exclude = set(exclude_columns or [])
    field_types = field_types or {}

    if self.record_count() == 0 and self._cpp_obj.Get_Field_Count() == 0:
        for col in df.columns:
            field_name = column_mapping.get(col, col)
            ftype = field_types.get(field_name)
            if ftype is None:
                ftype = _infer_field_type_from_series(df[col])
            self.add_field(field_name, ftype)

    for _, row in df.iterrows():
        record = self.add_record()
        for col in df.columns:
            field_name = column_mapping.get(col, col)
            if field_name in exclude:
                continue
            value = row[col]
            if pd.isna(value):
                continue
            if isinstance(value, bool):
                value = int(value)
            elif isinstance(value, bytes):
                pass  # 原样写入（binary 字段）
            elif not isinstance(value, (str, int, float)):
                value = str(value)
            record[field_name] = value

load(file_path)

加载数据表文件

Source code in dimine_python_sdk\lib\native\data_table.py

def load(self, file_path: str) -> None:
    """加载数据表文件"""
    result = self._cpp_obj.Load(file_path)
    if result is False:
        raise NativeDataTableError(f"加载数据表失败: {file_path}")

record_count()

返回记录总数

Source code in dimine_python_sdk\lib\native\data_table.py

def record_count(self) -> int:
    """返回记录总数"""
    if self._cpp_obj is None:
        return 0
    return self._cpp_obj.Get_Record_Count()

save(file_path=None)

保存数据表

Source code in dimine_python_sdk\lib\native\data_table.py

def save(self, file_path: str | None = None) -> None:
    """保存数据表"""
    if file_path is not None:
        self._cpp_obj.SetFileName(file_path)
    self._cpp_obj.Save()

to_dataframe()

转换为 pandas DataFrame

Source code in dimine_python_sdk\lib\native\data_table.py

def to_dataframe(self) -> "pd.DataFrame":
    """转换为 pandas DataFrame"""
    require_pandas()
    import pandas as pd

    rows = [self.get_record(i) for i in range(self.record_count())]
    return pd.DataFrame(rows, columns=self.field_names())

NativeFieldType

Bases: IntEnum

CDataTable 字段类型编码

Source code in dimine_python_sdk\lib\native\data_table.py

class NativeFieldType(IntEnum):
    """CDataTable 字段类型编码"""

    UNKNOWN = 0
    BYTE = 1
    SHORT = 2
    INT = 3
    LONG = 4
    FLOAT = 5
    DOUBLE = 6
    STRING = 7
    COLOR = 8
    DATE = 9
    BINARY = 10
    LONG64 = 11