You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/08/07 09:35:51 UTC
(tsfile) 02/02: Add dataframe descriptions and device statistics
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch dataframe_desc
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 6cdcc9aba172f0a913d6cc0a2acb8a6f71563914
Author: ColinLee <sh...@163.com>
AuthorDate: Fri Aug 7 17:25:20 2026 +0800
Add dataframe descriptions and device statistics
---
python/README-zh.md | 43 ++++-
python/README.md | 44 ++++-
python/tests/test_tsfile_dataset.py | 231 ++++++++++++++++++++++
python/tsfile/dataset/dataframe.py | 370 +++++++++++++++++++++++++++++++++++-
4 files changed, 683 insertions(+), 5 deletions(-)
diff --git a/python/README-zh.md b/python/README-zh.md
index 660c001e8..91dc06698 100644
--- a/python/README-zh.md
+++ b/python/README-zh.md
@@ -37,6 +37,47 @@
你可以在 `./examples/examples.py` 中找到读写示例。
+## TsFileDataFrame 列角色描述
+
+`TsFileDataFrame` 可以加载表和列的 JSON 描述文件。列值使用 `C` 表示协变量，
+使用 `T` 表示目标变量：
+
+```json
+{
+ "weather": {
+ "temperature": "C",
+ "humidity": "T"
+ }
+}
+```
+
+创建 dataframe 时传入描述文件路径。`get` 和 `set` 提供对 JSON 顶层键的字典式
+访问，其中 `set` 会将更新后的描述写回文件；另外还可以按表获取协变量列、目标
+变量列及其数量：
+
+```python
+df = TsFileDataFrame("weather.tsfile", description_path="description.json")
+df.get("weather")
+df.get_covariate_columns("weather")
+df.get_target_columns("weather")
+df.get_covariate_column_count("weather")
+df.get_target_column_count("weather")
+df.get_device_count("weather")
+df.get_device_node_counts("weather")
+df.get_device_point_counts("weather")
+df.get_device_statistics("weather")
+df.get_device_point_count("weather", "device_a")
+df.get_device_stats("weather", {"device": "device_a"})
+df.list_device_metadata("weather")
+df.set("weather", {"temperature": "T", "humidity": "T"})
+df.close()
+```
+
+这里的节点数量指设备下实际存在的数值时间序列数量。`get_device_node_counts`
+使用设备有序标签值元组作为 key。`get_device_statistics` 还会返回点数、非空值数量
+和时间范围，`list_device_metadata` 则将标签展开为带名称的列。其中 `point_count`
+遵循 DataFrame 的时间线数量（包含 NaN 行），`value_count` 只统计非空值。
+
---
## 如何贡献
@@ -66,4 +107,4 @@ mvn -P with-cpp,with-python clean verify
```sh
python setup.py build_ext --inplace
-```
\ No newline at end of file
+```
diff --git a/python/README.md b/python/README.md
index 51cb498ec..42332c215 100644
--- a/python/README.md
+++ b/python/README.md
@@ -34,6 +34,49 @@ This directory contains the Python implementation of TsFile. The Python version
The source code can be found in the `./tsfile` directory. Files ending with `.pyx` and `.pyd` are wrapper code written in Cython. The `tsfile/tsfile.py` defines some user interfaces. You can find some examples of reading and writing in the `.examples/examples.py`.
+## TsFileDataFrame column roles
+
+`TsFileDataFrame` can load a JSON description for table columns. Use `C` for a
+covariate and `T` for a target column:
+
+```json
+{
+ "weather": {
+ "temperature": "C",
+ "humidity": "T"
+ }
+}
+```
+
+Pass the description path when opening the dataframe. `get` and `set` provide
+dictionary-style access to top-level JSON values; `set` writes the updated
+description back to disk.
+
+```python
+df = TsFileDataFrame("weather.tsfile", description_path="description.json")
+df.get("weather")
+df.get_covariate_columns("weather")
+df.get_target_columns("weather")
+df.get_covariate_column_count("weather")
+df.get_target_column_count("weather")
+df.get_device_count("weather")
+df.get_device_node_counts("weather")
+df.get_device_point_counts("weather")
+df.get_device_statistics("weather")
+df.get_device_point_count("weather", "device_a")
+df.get_device_stats("weather", {"device": "device_a"})
+df.list_device_metadata("weather")
+df.set("weather", {"temperature": "T", "humidity": "T"})
+df.close()
+```
+
+Device node counts refer to the physically present numeric time series under
+each device. `get_device_node_counts` uses each device's ordered tag-value tuple
+as its key, while `get_device_statistics` also reports point/value counts and
+time bounds, and `list_device_metadata` expands those tags into named columns.
+`point_count` follows the DataFrame timeline count (including NaN rows), while
+`value_count` counts non-null values.
+
## How to make contributions
@@ -60,4 +103,3 @@ Build by python command:
```sh
python setup.py build_ext --inplace
```
-
diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py
index 12b847e9e..53d272f0b 100644
--- a/python/tests/test_tsfile_dataset.py
+++ b/python/tests/test_tsfile_dataset.py
@@ -16,6 +16,8 @@
# under the License.
#
+import json
+
import numpy as np
import pandas as pd
import pytest
@@ -257,6 +259,212 @@ def test_dataset_basic_access_patterns(tmp_path, capsys):
assert "AlignedTimeseries(6 rows, 2 series)" in capsys.readouterr().out
+def test_dataset_description_get_set_and_column_roles(tmp_path):
+ path = tmp_path / "weather.tsfile"
+ description_path = tmp_path / "description.json"
+ _write_weather_file(path, 0)
+ description_path.write_text(
+ json.dumps(
+ {
+ "weather": {
+ "temperature": "C",
+ "humidity": "T",
+ }
+ }
+ ),
+ encoding="utf-8",
+ )
+
+ with TsFileDataFrame(
+ str(path),
+ show_progress=False,
+ description_path=description_path,
+ ) as tsdf:
+ assert tsdf.get("weather") == {
+ "temperature": "C",
+ "humidity": "T",
+ }
+ assert tsdf.get("missing") is None
+ assert tsdf.get("missing", {"default": True}) == {"default": True}
+ assert tsdf.get_covariate_columns("weather") == ["temperature"]
+ assert tsdf.get_target_columns("weather") == ["humidity"]
+ assert tsdf.get_covariate_column_count("weather") == 1
+ assert tsdf.get_target_column_count("weather") == 1
+ assert tsdf.get_covariate_columns("missing") == []
+ assert tsdf.get_target_column_count("missing") == 0
+
+ description = tsdf.get("weather")
+ description["temperature"] = "T"
+ assert tsdf.get_covariate_columns("weather") == ["temperature"]
+
+ with pytest.raises(TypeError, match="must be JSON serializable"):
+ tsdf.set("invalid", object())
+ assert tsdf.get("invalid") is None
+
+ subset = tsdf[:1]
+ subset.set(
+ "weather",
+ {
+ "temperature": "T",
+ "humidity": "T",
+ },
+ )
+ assert tsdf.get_covariate_column_count("weather") == 0
+ assert tsdf.get_target_columns("weather") == ["temperature", "humidity"]
+
+ assert json.loads(description_path.read_text(encoding="utf-8")) == {
+ "weather": {
+ "temperature": "T",
+ "humidity": "T",
+ }
+ }
+
+
+def test_dataset_description_set_creates_missing_file(tmp_path):
+ path = tmp_path / "weather.tsfile"
+ description_path = tmp_path / "nested" / "description.json"
+ _write_weather_file(path, 0)
+
+ with TsFileDataFrame(
+ str(path),
+ show_progress=False,
+ description_path=description_path,
+ ) as tsdf:
+ assert tsdf.get("weather") is None
+ tsdf.set("weather", {"temperature": "C", "humidity": "T"})
+
+ assert json.loads(description_path.read_text(encoding="utf-8")) == {
+ "weather": {"temperature": "C", "humidity": "T"}
+ }
+
+
+def test_dataset_description_rejects_invalid_json_and_roles(tmp_path):
+ path = tmp_path / "weather.tsfile"
+ description_path = tmp_path / "description.json"
+ _write_weather_file(path, 0)
+
+ description_path.write_text("[]", encoding="utf-8")
+ with pytest.raises(ValueError, match="must be a JSON object"):
+ TsFileDataFrame(
+ str(path),
+ show_progress=False,
+ description_path=description_path,
+ )
+
+ description_path.write_text(
+ json.dumps({"weather": {"temperature": "X"}}), encoding="utf-8"
+ )
+ with TsFileDataFrame(
+ str(path),
+ show_progress=False,
+ description_path=description_path,
+ ) as tsdf:
+ with pytest.raises(ValueError, match="expected 'C' or 'T'"):
+ tsdf.get_covariate_columns("weather")
+
+ tsdf.set("weather", {"temperature": ["C"]})
+ with pytest.raises(ValueError, match="expected 'C' or 'T'"):
+ tsdf.get_target_columns("weather")
+
+
+def test_dataset_device_counts_and_metadata_follow_dataframe_view(tmp_path):
+ path = tmp_path / "multi_device.tsfile"
+ _write_multi_tag_file(path)
+
+ with TsFileDataFrame(str(path), show_progress=False) as tsdf:
+ assert tsdf.get_device_count("weather") == 2
+ assert tsdf.get_device_node_counts("weather") == {
+ ("beijing", "device_a"): 2,
+ ("shanghai", "device_b"): 2,
+ }
+ assert tsdf.get_device_point_counts("weather") == {
+ ("beijing", "device_a"): 4,
+ ("shanghai", "device_b"): 4,
+ }
+ assert tsdf.get_device_point_count("weather", ("beijing", "device_a")) == 4
+ assert (
+ tsdf.get_device_stats("weather", {"city": "beijing", "device": "device_a"})[
+ "point_count"
+ ]
+ == 4
+ )
+ assert tsdf.get_device_statistics("weather") == {
+ ("beijing", "device_a"): {
+ "node_count": 2,
+ "point_count": 4,
+ "value_count": 4,
+ "start_time": 0,
+ "end_time": 1,
+ },
+ ("shanghai", "device_b"): {
+ "node_count": 2,
+ "point_count": 4,
+ "value_count": 4,
+ "start_time": 0,
+ "end_time": 1,
+ },
+ }
+
+ device_metadata = tsdf.list_device_metadata("weather")
+ assert list(device_metadata.columns) == [
+ "table",
+ "node_count",
+ "point_count",
+ "value_count",
+ "start_time",
+ "end_time",
+ "city",
+ "device",
+ ]
+ assert device_metadata[
+ ["city", "device", "node_count", "point_count", "value_count"]
+ ].to_dict("records") == [
+ {
+ "city": "beijing",
+ "device": "device_a",
+ "node_count": 2,
+ "point_count": 4,
+ "value_count": 4,
+ },
+ {
+ "city": "shanghai",
+ "device": "device_b",
+ "node_count": 2,
+ "point_count": 4,
+ "value_count": 4,
+ },
+ ]
+
+ subset = tsdf[[0, 2]]
+ assert subset.get_device_count("weather") == 2
+ assert subset.get_device_node_counts("weather") == {
+ ("beijing", "device_a"): 1,
+ ("shanghai", "device_b"): 1,
+ }
+ assert subset.get_device_point_counts("weather") == {
+ ("beijing", "device_a"): 2,
+ ("shanghai", "device_b"): 2,
+ }
+
+ assert tsdf.get_device_count("missing") == 0
+ assert tsdf.get_device_node_counts("missing") == {}
+
+ numeric_path = tmp_path / "numeric_and_text.tsfile"
+ _write_numeric_and_text_file(numeric_path)
+ with TsFileDataFrame(str(numeric_path), show_progress=False) as tsdf:
+ # The NaN row is part of the returned time series timeline, but is not
+ # counted as a non-null value by native value statistics.
+ assert tsdf.get_device_statistics("weather") == {
+ ("device_a",): {
+ "node_count": 1,
+ "point_count": 3,
+ "value_count": 2,
+ "start_time": 0,
+ "end_time": 2,
+ }
+ }
+
+
def test_dataset_loc_aligns_timestamp_union_and_preserves_requested_order(tmp_path):
path = tmp_path / "weather_sparse.tsfile"
_write_weather_rows_file(
@@ -1479,6 +1687,29 @@ def test_dataset_tree_model_metadata_and_repr(tmp_path):
with pytest.raises(KeyError):
tsdf["table"]
+ assert tsdf.get_device_count("root") == 2
+ assert tsdf.get_device_node_counts("root") == {
+ ("ln", "wf01", "wt01"): 2,
+ ("ln", "wf02", "wt02"): 1,
+ }
+ assert tsdf.get_device_point_counts("root") == {
+ ("ln", "wf01", "wt01"): 10,
+ ("ln", "wf02", "wt02"): 5,
+ }
+ assert tsdf.get_device_point_count("root", ("ln", "wf02", "wt02")) == 5
+ device_metadata = tsdf.list_device_metadata("root")
+ assert list(device_metadata.columns) == [
+ "node_count",
+ "point_count",
+ "value_count",
+ "start_time",
+ "end_time",
+ "_col_1",
+ "_col_2",
+ "_col_3",
+ ]
+ assert device_metadata["node_count"].tolist() == [2, 1]
+
def test_dataset_tree_model_series_access(tmp_path):
path = tmp_path / "tree.tsfile"
diff --git a/python/tsfile/dataset/dataframe.py b/python/tsfile/dataset/dataframe.py
index a65b231bc..0e61f1e15 100644
--- a/python/tsfile/dataset/dataframe.py
+++ b/python/tsfile/dataset/dataframe.py
@@ -19,11 +19,14 @@
"""Top-level dataset accessors for TsFile shards."""
from collections import defaultdict
+import copy
from dataclasses import dataclass, field
import heapq
+import json
import os
import sys
-from typing import Dict, List, Optional, Tuple, Union
+import tempfile
+from typing import Any, Dict, List, Optional, Tuple, Union
import warnings
import numpy as np
@@ -55,6 +58,56 @@ _DATACLASS_SLOTS = {"slots": True} if sys.version_info >= (3, 10) else {}
# multiple shards.
_OVERLAP_ROW_CHUNK_SIZE = 256
+_COVARIATE_ROLE = "C"
+_TARGET_ROLE = "T"
+_DESCRIPTION_ROLES = {_COVARIATE_ROLE, _TARGET_ROLE}
+
+
+def _load_dataframe_description(description_path: Optional[str]) -> dict:
+ """Load a dataframe description JSON object.
+
+ A missing optional description file is treated as an empty description so
+ callers can create it later through :meth:`TsFileDataFrame.set`.
+ """
+ if description_path is None:
+ return {}
+
+ try:
+ with open(description_path, "r", encoding="utf-8") as description_file:
+ description = json.load(description_file)
+ except FileNotFoundError:
+ return {}
+ except json.JSONDecodeError as exc:
+ raise ValueError(
+ f"Invalid dataframe description JSON in '{description_path}': {exc.msg}"
+ ) from exc
+
+ if not isinstance(description, dict):
+ raise ValueError(
+ f"Dataframe description in '{description_path}' must be a JSON object"
+ )
+ return description
+
+
+def _write_dataframe_description(description_path: str, description: dict) -> None:
+ """Atomically write a dataframe description JSON object to disk."""
+ directory = os.path.dirname(description_path) or "."
+ os.makedirs(directory, exist_ok=True)
+ fd, temporary_path = tempfile.mkstemp(
+ prefix=".tsfile-description-", suffix=".tmp", dir=directory
+ )
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8") as description_file:
+ json.dump(description, description_file, ensure_ascii=False, indent=2)
+ description_file.write("\n")
+ os.replace(temporary_path, description_path)
+ except Exception:
+ try:
+ os.unlink(temporary_path)
+ except FileNotFoundError:
+ pass
+ raise
+
@dataclass(**_DATACLASS_SLOTS)
class _DataFrameCatalog:
@@ -618,11 +671,37 @@ class _LocIndexer:
class TsFileDataFrame:
- """Lazy-loaded unified numeric dataset view over multiple TsFile shards."""
+ """Lazy-loaded unified numeric dataset view over multiple TsFile shards.
+
+ ``description_path`` optionally points to a JSON object describing the
+ columns in each table. A description uses the following shape::
+
+ {
+ "weather": {
+ "temperature": "C",
+ "humidity": "T"
+ }
+ }
- def __init__(self, paths: Union[str, List[str]], show_progress: bool = True):
+ ``C`` marks a covariate column and ``T`` marks a target column. The
+ description is independent of the TsFile data and is shared by dataframe
+ subsets created through slicing or boolean selection.
+ """
+
+ def __init__(
+ self,
+ paths: Union[str, List[str]],
+ show_progress: bool = True,
+ description_path: Optional[Union[str, os.PathLike]] = None,
+ ):
self._paths = _expand_paths(paths)
self._show_progress = show_progress
+ self._description_path = (
+ os.path.abspath(os.fspath(description_path))
+ if description_path is not None
+ else None
+ )
+ self._description = _load_dataframe_description(self._description_path)
self._readers: Dict[str, object] = {}
self._index = _DataFrameCatalog()
self._is_view = False
@@ -640,6 +719,8 @@ class TsFileDataFrame:
obj._is_view = True
obj._paths = parent._paths
obj._show_progress = parent._show_progress
+ obj._description_path = parent._description_path
+ obj._description = parent._description
obj._readers = parent._readers
# Reuse the parent's full mapping but restrict the membership scope to
# the requested subset.
@@ -661,6 +742,88 @@ class TsFileDataFrame:
def _owner(self) -> "TsFileDataFrame":
return self._root if self._is_view else self
+ def get(self, key: str, default: Any = None) -> Any:
+ """Return a top-level value from the JSON description.
+
+ The returned value is copied, so mutating a dictionary or list from a
+ ``get`` call cannot change the in-memory description without going
+ through :meth:`set`.
+ """
+ if not isinstance(key, str):
+ raise TypeError(f"Description key must be a string, got {type(key)}")
+ owner = self._owner()
+ return copy.deepcopy(owner._description.get(key, default))
+
+ def set(self, key: str, value: Any) -> None:
+ """Set a top-level JSON description value and persist it if configured.
+
+ ``value`` must be JSON serializable. If no ``description_path`` was
+ supplied, the update is kept in memory for the lifetime of the root
+ dataframe and can still be read through :meth:`get` or the role APIs.
+ """
+ if not isinstance(key, str):
+ raise TypeError(f"Description key must be a string, got {type(key)}")
+
+ owner = self._owner()
+ updated_description = copy.deepcopy(owner._description)
+ updated_description[key] = copy.deepcopy(value)
+
+ try:
+ json.dumps(updated_description, ensure_ascii=False)
+ except (TypeError, ValueError) as exc:
+ raise TypeError("Description value must be JSON serializable") from exc
+
+ if owner._description_path is not None:
+ _write_dataframe_description(owner._description_path, updated_description)
+ owner._description = updated_description
+
+ def _get_role_columns(self, table_name: str, role: str) -> List[str]:
+ if not isinstance(table_name, str):
+ raise TypeError(
+ f"Description table name must be a string, got {type(table_name)}"
+ )
+
+ table_description = self._owner()._description.get(table_name)
+ if table_description is None:
+ return []
+ if not isinstance(table_description, dict):
+ raise ValueError(
+ f"Description for table '{table_name}' must be a JSON object"
+ )
+
+ invalid_roles = {
+ column: column_role
+ for column, column_role in table_description.items()
+ if not isinstance(column_role, str) or column_role not in _DESCRIPTION_ROLES
+ }
+ if invalid_roles:
+ column, column_role = next(iter(invalid_roles.items()))
+ raise ValueError(
+ f"Invalid role {column_role!r} for column '{column}' in table "
+ f"'{table_name}'; expected 'C' or 'T'"
+ )
+ return [
+ column
+ for column, column_role in table_description.items()
+ if column_role == role
+ ]
+
+ def get_covariate_columns(self, table_name: str) -> List[str]:
+ """Return the columns marked ``C`` for ``table_name``."""
+ return self._get_role_columns(table_name, _COVARIATE_ROLE)
+
+ def get_target_columns(self, table_name: str) -> List[str]:
+ """Return the columns marked ``T`` for ``table_name``."""
+ return self._get_role_columns(table_name, _TARGET_ROLE)
+
+ def get_covariate_column_count(self, table_name: str) -> int:
+ """Return the number of columns marked ``C`` for ``table_name``."""
+ return len(self.get_covariate_columns(table_name))
+
+ def get_target_column_count(self, table_name: str) -> int:
+ """Return the number of columns marked ``T`` for ``table_name``."""
+ return len(self.get_target_columns(table_name))
+
def _assert_open(self):
if self._owner()._closed:
raise RuntimeError("Current TsFileDataFrame is closed.")
@@ -826,6 +989,207 @@ class TsFileDataFrame:
def model(self) -> str:
return self._index.model
+ def _get_device_node_counts_by_index(self, table_name: str) -> Dict[int, int]:
+ """Count unique physical series per device in the current dataframe view."""
+ if not isinstance(table_name, str):
+ raise TypeError(f"Table name must be a string, got {type(table_name)}")
+
+ node_indices_by_device: Dict[int, set] = {}
+ seen_series = set()
+ for device_idx, field_idx in self._index.series:
+ device_table_name, _ = self._index.devices[device_idx]
+ if device_table_name != table_name:
+ continue
+
+ series_ref = (device_idx, field_idx)
+ if series_ref in seen_series:
+ continue
+ seen_series.add(series_ref)
+ node_indices_by_device.setdefault(device_idx, set()).add(field_idx)
+
+ return {
+ device_idx: len(node_indices)
+ for device_idx, node_indices in node_indices_by_device.items()
+ }
+
+ def _get_device_statistics_by_index(self, table_name: str) -> Dict[int, dict]:
+ """Aggregate physical-series statistics per device in the current view."""
+ node_counts_by_device = self._get_device_node_counts_by_index(table_name)
+ point_counts_by_device: Dict[int, int] = {}
+ value_counts_by_device: Dict[int, int] = {}
+ seen_series = set()
+ for device_idx, field_idx in self._index.series:
+ if device_idx not in node_counts_by_device:
+ continue
+
+ series_ref = (device_idx, field_idx)
+ if series_ref in seen_series:
+ continue
+ seen_series.add(series_ref)
+
+ for reader, reader_device_id, reader_field_idx in self._index.series_shards[
+ series_ref
+ ]:
+ series_info = reader.get_series_info_by_ref(
+ reader_device_id, reader_field_idx
+ )
+ point_counts_by_device[device_idx] = point_counts_by_device.get(
+ device_idx, 0
+ ) + int(series_info["timeline_length"])
+ value_counts_by_device[device_idx] = value_counts_by_device.get(
+ device_idx, 0
+ ) + int(series_info["length"])
+
+ statistics_by_device = {}
+ for device_idx, node_count in node_counts_by_device.items():
+ if device_idx < len(self._index.device_time_bounds):
+ min_time, max_time = self._index.device_time_bounds[device_idx]
+ else:
+ min_time, max_time = None, None
+ statistics_by_device[device_idx] = {
+ "node_count": node_count,
+ # point_count follows Timeseries.stats['count']: every row on
+ # every physically present node, including NaN rows.
+ "point_count": point_counts_by_device.get(device_idx, 0),
+ # value_count counts only non-null values from native stats.
+ "value_count": value_counts_by_device.get(device_idx, 0),
+ "start_time": min_time,
+ "end_time": max_time,
+ }
+ return statistics_by_device
+
+ def get_device_count(self, table_name: str) -> int:
+ """Return the number of devices represented under ``table_name``.
+
+ A subset dataframe only counts devices that own at least one series in
+ that subset. A table that is not present returns ``0``.
+ """
+ return len(self._get_device_node_counts_by_index(table_name))
+
+ def get_device_node_counts(self, table_name: str) -> Dict[Tuple[Any, ...], int]:
+ """Return ``{device_tag_values: node_count}`` for ``table_name``.
+
+ Each key is the device's ordered tag-value tuple. For tree-model files,
+ it is the tuple of device path segments after the root. ``node_count``
+ is the number of unique, physically present numeric time series for the
+ device in the current dataframe view.
+ """
+ counts_by_index = self._get_device_node_counts_by_index(table_name)
+ return {
+ self._index.devices[device_idx][1]: counts_by_index[device_idx]
+ for device_idx in range(len(self._index.devices))
+ if device_idx in counts_by_index
+ }
+
+ def get_device_statistics(self, table_name: str) -> Dict[Tuple[Any, ...], dict]:
+ """Return per-device counts and time bounds for ``table_name``.
+
+ Each value contains ``node_count``, ``point_count`` (the sum of the
+ per-node timeline counts), ``value_count`` (the sum of non-null values),
+ ``start_time``, and ``end_time``. The key is the device's ordered
+ tag-value tuple.
+ """
+ statistics_by_index = self._get_device_statistics_by_index(table_name)
+ return {
+ self._index.devices[device_idx][1]: stats
+ for device_idx, stats in statistics_by_index.items()
+ }
+
+ def get_device_point_counts(self, table_name: str) -> Dict[Tuple[Any, ...], int]:
+ """Return ``{device_tag_values: point_count}`` for ``table_name``."""
+ return {
+ tag_values: stats["point_count"]
+ for tag_values, stats in self.get_device_statistics(table_name).items()
+ }
+
+ def _resolve_device_index(self, table_name: str, device) -> int:
+ """Resolve a public device selector to a dataframe-local device index."""
+ if not isinstance(table_name, str):
+ raise TypeError(f"Table name must be a string, got {type(table_name)}")
+ table_entry = self._index.table_entries.get(table_name)
+ if table_entry is None:
+ raise KeyError(f"Table not found: '{table_name}'")
+
+ if isinstance(device, SeriesPath):
+ if device.table != table_name:
+ raise KeyError(
+ f"Device {device!r} does not belong to table '{table_name}'"
+ )
+ tag_values = device.tags
+ elif isinstance(device, dict):
+ tag_values = tuple(device.get(column) for column in table_entry.tag_columns)
+ elif isinstance(device, (tuple, list)):
+ tag_values = tuple(device)
+ elif len(table_entry.tag_columns) == 1:
+ tag_values = (device,)
+ else:
+ raise TypeError(
+ "A device with multiple tag columns must be identified by an "
+ "ordered tuple/list or a tag-value dictionary"
+ )
+
+ device_idx = self._index.device_index.get(
+ (table_name, _normalize_tag_values(tag_values))
+ )
+ if device_idx is None:
+ raise KeyError(f"Device not found in table '{table_name}': {device!r}")
+ if device_idx not in self._get_device_node_counts_by_index(table_name):
+ raise KeyError(
+ f"Device is not represented in the current dataframe view: {device!r}"
+ )
+ return device_idx
+
+ def get_device_stats(self, table_name: str, device) -> dict:
+ """Return statistics for one device selected by tags or ``SeriesPath``."""
+ device_idx = self._resolve_device_index(table_name, device)
+ return self._get_device_statistics_by_index(table_name)[device_idx].copy()
+
+ def get_device_point_count(self, table_name: str, device) -> int:
+ """Return the timeline point count for one selected device."""
+ return self.get_device_stats(table_name, device)["point_count"]
+
+ def list_device_metadata(self, table_name: str):
+ """Return a pandas DataFrame with one row and statistics per device.
+
+ Declared tag columns are expanded into named columns. Short tag tuples
+ are padded with ``None`` so nullable table tags and tree devices of
+ different depths remain position preserving.
+ """
+ import pandas as pd
+
+ statistics_by_index = self._get_device_statistics_by_index(table_name)
+ table_entry = self._index.table_entries.get(table_name)
+ tag_columns = list(table_entry.tag_columns) if table_entry else []
+ leading_columns = [
+ "node_count",
+ "point_count",
+ "value_count",
+ "start_time",
+ "end_time",
+ ]
+ if self._index.model != MODEL_TREE:
+ leading_columns.insert(0, "table")
+ columns = leading_columns + tag_columns
+
+ rows = []
+ for device_idx in range(len(self._index.devices)):
+ if device_idx not in statistics_by_index:
+ continue
+ _, tag_values = self._index.devices[device_idx]
+ ordered_tag_values = list(tag_values)
+ if len(ordered_tag_values) < len(tag_columns):
+ ordered_tag_values.extend(
+ [None] * (len(tag_columns) - len(ordered_tag_values))
+ )
+
+ row = statistics_by_index[device_idx].copy()
+ if self._index.model != MODEL_TREE:
+ row["table"] = table_name
+ row.update(zip(tag_columns, ordered_tag_values))
+ rows.append(row)
+
+ return pd.DataFrame(rows, columns=columns)
+
def list_timeseries(self, path_prefix: str = "") -> List[SeriesPath]:
if not path_prefix:
return [