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 598d4a5cabbde9b170d6de1eb96b94e0145bfcf6 Author: ColinLee <[email protected]> AuthorDate: Fri Aug 7 17:25:20 2026 +0800 Add dataframe descriptions and device statistics --- python/README-zh.md | 42 +++- python/README.md | 44 ++++- python/tests/test_tsfile_dataset.py | 231 ++++++++++++++++++++++ python/tsfile/dataset/dataframe.py | 370 +++++++++++++++++++++++++++++++++++- 4 files changed, 682 insertions(+), 5 deletions(-) diff --git a/python/README-zh.md b/python/README-zh.md index dd3a59ea4..672ec1342 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` 只统计非空值。 + --- ## 如何贡献 @@ -67,7 +108,6 @@ mvn -P with-cpp,with-python clean verify ```sh python setup.py build_ext --inplace ``` - ## 文件级 Properties `TsFileWriter` 和 `TsFileTableWriter` 可以在打开期间写入二进制 property。 diff --git a/python/README.md b/python/README.md index 8e2716a2c..424006e57 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,7 +103,6 @@ Build by python command: ```sh python setup.py build_ext --inplace ``` - ## File-level properties `TsFileWriter` and `TsFileTableWriter` accept binary properties while they are diff --git a/python/tests/test_tsfile_dataset.py b/python/tests/test_tsfile_dataset.py index ee016d087..3e42f9061 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 [
