Coverage for foxplot / fox.py: 99%
93 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-15 10:53 +0000
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3#
4# SPDX-License-Identifier: Apache-2.0
6"""The :class:`Fox` class is where we manipulate dictionary-series data."""
8import logging
9from pathlib import PosixPath
10from typing import Dict, List, Optional, Union
12import numpy as np
13import uplot
14from numpy.typing import NDArray
15from uplot.plot2 import add_series as _uplot_add_series
16from uplot.plot2 import prepare_data as _uplot_prepare_data
17from uplot.utils import js as _uplot_js
19from .decode import decode
20from .node import Node
21from .series import Series
23_INTEGER_VALUE_FMT = _uplot_js(
24 "(self, rawValue) => {"
25 "if (rawValue === null) return '--';"
26 "const v = rawValue;"
27 "const av = Math.abs(v);"
28 "if (av >= 1e9) return (v / 1e9).toFixed(0) + 'B';"
29 "if (av >= 1e6) return (v / 1e6).toFixed(0) + 'M';"
30 "if (av >= 1e3) return (v / 1e3).toFixed(0) + 'k';"
31 "return String(v);}"
32)
35def _is_integer_valued(values: NDArray[np.float64]) -> bool:
36 finite = values[np.isfinite(values)]
37 return len(finite) > 0 and bool(np.all(finite == np.floor(finite)))
40class Fox:
41 """Frequent Observation diXionaries, our main class.
43 Our main class to read, access and manipulate series of dictionary data.
44 """
46 __source: Union[str, PosixPath]
47 __times: Optional[NDArray[np.float64]]
48 data: Node
49 length: int
51 @staticmethod
52 def empty() -> "Fox":
53 """Initialize from empty time series."""
54 return Fox(filename=None)
56 def __init__(self, filename: Union[str, PosixPath, None]) -> None:
57 """Initialize time series.
59 Args:
60 filename: Name (e.g. "stdin") or path of file to read time series
61 from, or ``None`` to start from an empty state.
62 """
63 self.__source = filename or "custom data"
64 self.__times = None
65 self.data = Node("/")
66 self.length = 0
67 if filename is not None:
68 for unpacked in decode(filename):
69 self.unpack(unpacked)
70 self.data._freeze(self.length)
72 def __list_to_dict(
73 self, series_list: List[Union[Series, Node]]
74 ) -> Dict[str, NDArray[np.float64]]:
75 """Convert a list of series (or nodes) to a dictionary.
77 The output dictionary has one key per series in the list. Nodes are
78 expanded just once, assuming all their children in the data tree
79 are series.
81 Args:
82 series_list: Input list of series;
84 Returns:
85 Dictionary mapping series names to their values.
86 """
87 series_dict = {}
88 for series in series_list:
89 if isinstance(series, Series):
90 series_dict[series._label] = series._values
91 elif isinstance(series, Node):
92 for key, child in series._items():
93 label = series._label + f"/{key}"
94 if isinstance(child, Series):
95 series_dict[label] = child._values
96 else:
97 logging.warning(
98 "Skipping '%s' as it is not an indexed series",
99 label,
100 )
101 else:
102 raise TypeError(
103 f"Series '{series}' has unhandled type {type(series)}"
104 )
105 return series_dict
107 def detect_time(self) -> None:
108 """Search for a time key in root keys."""
109 candidates = ("time", "timestamp")
110 for key in candidates:
111 if key in self.data.__dict__:
112 self.set_time(self.data.__dict__[key])
113 print(
114 f'Detected "{key}" as time key from the input '
115 "(call `fox.set_time` to select a different one)"
116 )
117 return
119 def get_series(self, label: str) -> Series:
120 """Get time-series data from a given label.
122 Args:
123 label: Label to the data in input dictionaries, for example
124 ``/observation/cpu_temperature``.
126 Returns:
127 Corresponding time series.
128 """
129 keys = label.strip("/").split("/")
130 series = self.data._get_child(keys)
131 if not isinstance(series, Series):
132 raise TypeError(f"Series {label} is not finalized")
133 return series
135 def plot(
136 self,
137 left: Union[Series, Node, List[Union[Series, Node]]],
138 right: Optional[Union[Series, Node, List[Union[Series, Node]]]] = None,
139 title: Optional[str] = None,
140 ) -> None:
141 """Plot a set of indexed series.
143 Args:
144 left: Series to plot on the left axis.
145 right: Series to plot on the right axis.
146 title: Plot title.
147 """
148 if isinstance(left, (Node, Series)):
149 left = [left]
150 if isinstance(right, (Node, Series)):
151 right = [right]
152 if title is None:
153 title = f"Plot from {self.__source}"
155 times: NDArray[np.float64] = (
156 self.__times
157 if self.__times is not None
158 else np.array(range(self.length), dtype=np.float64)
159 )
161 left_series: Dict[str, NDArray[np.float64]] = self.__list_to_dict(left)
162 right_series: Dict[str, NDArray[np.float64]] = {}
163 if right is not None:
164 right_series = self.__list_to_dict(right)
166 left_values = list(left_series.values())
167 right_values = list(right_series.values())
168 data = _uplot_prepare_data(times, left_values, right_values)
169 series_opts: Dict = {}
170 _uplot_add_series(
171 series_opts,
172 data,
173 len(left_series),
174 list(left_series.keys()),
175 list(right_series.keys()),
176 )
177 for i, values in enumerate(left_values + right_values):
178 if _is_integer_valued(values):
179 series_opts["series"][i + 1]["value"] = _INTEGER_VALUE_FMT
181 uplot.plot2(
182 times,
183 left_values,
184 right_values,
185 title=title,
186 timestamped=self.__times is not None,
187 series=series_opts["series"],
188 )
190 def unpack(self, unpacked: dict) -> None:
191 """Append data from an unpacked dictionary.
193 Args:
194 unpacked: Unpacked dictionary.
195 """
196 self.data._update(self.length, unpacked)
197 self.length += 1
199 def set_time(self, time: Series):
200 """Set label of time index in input dictionaries.
202 Args:
203 time: Time index as a series.
204 """
205 time._values = time._values.astype(np.float64)
206 self.__times = time._values
208 def set_series_times(series: Union[Series, Node]):
209 if isinstance(series, Series):
210 series._times = self.__times
211 if isinstance(series, Node):
212 for _, child in series._items():
213 set_series_times(child)
215 set_series_times(self.data)