You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
354 lines
13 KiB
354 lines
13 KiB
import os
|
|
import re
|
|
import uuid
|
|
import threading
|
|
from collections import deque
|
|
import dearpygui.dearpygui as dpg
|
|
from abc import ABC, abstractmethod
|
|
from openpilot.tools.jotpluggler.data import DataManager
|
|
|
|
|
|
class ViewPanel(ABC):
|
|
"""Abstract base class for all view panels that can be displayed in a plot container"""
|
|
|
|
def __init__(self, panel_id: str = None):
|
|
self.panel_id = panel_id or str(uuid.uuid4())
|
|
self.title = "Untitled Panel"
|
|
|
|
@abstractmethod
|
|
def create_ui(self, parent_tag: str):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def destroy_ui(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_panel_type(self) -> str:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def preserve_data(self):
|
|
pass
|
|
|
|
|
|
class TimeSeriesPanel(ViewPanel):
|
|
def __init__(self, data_manager: DataManager, playback_manager, panel_id: str | None = None):
|
|
super().__init__(panel_id)
|
|
self.data_manager = data_manager
|
|
self.playback_manager = playback_manager
|
|
self.title = "Time Series Plot"
|
|
self.plotted_series: set[str] = set()
|
|
self.plot_tag: str | None = None
|
|
self.x_axis_tag: str | None = None
|
|
self.y_axis_tag: str | None = None
|
|
self.timeline_indicator_tag: str | None = None
|
|
self._ui_created = False
|
|
self._preserved_series_data: list[tuple[str, tuple]] = [] # TODO: the way we do this right now doesn't make much sense
|
|
self._series_legend_tags: dict[str, str] = {} # Maps series_path to legend tag
|
|
self.data_manager.add_observer(self.on_data_loaded)
|
|
|
|
def preserve_data(self):
|
|
self._preserved_series_data = []
|
|
if self.plotted_series and self._ui_created:
|
|
for series_path in self.plotted_series:
|
|
time_value_data = self.data_manager.get_timeseries(series_path)
|
|
if time_value_data:
|
|
self._preserved_series_data.append((series_path, time_value_data))
|
|
|
|
def create_ui(self, parent_tag: str):
|
|
self.plot_tag = f"plot_{self.panel_id}"
|
|
self.x_axis_tag = f"{self.plot_tag}_x_axis"
|
|
self.y_axis_tag = f"{self.plot_tag}_y_axis"
|
|
self.timeline_indicator_tag = f"{self.plot_tag}_timeline"
|
|
|
|
with dpg.plot(height=-1, width=-1, tag=self.plot_tag, parent=parent_tag, drop_callback=self._on_series_drop, payload_type="TIMESERIES_PAYLOAD"):
|
|
dpg.add_plot_legend()
|
|
dpg.add_plot_axis(dpg.mvXAxis, label="", tag=self.x_axis_tag)
|
|
dpg.add_plot_axis(dpg.mvYAxis, label="", tag=self.y_axis_tag)
|
|
|
|
timeline_series_tag = dpg.add_inf_line_series(x=[0], label="Timeline", parent=self.y_axis_tag, tag=self.timeline_indicator_tag)
|
|
dpg.bind_item_theme(timeline_series_tag, "global_timeline_theme")
|
|
|
|
# Restore series from preserved data
|
|
if self._preserved_series_data:
|
|
self.plotted_series.clear()
|
|
for series_path, (rel_time_array, value_array) in self._preserved_series_data:
|
|
self._add_series_with_data(series_path, rel_time_array, value_array)
|
|
self._preserved_series_data = []
|
|
|
|
self._ui_created = True
|
|
|
|
def update_timeline_indicator(self, current_time_s: float):
|
|
if not self._ui_created or not dpg.does_item_exist(self.timeline_indicator_tag):
|
|
return
|
|
|
|
dpg.set_value(self.timeline_indicator_tag, [[current_time_s], [0]]) # vertical line position
|
|
|
|
if self.plotted_series: # update legend labels with current values
|
|
for series_path in self.plotted_series:
|
|
value = self.data_manager.get_value_at(series_path, current_time_s)
|
|
|
|
if value is not None:
|
|
if isinstance(value, (int, float)):
|
|
if isinstance(value, float):
|
|
formatted_value = f"{value:.4f}" if abs(value) < 1000 else f"{value:.3e}"
|
|
else:
|
|
formatted_value = str(value)
|
|
else:
|
|
formatted_value = str(value)
|
|
|
|
series_tag = f"series_{self.panel_id}_{series_path.replace('/', '_')}"
|
|
legend_label = f"{series_path}: {formatted_value}"
|
|
|
|
if dpg.does_item_exist(series_tag):
|
|
dpg.configure_item(series_tag, label=legend_label)
|
|
|
|
def _add_series_with_data(self, series_path: str, rel_time_array, value_array) -> bool:
|
|
if series_path in self.plotted_series:
|
|
return False
|
|
|
|
series_tag = f"series_{self.panel_id}_{series_path.replace('/', '_')}"
|
|
line_series_tag = dpg.add_line_series(x=rel_time_array.tolist(), y=value_array.tolist(), label=series_path, parent=self.y_axis_tag, tag=series_tag)
|
|
|
|
dpg.bind_item_theme(line_series_tag, "global_line_theme")
|
|
|
|
self.plotted_series.add(series_path)
|
|
dpg.fit_axis_data(self.x_axis_tag)
|
|
dpg.fit_axis_data(self.y_axis_tag)
|
|
return True
|
|
|
|
def destroy_ui(self):
|
|
if self.plot_tag and dpg.does_item_exist(self.plot_tag):
|
|
dpg.delete_item(self.plot_tag)
|
|
|
|
self._series_legend_tags.clear()
|
|
self._ui_created = False
|
|
|
|
def get_panel_type(self) -> str:
|
|
return "timeseries"
|
|
|
|
def add_series(self, series_path: str) -> bool:
|
|
if series_path in self.plotted_series:
|
|
return False
|
|
|
|
time_value_data = self.data_manager.get_timeseries(series_path)
|
|
if time_value_data is None:
|
|
return False
|
|
|
|
rel_time_array, value_array = time_value_data
|
|
return self._add_series_with_data(series_path, rel_time_array, value_array)
|
|
|
|
def clear_all_series(self):
|
|
for series_path in self.plotted_series.copy():
|
|
self.remove_series(series_path)
|
|
|
|
def remove_series(self, series_path: str):
|
|
if series_path in self.plotted_series:
|
|
series_tag = f"series_{self.panel_id}_{series_path.replace('/', '_')}"
|
|
if dpg.does_item_exist(series_tag):
|
|
dpg.delete_item(series_tag)
|
|
self.plotted_series.remove(series_path)
|
|
if series_path in self._series_legend_tags:
|
|
del self._series_legend_tags[series_path]
|
|
|
|
def on_data_loaded(self, data: dict):
|
|
for series_path in self.plotted_series.copy():
|
|
self._update_series_data(series_path)
|
|
|
|
def _update_series_data(self, series_path: str) -> bool:
|
|
time_value_data = self.data_manager.get_timeseries(series_path)
|
|
if time_value_data is None:
|
|
return False
|
|
|
|
rel_time_array, value_array = time_value_data
|
|
series_tag = f"series_{self.panel_id}_{series_path.replace('/', '_')}"
|
|
|
|
if dpg.does_item_exist(series_tag):
|
|
dpg.set_value(series_tag, [rel_time_array.tolist(), value_array.tolist()])
|
|
dpg.fit_axis_data(self.x_axis_tag)
|
|
dpg.fit_axis_data(self.y_axis_tag)
|
|
return True
|
|
else:
|
|
self.plotted_series.discard(series_path)
|
|
return False
|
|
|
|
def _on_series_drop(self, sender, app_data, user_data):
|
|
series_path = app_data
|
|
self.add_series(series_path)
|
|
|
|
|
|
class DataTreeNode:
|
|
def __init__(self, name: str, full_path: str = ""):
|
|
self.name = name
|
|
self.full_path = full_path
|
|
self.children: dict[str, DataTreeNode] = {}
|
|
self.is_leaf = False
|
|
self.child_count = 0
|
|
self.is_plottable_cached = None
|
|
self.ui_created = False
|
|
self.ui_tag: str | None = None
|
|
|
|
|
|
class DataTreeView:
|
|
MAX_ITEMS_PER_FRAME = 50
|
|
|
|
def __init__(self, data_manager: DataManager, ui_lock: threading.Lock):
|
|
self.data_manager = data_manager
|
|
self.ui_lock = ui_lock
|
|
self.current_search = ""
|
|
self.data_tree = DataTreeNode(name="root")
|
|
self.ui_render_queue: deque[tuple[DataTreeNode, str, str, bool]] = deque() # (node, parent_tag, search_term, is_leaf)
|
|
self.visible_expanded_nodes: set[str] = set()
|
|
self.created_leaf_paths: set[str] = set()
|
|
self._all_paths_cache: list[str] = []
|
|
self.data_manager.add_observer(self._on_data_loaded)
|
|
|
|
def _on_data_loaded(self, data: dict):
|
|
if data.get('loading_complete'):
|
|
with self.ui_lock:
|
|
self._all_paths_cache = self.data_manager.get_all_paths()
|
|
self._populate_tree()
|
|
|
|
def update_frame(self):
|
|
items_processed = 0
|
|
while self.ui_render_queue and items_processed < self.MAX_ITEMS_PER_FRAME: # process up to MAX_ITEMS_PER_FRAME to maintain perforamnce
|
|
node, parent_tag, search_term, is_leaf = self.ui_render_queue.popleft()
|
|
if is_leaf:
|
|
self._create_leaf_ui(node, parent_tag)
|
|
else:
|
|
self._create_node_ui(node, parent_tag, search_term)
|
|
items_processed += 1
|
|
|
|
def search_data(self, search_term: str):
|
|
self.current_search = search_term
|
|
self._all_paths_cache = self.data_manager.get_all_paths()
|
|
self._populate_tree()
|
|
|
|
def _populate_tree(self):
|
|
self._clear_ui()
|
|
search_term = self.current_search.strip().lower()
|
|
self.data_tree = self._build_tree_structure(search_term)
|
|
for child in sorted(self.data_tree.children.values(), key=self._natural_sort_key): # queue top level nodes
|
|
self.ui_render_queue.append((child, "data_tree_container", search_term, child.is_leaf))
|
|
|
|
def _clear_ui(self):
|
|
dpg.delete_item("data_tree_container", children_only=True)
|
|
self.ui_render_queue.clear()
|
|
self.visible_expanded_nodes.clear()
|
|
self.created_leaf_paths.clear()
|
|
|
|
def _build_tree_structure(self, search_term: str) -> DataTreeNode:
|
|
root = DataTreeNode(name="root")
|
|
for path in sorted(self._all_paths_cache):
|
|
if not self._should_show_path(path, search_term):
|
|
continue
|
|
|
|
parts = path.split('/')
|
|
current_node = root
|
|
current_path_prefix = ""
|
|
|
|
for part in parts:
|
|
current_path_prefix = f"{current_path_prefix}/{part}" if current_path_prefix else part
|
|
if part not in current_node.children:
|
|
current_node.children[part] = DataTreeNode(name=part, full_path=current_path_prefix)
|
|
current_node = current_node.children[part]
|
|
current_node.is_leaf = True
|
|
self._calculate_child_counts(root)
|
|
return root
|
|
|
|
def _calculate_child_counts(self, node: DataTreeNode):
|
|
if node.is_leaf:
|
|
node.child_count = 0
|
|
else:
|
|
node.child_count = len(node.children)
|
|
for child in node.children.values():
|
|
self._calculate_child_counts(child)
|
|
|
|
def _create_node_ui(self, node: DataTreeNode, parent_tag: str, search_term: str):
|
|
if node.is_leaf:
|
|
self._create_leaf_ui(node, parent_tag)
|
|
else:
|
|
self._create_tree_node_ui(node, parent_tag, search_term)
|
|
|
|
def _create_tree_node_ui(self, node: DataTreeNode, parent_tag: str, search_term: str):
|
|
node_tag = f"tree_{node.full_path}"
|
|
node.ui_tag = node_tag
|
|
|
|
label = f"{node.name} ({node.child_count} fields)"
|
|
should_open = (bool(search_term) and len(search_term) > 1 and any(search_term in path for path in self._get_descendant_paths(node)))
|
|
|
|
with dpg.tree_node(label=label, parent=parent_tag, tag=node_tag, default_open=should_open, open_on_arrow=True, open_on_double_click=True) as tree_node:
|
|
with dpg.item_handler_registry() as handler:
|
|
dpg.add_item_toggled_open_handler(callback=lambda s, d, u: self._on_node_expanded(node, search_term))
|
|
dpg.bind_item_handler_registry(tree_node, handler)
|
|
|
|
node.ui_created = True
|
|
|
|
if should_open:
|
|
self.visible_expanded_nodes.add(node.full_path)
|
|
self._queue_children(node, node_tag, search_term)
|
|
|
|
def _create_leaf_ui(self, node: DataTreeNode, parent_tag: str):
|
|
half_split_size = dpg.get_item_rect_size("data_pool_window")[0] // 2
|
|
with dpg.group(parent=parent_tag, horizontal=True, xoffset=half_split_size, tag=f"group_{node.full_path}") as draggable_group:
|
|
dpg.add_text(node.name)
|
|
dpg.add_text("N/A", tag=f"value_{node.full_path}")
|
|
|
|
if node.is_plottable_cached is None:
|
|
node.is_plottable_cached = self.data_manager.is_plottable(node.full_path)
|
|
|
|
if node.is_plottable_cached:
|
|
with dpg.drag_payload(parent=draggable_group, drag_data=node.full_path,
|
|
payload_type="TIMESERIES_PAYLOAD"):
|
|
dpg.add_text(f"Plot: {node.full_path}")
|
|
|
|
node.ui_created = True
|
|
node.ui_tag = f"value_{node.full_path}"
|
|
self.created_leaf_paths.add(node.full_path)
|
|
|
|
def _queue_children(self, node: DataTreeNode, parent_tag: str, search_term: str):
|
|
for child in sorted(node.children.values(), key=self._natural_sort_key):
|
|
self.ui_render_queue.append((child, parent_tag, search_term, child.is_leaf))
|
|
|
|
def _on_node_expanded(self, node: DataTreeNode, search_term: str):
|
|
node_tag = f"tree_{node.full_path}"
|
|
if not dpg.does_item_exist(node_tag):
|
|
return
|
|
|
|
is_expanded = dpg.get_value(node_tag)
|
|
|
|
if is_expanded:
|
|
if node.full_path not in self.visible_expanded_nodes:
|
|
self.visible_expanded_nodes.add(node.full_path)
|
|
self._queue_children(node, node_tag, search_term)
|
|
else:
|
|
self.visible_expanded_nodes.discard(node.full_path)
|
|
self._remove_children_from_queue(node.full_path)
|
|
|
|
def _remove_children_from_queue(self, collapsed_node_path: str):
|
|
new_queue = deque()
|
|
for node, parent_tag, search_term, is_leaf in self.ui_render_queue:
|
|
# Keep items that are not children of the collapsed node
|
|
if not node.full_path.startswith(collapsed_node_path + "/"):
|
|
new_queue.append((node, parent_tag, search_term, is_leaf))
|
|
self.ui_render_queue = new_queue
|
|
|
|
def _should_show_path(self, path: str, search_term: str) -> bool:
|
|
if 'DEPRECATED' in path and not os.environ.get('SHOW_DEPRECATED'):
|
|
return False
|
|
return not search_term or search_term in path.lower()
|
|
|
|
def _natural_sort_key(self, node: DataTreeNode):
|
|
node_type_key = node.is_leaf
|
|
parts = [int(p) if p.isdigit() else p.lower() for p in re.split(r'(\d+)', node.name) if p]
|
|
return (node_type_key, parts)
|
|
|
|
def _get_descendant_paths(self, node: DataTreeNode):
|
|
for child_name, child_node in node.children.items():
|
|
child_name_lower = child_name.lower()
|
|
if child_node.is_leaf:
|
|
yield child_name_lower
|
|
else:
|
|
for path in self._get_descendant_paths(child_node):
|
|
yield f"{child_name_lower}/{path}"
|
|
|