Skip to content

dmt_file

DMT 文件业务层：DmtFile、dmt_conn 与相关异常

职责： - 提供 Pythonic 的 DMT 数据表文件接口； - 内部以 pandas DataFrame 作为数据表内存模型； - 加载/保存均委托给 native.data_table 的 NativeDataTableHandle。

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

DmtError

Bases: RuntimeError

DMT 操作异常基类

Source code in dimine_python_sdk\lib\io\dmt_file.py 
41
42
43
44
class DmtError(RuntimeError):
    """DMT 操作异常基类"""

    pass

DmtFile

DMT 数据表文件管理器。

DMT 文件在 Dimine 中用于存储结构化数据表。本类将 DMT 文件映射为 pandas DataFrame，加载时通过 native 层读取为 DataFrame，保存时将 DataFrame 写回 native 层。

用法::

# 从文件加载
dmt = DmtFile("data.dmt")
print(dmt.columns)
print(dmt.data.head())

# 修改后保存
dmt.data["new_col"] = 0
dmt.save("output.dmt")

# 从 DataFrame 创建并保存
import pandas as pd
df = pd.DataFrame({"x": [1, 2], "y": [3, 4]})
DmtFile(data=df).save("new.dmt")
Source code in dimine_python_sdk\lib\io\dmt_file.py 
 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
118
119
120
121
122
123
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
174
175
176
177
178
179
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
class DmtFile:
    """
    DMT 数据表文件管理器。

    DMT 文件在 Dimine 中用于存储结构化数据表。本类将 DMT 文件映射为
    pandas DataFrame，加载时通过 native 层读取为 DataFrame，保存时将
    DataFrame 写回 native 层。

    用法::

        # 从文件加载
        dmt = DmtFile("data.dmt")
        print(dmt.columns)
        print(dmt.data.head())

        # 修改后保存
        dmt.data["new_col"] = 0
        dmt.save("output.dmt")

        # 从 DataFrame 创建并保存
        import pandas as pd
        df = pd.DataFrame({"x": [1, 2], "y": [3, 4]})
        DmtFile(data=df).save("new.dmt")
    """

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

        if data is not None:
            self.data = data

        if file_path is not None:
            self.load(file_path)

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

    def load(self, file_path: Union[str, Path]) -> "DmtFile":
        """
        加载 .dmt 文件到 DataFrame。

        Returns:
            self（支持链式调用）

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

        try:
            handle = NativeDataTableHandle.create()
            handle.load(str(path))
            self._df = handle.to_dataframe()
        except NativeDataTableError as exc:
            raise DmtLoadError(f"加载 .dmt 失败: {file_path}") from exc
        except Exception as exc:
            raise DmtLoadError(f"加载 .dmt 异常: {file_path}") from exc

        self.file_path = str(path)
        return self

    def save(self, file_path: Optional[Union[str, Path]] = None) -> None:
        """
        将当前 DataFrame 保存为 .dmt 文件。

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

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

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

        try:
            handle = NativeDataTableHandle.from_dataframe(self._df)
            handle.save(str(save_path))
        except NativeDataTableError as exc:
            raise DmtSaveError(f"保存 .dmt 失败: {save_path}") from exc
        except Exception as exc:
            raise DmtSaveError(f"保存 .dmt 异常: {save_path}") from exc

        self.file_path = str(save_path)

    def close(self) -> None:
        """释放内存中的数据"""
        self._df = None
        self.file_path = None

    # ------------------------------------------------------------------
    # 数据访问
    # ------------------------------------------------------------------

    @property
    def data(self) -> "pd.DataFrame":
        """当前数据表（DataFrame）"""
        if self._df is None:
            raise DmtError("数据表未加载")
        return self._df

    @data.setter
    def data(self, value: "pd.DataFrame") -> None:
        import pandas as pd

        if not isinstance(value, pd.DataFrame):
            raise TypeError("data 必须为 pandas DataFrame")
        self._df = value.copy()

    def to_dataframe(self) -> "pd.DataFrame":
        """返回当前数据表的副本"""
        return self.data.copy()

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

    @property
    def name(self) -> str:
        """文件名（不含扩展名）"""
        return Path(self.file_path).stem if self.file_path else ""

    @property
    def columns(self) -> List[str]:
        """字段名列表"""
        return list(self._df.columns) if self._df is not None else []

    @property
    def shape(self) -> tuple[int, int]:
        """数据表形状 (行数, 列数)"""
        if self._df is None:
            return (0, 0)
        return self._df.shape

    @property
    def empty(self) -> bool:
        """数据表是否为空（无行或无列）"""
        if self._df is None:
            return True
        return self._df.empty

    @property
    def record_count(self) -> int:
        """记录行数"""
        return len(self._df) if self._df is not None else 0

    @property
    def info(self) -> dict[str, Any]:
        """DMT 文件摘要信息"""
        return {
            "file": self.file_path or "",
            "name": self.name,
            "columns": self.columns,
            "record_count": self.record_count,
            "shape": self.shape,
        }

    # ------------------------------------------------------------------
    # Pythonic 访问
    # ------------------------------------------------------------------

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

    def __iter__(self):
        """按行迭代，每行为字典"""
        df = self.data
        for _, row in df.iterrows():
            yield row.to_dict()

    def __getitem__(self, key: Union[str, int, slice]):
        """
        支持：
          - str: 返回整列数据列表
          - int: 返回单行字典
          - slice: 返回行字典列表
        """
        df = self.data
        if isinstance(key, str):
            return df[key].tolist()
        if isinstance(key, int):
            if key < 0 or key >= len(df):
                raise IndexError(f"行索引 {key} 超出范围 [0, {len(df)})")
            return df.iloc[key].to_dict()
        if isinstance(key, slice):
            return [df.iloc[i].to_dict() for i in range(*key.indices(len(df)))]
        raise TypeError(f"不支持的 key 类型: {type(key)}")

    def __contains__(self, column: str) -> bool:
        return column in self.columns

    # ------------------------------------------------------------------
    # 数据修改
    # ------------------------------------------------------------------

    def get_record(self, index: int) -> dict:
        """获取指定索引的单行记录（字典）"""
        return self.__getitem__(index)

    def add_record(self, data: Optional[dict[str, Any]] = None) -> int:
        """
        添加一行新记录。

        Args:
            data: 可选的字段名→值字典

        Returns:
            新记录的索引位置
        """
        import pandas as pd

        df = self.data
        new_index = len(df)
        # 用 NaN 占位，后续 set_value 填充
        row = {col: pd.NA for col in df.columns}
        if data:
            row.update(data)
        self._df = pd.concat([df, pd.DataFrame([row])], ignore_index=True)
        return new_index

    def set_value(self, index: int, field: str, value: Any) -> None:
        """
        设置指定行指定字段的值。

        Args:
            index: 行索引（0-based）
            field: 字段名
            value: 要设置的值

        Raises:
            IndexError: 行索引超出范围
            KeyError: 字段不存在
        """
        df = self.data
        if index < 0 or index >= len(df):
            raise IndexError(f"行索引 {index} 超出范围 [0, {len(df)})")
        if field not in df.columns:
            raise KeyError(f"字段 '{field}' 不存在。可用字段: {list(df.columns)}")
        self._df.at[index, field] = value

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

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

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

columns property

字段名列表

data property writable

当前数据表（DataFrame）

empty property

数据表是否为空（无行或无列）

info property

DMT 文件摘要信息

name property

文件名（不含扩展名）

record_count property

记录行数

shape property

数据表形状 (行数, 列数)

__getitem__(key)

支持： - str: 返回整列数据列表 - int: 返回单行字典 - slice: 返回行字典列表

Source code in dimine_python_sdk\lib\io\dmt_file.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def __getitem__(self, key: Union[str, int, slice]):
    """
    支持：
      - str: 返回整列数据列表
      - int: 返回单行字典
      - slice: 返回行字典列表
    """
    df = self.data
    if isinstance(key, str):
        return df[key].tolist()
    if isinstance(key, int):
        if key < 0 or key >= len(df):
            raise IndexError(f"行索引 {key} 超出范围 [0, {len(df)})")
        return df.iloc[key].to_dict()
    if isinstance(key, slice):
        return [df.iloc[i].to_dict() for i in range(*key.indices(len(df)))]
    raise TypeError(f"不支持的 key 类型: {type(key)}")

__iter__()

按行迭代，每行为字典

Source code in dimine_python_sdk\lib\io\dmt_file.py 
239
240
241
242
243
def __iter__(self):
    """按行迭代，每行为字典"""
    df = self.data
    for _, row in df.iterrows():
        yield row.to_dict()

add_record(data=None)

添加一行新记录。

Parameters:

Name Type Description Default
data Optional[dict[str, Any]]

可选的字段名→值字典

 None

Returns:

Type Description
int

新记录的索引位置
Source code in dimine_python_sdk\lib\io\dmt_file.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def add_record(self, data: Optional[dict[str, Any]] = None) -> int:
    """
    添加一行新记录。

    Args:
        data: 可选的字段名→值字典

    Returns:
        新记录的索引位置
    """
    import pandas as pd

    df = self.data
    new_index = len(df)
    # 用 NaN 占位，后续 set_value 填充
    row = {col: pd.NA for col in df.columns}
    if data:
        row.update(data)
    self._df = pd.concat([df, pd.DataFrame([row])], ignore_index=True)
    return new_index

close()

释放内存中的数据

Source code in dimine_python_sdk\lib\io\dmt_file.py 
160
161
162
163
def close(self) -> None:
    """释放内存中的数据"""
    self._df = None
    self.file_path = None

get_record(index)

获取指定索引的单行记录（字典）

Source code in dimine_python_sdk\lib\io\dmt_file.py 
270
271
272
def get_record(self, index: int) -> dict:
    """获取指定索引的单行记录（字典）"""
    return self.__getitem__(index)

load(file_path)

加载 .dmt 文件到 DataFrame。

Returns:

Type Description
'DmtFile'

self（支持链式调用）

Raises:

Type Description
DmtLoadError

文件不存在或加载失败
Source code in dimine_python_sdk\lib\io\dmt_file.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def load(self, file_path: Union[str, Path]) -> "DmtFile":
    """
    加载 .dmt 文件到 DataFrame。

    Returns:
        self（支持链式调用）

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

    try:
        handle = NativeDataTableHandle.create()
        handle.load(str(path))
        self._df = handle.to_dataframe()
    except NativeDataTableError as exc:
        raise DmtLoadError(f"加载 .dmt 失败: {file_path}") from exc
    except Exception as exc:
        raise DmtLoadError(f"加载 .dmt 异常: {file_path}") from exc

    self.file_path = str(path)
    return self

save(file_path=None)

将当前 DataFrame 保存为 .dmt 文件。

Parameters:

Name Type Description Default
file_path Optional[Union[str, Path]]

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

 None

Raises:

Type Description
DmtSaveError

未指定路径或无数据时抛出
Source code in dimine_python_sdk\lib\io\dmt_file.py 
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
def save(self, file_path: Optional[Union[str, Path]] = None) -> None:
    """
    将当前 DataFrame 保存为 .dmt 文件。

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

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

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

    try:
        handle = NativeDataTableHandle.from_dataframe(self._df)
        handle.save(str(save_path))
    except NativeDataTableError as exc:
        raise DmtSaveError(f"保存 .dmt 失败: {save_path}") from exc
    except Exception as exc:
        raise DmtSaveError(f"保存 .dmt 异常: {save_path}") from exc

    self.file_path = str(save_path)

set_value(index, field, value)

设置指定行指定字段的值。

Parameters:

Name Type Description Default
index int

行索引（0-based）

 required
field str

字段名

 required
value Any

要设置的值

 required

Raises:

Type Description
IndexError

行索引超出范围
KeyError

字段不存在
Source code in dimine_python_sdk\lib\io\dmt_file.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def set_value(self, index: int, field: str, value: Any) -> None:
    """
    设置指定行指定字段的值。

    Args:
        index: 行索引（0-based）
        field: 字段名
        value: 要设置的值

    Raises:
        IndexError: 行索引超出范围
        KeyError: 字段不存在
    """
    df = self.data
    if index < 0 or index >= len(df):
        raise IndexError(f"行索引 {index} 超出范围 [0, {len(df)})")
    if field not in df.columns:
        raise KeyError(f"字段 '{field}' 不存在。可用字段: {list(df.columns)}")
    self._df.at[index, field] = value

to_dataframe()

返回当前数据表的副本

Source code in dimine_python_sdk\lib\io\dmt_file.py 
184
185
186
def to_dataframe(self) -> "pd.DataFrame":
    """返回当前数据表的副本"""
    return self.data.copy()

DmtLoadError

Bases: DmtError

加载失败

Source code in dimine_python_sdk\lib\io\dmt_file.py 
47
48
49
50
class DmtLoadError(DmtError):
    """加载失败"""

    pass

DmtSaveError

Bases: DmtError

保存失败

Source code in dimine_python_sdk\lib\io\dmt_file.py 
53
54
55
56
class DmtSaveError(DmtError):
    """保存失败"""

    pass

dmt_conn(file_path=None)

DMT 文件连接上下文管理器

Source code in dimine_python_sdk\lib\io\dmt_file.py 
326
327
328
329
330
331
332
333
@contextmanager
def dmt_conn(file_path: Optional[Union[str, Path]] = None) -> Generator[DmtFile, None, None]:
    """DMT 文件连接上下文管理器"""
    dmt = DmtFile(file_path)
    try:
        yield dmt
    finally:
        dmt.close()