from __future__ import annotations
import warnings
from importlib import resources
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Optional
from urllib.parse import quote
from uuid import uuid4
import numpy as np
import pandas as pd
import pyvista as pv
from parq_blockmodel.utils.pyvista.categorical_utils import load_mapping_dict
from parq_blockmodel.utils.pyvista.custom_plotter import CustomPlotter
from parq_blockmodel.visualization.blockmodel_plot import (
BlockModelPlotState,
_bin_to_deciles,
_calculate_deciles,
_plotter_add_mesh_kwargs,
prepare_plot_state,
)
from parq_blockmodel.visualization.asset_selector import HivePbmCatalog
if TYPE_CHECKING:
from parq_blockmodel.blockmodel import ParquetBlockModel
[docs]
@dataclass(slots=True)
class ThresholdRange:
minimum: float
maximum: float
value: float
step: float
[docs]
@dataclass(slots=True)
class DataFilterSlot:
attribute: str = ""
is_categorical: bool = False
minimum: float = 0.0
maximum: float = 1.0
step: float = 0.005
range_values: list[float] = None # type: ignore[assignment]
category_options: list[str] = None # type: ignore[assignment]
selected_categories: list[str] = None # type: ignore[assignment]
preset_min: Optional[float] = None
preset_max: Optional[float] = None
preset_categories: Optional[list[str]] = None
def __post_init__(self) -> None:
if self.range_values is None:
self.range_values = [0.0, 1.0]
if self.category_options is None:
self.category_options = []
if self.selected_categories is None:
self.selected_categories = []
[docs]
class BlockModelTrameApp:
"""Read-only Trame-friendly block-model session.
The app keeps a PBM model separate from the rendered scene state. It is
intentionally backend-first so the same state can be reused by a future
loaded-PBM view or a hive-directory browser.
"""
[docs]
def __init__(
self,
blockmodel: Optional["ParquetBlockModel"] = None,
*,
scalar: Optional[str] = None,
threshold_value: Optional[float] = None,
data_filter_1_attribute: Optional[str] = None,
data_filter_1_min: Optional[float] = None,
data_filter_1_max: Optional[float] = None,
data_filter_1_categories: Optional[list[str]] = None,
data_filter_2_attribute: Optional[str] = None,
data_filter_2_min: Optional[float] = None,
data_filter_2_max: Optional[float] = None,
data_filter_2_categories: Optional[list[str]] = None,
grid_type: str = "image",
frame: str = "world",
title: Optional[str] = None,
show_edges: bool = True,
z_up_lock: bool = False,
z_up_hotkey: str = "z",
app_name: str = "ParquetBlockModel Viewer",
asset_catalog: Optional[HivePbmCatalog] = None,
asset_catalog_root: Optional[str | Path] = None,
) -> None:
self.blockmodel = blockmodel
self.grid_type = grid_type
self.frame = frame
self._title_override = title is not None
self.title = title or (blockmodel.name if blockmodel else "")
self.show_edges = show_edges
self.z_up_lock = bool(z_up_lock)
self.z_up_hotkey = str(z_up_hotkey).strip().lower()
self.app_name = str(app_name)
self._initial_scalar = scalar or (self._default_scalar() if blockmodel is not None else "")
self._initial_threshold_value = (
float(threshold_value) if threshold_value is not None else None
)
self.asset_catalog = asset_catalog
self.asset_catalog_root = (
Path(asset_catalog_root).resolve() if asset_catalog_root is not None else None
)
self._asset_level_keys: list[str] = []
self._asset_level_values: dict[str, str] = {}
self._selected_asset_name = ""
self._registered_asset_level_indices: set[int] = set()
self._asset_name_handler_registered = False
self._asset_selectors_autofilled = False
self._source_mode = "hive" if asset_catalog is not None else "file"
self._source_path = str(self.asset_catalog_root or (self.blockmodel.blockmodel_path if blockmodel else ""))
self._duckdb_query: Optional[str] = None
self._duckdb_path: Optional[str] = None
self._skip_initial_blockmodel_load = False
self.state: Optional[BlockModelPlotState] = None
self.threshold: Optional[ThresholdRange] = None
self.filter_enabled = False
self.show_model_bounds = False
self.colormap = "jet"
self.discretize_deciles = False
self.available_colormaps: list[str] = []
self.plotter = CustomPlotter(off_screen=True)
self._remote_view = None
self._server = None
self._syncing_state = False
self._drawer_default_open = True
self._attribute_data_range: dict[str, tuple[float, float]] = {}
self._filter_attribute_options: list[str] = []
self._filter_attribute_cache: dict[str, np.ndarray] = {}
self._filter_category_code_to_label: dict[str, dict[float, str]] = {}
self._active_display_mesh: Optional[pv.DataSet] = None
self._pick_debug_count = 0
self._pick_debug_last = ""
self._data_filters = [
DataFilterSlot(
attribute=str(data_filter_1_attribute or ""),
preset_min=(float(data_filter_1_min) if data_filter_1_min is not None else None),
preset_max=(float(data_filter_1_max) if data_filter_1_max is not None else None),
preset_categories=(
list(data_filter_1_categories)
if data_filter_1_categories is not None
else None
),
),
DataFilterSlot(
attribute=str(data_filter_2_attribute or ""),
preset_min=float(data_filter_2_min) if data_filter_2_min is not None else None,
preset_max=float(data_filter_2_max) if data_filter_2_max is not None else None,
preset_categories=(
list(data_filter_2_categories)
if data_filter_2_categories is not None
else None
),
),
]
self._view_initialized = False
@classmethod
def from_pbm_file(
cls,
blockmodel_path: str | Path,
*,
scalar: Optional[str] = None,
threshold_value: Optional[float] = None,
data_filter_1_attribute: Optional[str] = None,
data_filter_1_min: Optional[float] = None,
data_filter_1_max: Optional[float] = None,
data_filter_1_categories: Optional[list[str]] = None,
data_filter_2_attribute: Optional[str] = None,
data_filter_2_min: Optional[float] = None,
data_filter_2_max: Optional[float] = None,
data_filter_2_categories: Optional[list[str]] = None,
grid_type: str = "image",
frame: str = "world",
title: Optional[str] = None,
show_edges: bool = True,
z_up_lock: bool = False,
z_up_hotkey: str = "z",
app_name: str = "ParquetBlockModel Viewer",
) -> "BlockModelTrameApp":
from parq_blockmodel.blockmodel import ParquetBlockModel
path = Path(blockmodel_path).expanduser()
if not path.is_absolute():
path = path.resolve()
if not path.exists():
raise FileNotFoundError(f"Path does not exist: {path}")
if not path.is_file() or path.suffix.lower() != ".pbm":
raise ValueError(f"Selected file must be a .pbm file: {path}")
return cls(
ParquetBlockModel(blockmodel_path=path),
scalar=scalar,
threshold_value=threshold_value,
data_filter_1_attribute=data_filter_1_attribute,
data_filter_1_min=data_filter_1_min,
data_filter_1_max=data_filter_1_max,
data_filter_1_categories=data_filter_1_categories,
data_filter_2_attribute=data_filter_2_attribute,
data_filter_2_min=data_filter_2_min,
data_filter_2_max=data_filter_2_max,
data_filter_2_categories=data_filter_2_categories,
grid_type=grid_type,
frame=frame,
title=title,
show_edges=show_edges,
z_up_lock=z_up_lock,
z_up_hotkey=z_up_hotkey,
app_name=app_name,
)
@classmethod
def from_path(
cls,
blockmodel_path: str | Path,
*,
scalar: Optional[str] = None,
threshold_value: Optional[float] = None,
data_filter_1_attribute: Optional[str] = None,
data_filter_1_min: Optional[float] = None,
data_filter_1_max: Optional[float] = None,
data_filter_1_categories: Optional[list[str]] = None,
data_filter_2_attribute: Optional[str] = None,
data_filter_2_min: Optional[float] = None,
data_filter_2_max: Optional[float] = None,
data_filter_2_categories: Optional[list[str]] = None,
grid_type: str = "image",
frame: str = "world",
title: Optional[str] = None,
show_edges: bool = True,
z_up_lock: bool = False,
z_up_hotkey: str = "z",
app_name: str = "ParquetBlockModel Viewer",
) -> "BlockModelTrameApp":
return cls.from_pbm_file(
blockmodel_path,
scalar=scalar,
threshold_value=threshold_value,
data_filter_1_attribute=data_filter_1_attribute,
data_filter_1_min=data_filter_1_min,
data_filter_1_max=data_filter_1_max,
data_filter_1_categories=data_filter_1_categories,
data_filter_2_attribute=data_filter_2_attribute,
data_filter_2_min=data_filter_2_min,
data_filter_2_max=data_filter_2_max,
data_filter_2_categories=data_filter_2_categories,
grid_type=grid_type,
frame=frame,
title=title,
show_edges=show_edges,
z_up_lock=z_up_lock,
z_up_hotkey=z_up_hotkey,
app_name=app_name,
)
@classmethod
def from_hive_directory(
cls,
root_path: str | Path,
*,
scalar: Optional[str] = None,
threshold_value: Optional[float] = None,
data_filter_1_attribute: Optional[str] = None,
data_filter_1_min: Optional[float] = None,
data_filter_1_max: Optional[float] = None,
data_filter_1_categories: Optional[list[str]] = None,
data_filter_2_attribute: Optional[str] = None,
data_filter_2_min: Optional[float] = None,
data_filter_2_max: Optional[float] = None,
data_filter_2_categories: Optional[list[str]] = None,
grid_type: str = "image",
frame: str = "world",
title: Optional[str] = None,
show_edges: bool = True,
z_up_lock: bool = False,
z_up_hotkey: str = "z",
app_name: str = "ParquetBlockModel Viewer",
) -> "BlockModelTrameApp":
root = Path(root_path).expanduser()
if not root.is_absolute():
root = root.resolve()
if not root.exists():
raise FileNotFoundError(f"Path does not exist: {root}")
if not root.is_dir():
raise ValueError(f"Selected path is not a directory: {root}")
catalog = HivePbmCatalog.discover(root)
# For hive mode, don't load any blockmodel initially - user selects via UI
# Start with blockmodel=None to avoid unnecessary loading of first asset
app = cls(
blockmodel=None,
scalar=scalar,
threshold_value=threshold_value,
data_filter_1_attribute=data_filter_1_attribute,
data_filter_1_min=data_filter_1_min,
data_filter_1_max=data_filter_1_max,
data_filter_1_categories=data_filter_1_categories,
data_filter_2_attribute=data_filter_2_attribute,
data_filter_2_min=data_filter_2_min,
data_filter_2_max=data_filter_2_max,
data_filter_2_categories=data_filter_2_categories,
grid_type=grid_type,
frame=frame,
title=title,
show_edges=show_edges,
z_up_lock=z_up_lock,
z_up_hotkey=z_up_hotkey,
app_name=app_name,
asset_catalog=catalog,
asset_catalog_root=root,
)
# Mark that we should skip loading placeholder blockmodel on launch
app._skip_initial_blockmodel_load = True
return app
@classmethod
def from_duckdb_query(
cls,
duckdb_query: str,
*,
duckdb_path: Optional[str | Path] = None,
scalar: Optional[str] = None,
threshold_value: Optional[float] = None,
data_filter_1_attribute: Optional[str] = None,
data_filter_1_min: Optional[float] = None,
data_filter_1_max: Optional[float] = None,
data_filter_1_categories: Optional[list[str]] = None,
data_filter_2_attribute: Optional[str] = None,
data_filter_2_min: Optional[float] = None,
data_filter_2_max: Optional[float] = None,
data_filter_2_categories: Optional[list[str]] = None,
grid_type: str = "image",
frame: str = "world",
title: Optional[str] = None,
show_edges: bool = True,
z_up_lock: bool = False,
z_up_hotkey: str = "z",
app_name: str = "ParquetBlockModel Viewer",
) -> "BlockModelTrameApp":
query_text = str(duckdb_query).strip()
if not query_text:
raise ValueError("duckdb_query must be a non-empty SQL string.")
resolved_duckdb_path: Optional[str] = None
if duckdb_path is not None:
resolved_path = Path(duckdb_path).expanduser()
if not resolved_path.is_absolute():
resolved_path = resolved_path.resolve()
resolved_duckdb_path = str(resolved_path)
app = cls(
blockmodel=None,
scalar=scalar,
threshold_value=threshold_value,
data_filter_1_attribute=data_filter_1_attribute,
data_filter_1_min=data_filter_1_min,
data_filter_1_max=data_filter_1_max,
data_filter_1_categories=data_filter_1_categories,
data_filter_2_attribute=data_filter_2_attribute,
data_filter_2_min=data_filter_2_min,
data_filter_2_max=data_filter_2_max,
data_filter_2_categories=data_filter_2_categories,
grid_type=grid_type,
frame=frame,
title=title,
show_edges=show_edges,
z_up_lock=z_up_lock,
z_up_hotkey=z_up_hotkey,
app_name=app_name,
)
app._source_mode = "duckdb_query"
app._source_path = resolved_duckdb_path or "<duckdb in-memory>"
app._duckdb_query = query_text
app._duckdb_path = resolved_duckdb_path
app._skip_initial_blockmodel_load = True
return app
@classmethod
def from_source_path(
cls,
source_path: str | Path,
*,
scalar: Optional[str] = None,
threshold_value: Optional[float] = None,
data_filter_1_attribute: Optional[str] = None,
data_filter_1_min: Optional[float] = None,
data_filter_1_max: Optional[float] = None,
data_filter_1_categories: Optional[list[str]] = None,
data_filter_2_attribute: Optional[str] = None,
data_filter_2_min: Optional[float] = None,
data_filter_2_max: Optional[float] = None,
data_filter_2_categories: Optional[list[str]] = None,
grid_type: str = "image",
frame: str = "world",
title: Optional[str] = None,
show_edges: bool = True,
z_up_lock: bool = False,
z_up_hotkey: str = "z",
app_name: str = "ParquetBlockModel Viewer",
) -> "BlockModelTrameApp":
warnings.warn(
"BlockModelTrameApp.from_source_path(...) is deprecated; "
"use from_pbm_file(...) or from_hive_directory(...) for explicit startup modes.",
DeprecationWarning,
stacklevel=2,
)
path = Path(source_path).expanduser()
if not path.is_absolute():
path = path.resolve()
if not path.exists():
raise FileNotFoundError(f"Path does not exist: {path}")
if path.is_file():
return cls.from_pbm_file(
path,
scalar=scalar,
threshold_value=threshold_value,
data_filter_1_attribute=data_filter_1_attribute,
data_filter_1_min=data_filter_1_min,
data_filter_1_max=data_filter_1_max,
data_filter_1_categories=data_filter_1_categories,
data_filter_2_attribute=data_filter_2_attribute,
data_filter_2_min=data_filter_2_min,
data_filter_2_max=data_filter_2_max,
data_filter_2_categories=data_filter_2_categories,
grid_type=grid_type,
frame=frame,
title=title,
show_edges=show_edges,
z_up_lock=z_up_lock,
z_up_hotkey=z_up_hotkey,
app_name=app_name,
)
if path.is_dir():
return cls.from_hive_directory(
path,
scalar=scalar,
threshold_value=threshold_value,
data_filter_1_attribute=data_filter_1_attribute,
data_filter_1_min=data_filter_1_min,
data_filter_1_max=data_filter_1_max,
data_filter_1_categories=data_filter_1_categories,
data_filter_2_attribute=data_filter_2_attribute,
data_filter_2_min=data_filter_2_min,
data_filter_2_max=data_filter_2_max,
data_filter_2_categories=data_filter_2_categories,
grid_type=grid_type,
frame=frame,
title=title,
show_edges=show_edges,
z_up_lock=z_up_lock,
z_up_hotkey=z_up_hotkey,
app_name=app_name,
)
raise ValueError(f"Selected path is not a file or directory: {path}")
def _default_scalar(self) -> str:
if self.blockmodel is None:
return ""
return self.blockmodel.available_attributes[0]
def _resolve_initial_scalar(self, preferred_scalar: Optional[str] = None) -> str:
if self.blockmodel is None:
return ""
candidate = preferred_scalar or self._initial_scalar
if candidate in self.blockmodel.available_attributes:
return candidate
return self.blockmodel.available_attributes[0]
def _attribute_values(self, attribute: str) -> np.ndarray:
if self.state is None:
raise RuntimeError("Plot state has not been loaded yet.")
values = np.asarray(self.state.mesh.cell_data[attribute], dtype=float)
return values[np.isfinite(values)]
def _build_threshold_range(self, attribute: str) -> ThresholdRange:
values = self._attribute_values(attribute)
if values.size == 0:
return ThresholdRange(minimum=0.0, maximum=1.0, value=0.0, step=0.01)
minimum = float(np.min(values))
maximum = float(np.max(values))
span = max(maximum - minimum, 1.0)
return ThresholdRange(
minimum=minimum,
maximum=maximum,
value=minimum,
step=span / 200.0,
)
def _apply_initial_threshold_value(self) -> None:
if self.threshold is None or self._initial_threshold_value is None:
return
clamped = max(self.threshold.minimum, min(self.threshold.maximum, self._initial_threshold_value))
self.threshold.value = float(clamped)
def _ensure_startup_threshold_range(self) -> None:
if self.threshold is not None:
return
if self._initial_threshold_value is None:
self.threshold = ThresholdRange(minimum=0.0, maximum=1.0, value=0.0, step=0.005)
return
minimum = min(0.0, float(self._initial_threshold_value))
maximum = max(1.0, float(self._initial_threshold_value))
span = max(maximum - minimum, 1.0)
self.threshold = ThresholdRange(
minimum=minimum,
maximum=maximum,
value=float(self._initial_threshold_value),
step=span / 200.0,
)
def _clear_filter_cache(self) -> None:
self._filter_attribute_cache = {}
self._filter_category_code_to_label = {}
def _startup_filter_attribute_options(self) -> list[str]:
options: list[str] = []
if self._initial_scalar:
options.append(self._initial_scalar)
for slot in self._data_filters:
if slot.attribute and slot.attribute not in options:
options.append(slot.attribute)
return options
def _is_categorical_attribute(self, attribute: str) -> bool:
if self.blockmodel is None:
return False
dtype = self.blockmodel.column_dtypes.get(attribute)
if dtype is None:
return False
return isinstance(dtype, pd.CategoricalDtype)
def _is_continuous_attribute(self, attribute: str) -> bool:
if self.blockmodel is None:
return False
dtype = self.blockmodel.column_dtypes.get(attribute)
if dtype is None:
return False
if self._is_categorical_attribute(attribute):
return False
try:
return bool(np.issubdtype(np.dtype(dtype), np.number))
except TypeError:
return False
def _all_filter_attribute_options(self) -> list[str]:
if self.blockmodel is None:
return self._startup_filter_attribute_options()
return list(self.blockmodel.available_attributes)
def _load_filter_category_mapping(self, mesh: pv.DataSet, attribute: str) -> dict[float, str]:
if f"{attribute}_json" not in mesh.field_data:
return {}
raw_mapping = load_mapping_dict(mesh, attribute)
code_to_label: dict[float, str] = {}
for raw_code, label in raw_mapping.items():
code_to_label[float(raw_code)] = str(label)
return code_to_label
def _load_filter_attribute_values(self, attribute: str) -> np.ndarray:
if self.blockmodel is None:
raise RuntimeError("No blockmodel is loaded.")
if self.state is not None and attribute in self.state.mesh.cell_data:
self._filter_category_code_to_label[attribute] = self._load_filter_category_mapping(
self.state.mesh, attribute
)
return np.asarray(self.state.mesh.cell_data[attribute], dtype=float)
attribute_mesh = self.blockmodel.to_pyvista(
grid_type=self.grid_type,
attributes=[attribute],
frame=self.frame,
)
self._filter_category_code_to_label[attribute] = self._load_filter_category_mapping(
attribute_mesh, attribute
)
return np.asarray(attribute_mesh.cell_data[attribute], dtype=float)
def _get_filter_attribute_values(self, attribute: str) -> np.ndarray:
cached = self._filter_attribute_cache.get(attribute)
if cached is not None:
return cached
loaded = self._load_filter_attribute_values(attribute)
self._filter_attribute_cache[attribute] = loaded
return loaded
def _build_data_filter_range(self, attribute: str) -> tuple[float, float, float]:
values = self._get_filter_attribute_values(attribute)
finite_values = values[np.isfinite(values)]
if finite_values.size == 0:
return 0.0, 1.0, 0.01
minimum = float(np.min(finite_values))
maximum = float(np.max(finite_values))
span = max(maximum - minimum, 1.0)
return minimum, maximum, span / 200.0
def _filter_slot_name(self, slot_index: int) -> str:
return f"data_filter_{slot_index + 1}"
def _get_filter_slot(self, slot_index: int) -> DataFilterSlot:
if slot_index not in (0, 1):
raise ValueError("data filter slot_index must be 0 or 1.")
return self._data_filters[slot_index]
def _reset_filter_slot(self, slot_index: int) -> None:
slot = self._get_filter_slot(slot_index)
slot.attribute = ""
slot.is_categorical = False
slot.minimum = 0.0
slot.maximum = 1.0
slot.step = 0.005
slot.range_values = [0.0, 1.0]
slot.category_options = []
slot.selected_categories = []
def _set_filter_slot_attribute(
self,
slot_index: int,
attribute: str,
*,
selected_min: Optional[float] = None,
selected_max: Optional[float] = None,
selected_categories: Optional[list[str]] = None,
) -> None:
slot = self._get_filter_slot(slot_index)
normalized = str(attribute or "")
if normalized == "":
self._reset_filter_slot(slot_index)
return
if normalized not in self._filter_attribute_options:
raise ValueError(f"Filter attribute '{normalized}' is not available.")
slot.attribute = normalized
slot.is_categorical = self._is_categorical_attribute(normalized)
values = self._get_filter_attribute_values(normalized)
if slot.is_categorical:
code_to_label = self._filter_category_code_to_label.get(normalized, {})
ordered_labels = [label for _, label in sorted(code_to_label.items(), key=lambda item: item[0])]
if not ordered_labels:
finite_values = values[np.isfinite(values)]
unique_codes = sorted({float(v) for v in finite_values.tolist()})
ordered_labels = [str(int(code) if float(code).is_integer() else code) for code in unique_codes]
code_to_label = {
float(code): str(int(code) if float(code).is_integer() else code) for code in unique_codes
}
self._filter_category_code_to_label[normalized] = code_to_label
slot.category_options = ordered_labels
if selected_categories is None:
chosen = list(slot.preset_categories or ordered_labels)
else:
chosen = list(selected_categories)
slot.selected_categories = [label for label in chosen if label in slot.category_options]
slot.minimum = 0.0
slot.maximum = 1.0
slot.step = 0.005
slot.range_values = [0.0, 1.0]
else:
minimum, maximum, step = self._build_data_filter_range(normalized)
low_source = selected_min if selected_min is not None else slot.preset_min
high_source = selected_max if selected_max is not None else slot.preset_max
low = minimum if low_source is None else max(minimum, min(maximum, float(low_source)))
high = maximum if high_source is None else max(minimum, min(maximum, float(high_source)))
if low > high:
low, high = high, low
slot.minimum = minimum
slot.maximum = maximum
slot.step = step
slot.range_values = [low, high]
slot.category_options = []
slot.selected_categories = []
slot.preset_min = None
slot.preset_max = None
slot.preset_categories = None
def _apply_initial_data_filters(self) -> None:
for idx, slot in enumerate(self._data_filters):
if slot.attribute:
self._set_filter_slot_attribute(
idx,
slot.attribute,
selected_min=slot.preset_min,
selected_max=slot.preset_max,
selected_categories=slot.preset_categories,
)
def _apply_startup_filter_presets_without_data(self) -> None:
for slot in self._data_filters:
if not slot.attribute:
continue
low = slot.preset_min
high = slot.preset_max
if low is None and high is None:
continue
if low is None:
low = float(high)
if high is None:
high = float(low)
low_value = float(low)
high_value = float(high)
if low_value > high_value:
low_value, high_value = high_value, low_value
slot.minimum = min(0.0, low_value)
slot.maximum = max(1.0, high_value)
span = max(slot.maximum - slot.minimum, 1.0)
slot.step = span / 200.0
slot.range_values = [low_value, high_value]
def _data_filter_slot_summary(self, slot: DataFilterSlot) -> str:
if not slot.attribute:
return "None"
if slot.is_categorical:
selected = slot.selected_categories
total = len(slot.category_options)
if not selected:
return f"{slot.attribute}: 0/{total} selected"
if len(selected) <= 3:
return f"{slot.attribute}: {', '.join(selected)}"
return f"{slot.attribute}: {len(selected)}/{total} selected"
low, high = slot.range_values
return f"{slot.attribute}: {self._format_threshold(low)} to {self._format_threshold(high)}"
def _sync_data_filter_state(self) -> None:
if self._server is None:
return
state = self._server.state
state.data_filter_attribute_options = list(self._filter_attribute_options)
for idx, slot in enumerate(self._data_filters):
prefix = self._filter_slot_name(idx)
setattr(state, f"{prefix}_attribute", slot.attribute)
setattr(state, f"{prefix}_is_categorical", slot.is_categorical)
setattr(state, f"{prefix}_range_min", slot.minimum)
setattr(state, f"{prefix}_range_max", slot.maximum)
setattr(state, f"{prefix}_range_step", slot.step)
setattr(state, f"{prefix}_range", list(slot.range_values))
setattr(state, f"{prefix}_category_options", list(slot.category_options))
setattr(state, f"{prefix}_selected_categories", list(slot.selected_categories))
setattr(state, f"{prefix}_summary", self._data_filter_slot_summary(slot))
def set_data_filter_attribute(self, slot_index: int, attribute: str) -> None:
if self.blockmodel is None:
slot = self._get_filter_slot(slot_index)
normalized = str(attribute or "")
if normalized == "":
self._reset_filter_slot(slot_index)
else:
if normalized not in self._filter_attribute_options:
raise ValueError(f"Filter attribute '{normalized}' is not available.")
slot.attribute = normalized
slot.is_categorical = False
low = slot.preset_min if slot.preset_min is not None else slot.range_values[0]
high = slot.preset_max if slot.preset_max is not None else slot.range_values[1]
low_value = float(low)
high_value = float(high)
if low_value > high_value:
low_value, high_value = high_value, low_value
slot.minimum = min(0.0, low_value)
slot.maximum = max(1.0, high_value)
span = max(slot.maximum - slot.minimum, 1.0)
slot.step = span / 200.0
slot.range_values = [low_value, high_value]
slot.category_options = []
slot.selected_categories = []
if self._server is not None:
self._syncing_state = True
try:
self._sync_data_filter_state()
finally:
self._syncing_state = False
return
self._set_filter_slot_attribute(slot_index, attribute)
if self.state is not None:
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._syncing_state = True
try:
self._sync_data_filter_state()
finally:
self._syncing_state = False
def set_data_filter_range(self, slot_index: int, values: list[float] | tuple[float, float]) -> None:
slot = self._get_filter_slot(slot_index)
if not slot.attribute or slot.is_categorical:
return
if len(values) != 2:
raise ValueError("Filter range must have exactly two values.")
lower = max(slot.minimum, min(slot.maximum, float(values[0])))
upper = max(slot.minimum, min(slot.maximum, float(values[1])))
if lower > upper:
lower, upper = upper, lower
if lower == slot.range_values[0] and upper == slot.range_values[1]:
return
slot.range_values = [lower, upper]
if self.state is not None:
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._syncing_state = True
try:
prefix = self._filter_slot_name(slot_index)
setattr(self._server.state, f"{prefix}_range", list(slot.range_values))
setattr(self._server.state, f"{prefix}_summary", self._data_filter_slot_summary(slot))
finally:
self._syncing_state = False
def set_data_filter_categories(self, slot_index: int, categories: list[str]) -> None:
slot = self._get_filter_slot(slot_index)
if not slot.attribute or not slot.is_categorical:
return
slot.selected_categories = [label for label in categories if label in slot.category_options]
if self.state is not None:
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._syncing_state = True
try:
prefix = self._filter_slot_name(slot_index)
setattr(
self._server.state,
f"{prefix}_selected_categories",
list(slot.selected_categories),
)
setattr(self._server.state, f"{prefix}_summary", self._data_filter_slot_summary(slot))
finally:
self._syncing_state = False
def reset_data_filter(self, slot_index: Optional[int] = None) -> None:
if slot_index is None:
self._reset_filter_slot(0)
self._reset_filter_slot(1)
else:
self._reset_filter_slot(slot_index)
if self.state is not None:
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._syncing_state = True
try:
self._sync_data_filter_state()
finally:
self._syncing_state = False
def _refresh_filter_options(self) -> None:
self._filter_attribute_options = self._all_filter_attribute_options()
if self.blockmodel is None:
return
for idx, slot in enumerate(self._data_filters):
if slot.attribute and slot.attribute not in self._filter_attribute_options:
self._reset_filter_slot(idx)
def _format_threshold(self, value: float) -> str:
return f"{value:,.3f}".rstrip("0").rstrip(".")
def _branding_logo_resource(self):
return resources.files("parq_blockmodel.assets.branding").joinpath("parq-blockmodel.svg")
def _branding_logo_data_url(self) -> str:
with resources.as_file(self._branding_logo_resource()) as brand_logo_path:
svg_text = Path(brand_logo_path).read_text(encoding="utf-8")
return "data:image/svg+xml;charset=utf-8," + quote(svg_text)
def _setup_picking_callback(self):
"""Create and setup cell picking callback with categorical support."""
if self.state is None:
return
def _push_pick_state(
*,
dialog_open: Optional[bool] = None,
dialog_text: Optional[str] = None,
debug_text: Optional[str] = None,
) -> None:
if self._server is None:
return
state = self._server.state
changed_keys: list[str] = []
if dialog_open is not None:
state.picking_dialog_open = bool(dialog_open)
changed_keys.append("picking_dialog_open")
if dialog_text is not None:
state.picking_dialog_text = str(dialog_text)
changed_keys.append("picking_dialog_text")
if debug_text is not None:
state.picking_debug_last = str(debug_text)
changed_keys.append("picking_debug_last")
state.picking_debug_count = int(self._pick_debug_count)
changed_keys.append("picking_debug_count")
if hasattr(state, "dirty"):
for key in changed_keys:
try:
state.dirty(key)
except Exception:
break
if hasattr(state, "flush"):
try:
state.flush()
except Exception:
pass
def _extract_cell_id_from_pick(picked_cell) -> Optional[int]:
def _to_int(value: object) -> Optional[int]:
try:
arr = np.asarray(value).ravel()
if arr.size == 0:
return None
cell_id = int(np.rint(float(arr[0])))
except (TypeError, ValueError):
return None
return cell_id if cell_id >= 0 else None
data_sources = []
if isinstance(picked_cell, dict):
data_sources.append(picked_cell)
for attr_name in ("cell_data", "point_data", "field_data"):
data = getattr(picked_cell, attr_name, None)
if data is not None:
data_sources.append(data)
id_keys = (
"vtkOriginalCellIds",
"vtkOriginalCellId",
"vtkCellIds",
"cell_ids",
"cellIds",
"cell_id",
"cellId",
"id",
)
for source in data_sources:
for key in id_keys:
try:
if key in source:
maybe_id = _to_int(source[key])
if maybe_id is not None:
return maybe_id
except Exception:
continue
for attr_name in ("cell_id", "cellId", "id"):
maybe_id = _to_int(getattr(picked_cell, attr_name, None))
if maybe_id is not None:
return maybe_id
return None
def cell_callback(picked_cell):
if self.state is None:
return
self._pick_debug_count += 1
picked_cells = int(getattr(picked_cell, "n_cells", 0) or 0)
cell_id = _extract_cell_id_from_pick(picked_cell)
payload_keys: list[str] = []
for source_name in ("cell_data", "point_data", "field_data"):
source = getattr(picked_cell, source_name, None)
try:
if source is not None and hasattr(source, "keys"):
payload_keys.extend([f"{source_name}.{k}" for k in list(source.keys())[:4]])
except Exception:
continue
self._pick_debug_last = (
f"count={self._pick_debug_count}, n_cells={picked_cells}, cell_id={cell_id}, "
f"keys={payload_keys[:6]}"
)
if picked_cells != 1 or cell_id is None or cell_id >= self.state.mesh.n_cells:
_push_pick_state(
dialog_open=False,
dialog_text="",
debug_text=self._pick_debug_last,
)
return
cell_centers = self.state.mesh.cell_centers().points
centroid = cell_centers[cell_id]
centroid_str = f"({centroid[0]:.1f}, {centroid[1]:.1f}, {centroid[2]:.1f})"
values: dict[str, object] = {}
for attr in self.state.attributes:
raw_value = self.state.mesh.cell_data[attr][cell_id]
if attr in self.state.categorical_mappings:
if pd.isna(raw_value):
values[attr] = "<NA>"
else:
code = int(np.rint(float(raw_value)))
values[attr] = self.state.categorical_mappings[attr].get(code, f"<unknown:{code}>")
else:
values[attr] = raw_value
msg = f"Cell ID: {cell_id}, {centroid_str}, " + ", ".join(
f"{k}: {v}" for k, v in values.items()
)
_push_pick_state(
dialog_open=True,
dialog_text=msg,
debug_text=self._pick_debug_last,
)
self.plotter.setup_picking_with_callback(cell_callback)
def _filtered_mesh(self) -> pv.DataSet:
if self.state is None:
raise RuntimeError("Plot state has not been loaded yet.")
cell_count = self.state.mesh.n_cells
mask = np.ones(cell_count, dtype=bool)
if not self.state.scalar_is_categorical and self.filter_enabled:
threshold_values = np.asarray(self.state.mesh.cell_data[self.state.scalar], dtype=float)
mask &= np.isfinite(threshold_values) & (threshold_values >= self.threshold.value)
for slot in self._data_filters:
if not slot.attribute:
continue
values = self._get_filter_attribute_values(slot.attribute)
if values.shape[0] != cell_count:
continue
if slot.is_categorical:
code_to_label = self._filter_category_code_to_label.get(slot.attribute, {})
selected_labels = set(slot.selected_categories)
selected_codes = [
code
for code, label in code_to_label.items()
if label in selected_labels
]
if selected_codes:
mask &= np.isin(values, np.asarray(selected_codes, dtype=float))
else:
mask &= np.zeros(cell_count, dtype=bool)
else:
lower, upper = slot.range_values
mask &= np.isfinite(values) & (values >= lower) & (values <= upper)
if bool(np.all(mask)):
return self.state.mesh
return self.state.mesh.extract_cells(mask)
def _mesh_with_decile_scalars(self, mesh: pv.DataSet) -> tuple[pv.DataSet, np.ndarray]:
if self.state is None:
raise RuntimeError("Plot state has not been loaded yet.")
scalar_values = np.asarray(self.state.mesh.cell_data[self.state.scalar], dtype=float)
decile_edges = _calculate_deciles(scalar_values)
mesh_values = np.asarray(mesh.cell_data[self.state.scalar], dtype=float)
decile_bins = _bin_to_deciles(mesh_values, decile_edges)
display_mesh = mesh.copy(deep=True)
display_mesh.cell_data["__pbm_decile_bin__"] = decile_bins
return display_mesh, decile_edges
def _refresh_plot(self, *, preserve_camera: bool = True) -> None:
if self.state is None:
return
camera_position = None
if preserve_camera and self._view_initialized:
camera_position = getattr(self.plotter, "camera_position", None)
self.plotter.clear()
mesh = self._filtered_mesh()
self._active_display_mesh = mesh
is_decile_mode = self.discretize_deciles and not self.state.scalar_is_categorical
decile_edges: Optional[np.ndarray] = None
scalar_for_coloring = self.state.scalar
if is_decile_mode:
mesh, decile_edges = self._mesh_with_decile_scalars(mesh)
scalar_for_coloring = "__pbm_decile_bin__"
mesh_kwargs = _plotter_add_mesh_kwargs(
self.state,
colormap=self.colormap,
discretize_to_deciles=is_decile_mode,
decile_edges=decile_edges,
scalar_for_coloring=scalar_for_coloring,
)
mesh_kwargs["show_edges"] = self.show_edges
if not self.state.scalar_is_categorical and not is_decile_mode:
mesh_values = np.asarray(self.state.mesh.cell_data[self.state.scalar], dtype=float)
finite_values = mesh_values[np.isfinite(mesh_values)]
if finite_values.size > 0:
mesh_kwargs["clim"] = (float(np.min(finite_values)), float(np.max(finite_values)))
self.plotter.add_mesh(mesh, name="blockmodel", **mesh_kwargs)
if self.show_model_bounds:
outline = mesh.outline()
self.plotter.add_mesh(
outline,
color="dodgerblue",
line_width=2,
name="model_bounds",
)
self.plotter.title = self.title
self.plotter.add_axes()
if preserve_camera and camera_position not in (None, []):
self.plotter.camera_position = camera_position
else:
self.plotter.set_directional_view(direction='WSW', elevation_deg=30)
self.plotter.reset_camera_clipping_range()
if bool(getattr(self.plotter, "hotkey_pressed", {}).get("z")):
self.plotter.enforce_z_up()
if bool(getattr(self.plotter, "picking_enabled", False)):
self._setup_picking_callback()
self.plotter.render()
self._view_initialized = True
if self._remote_view is not None:
self._remote_view.update()
def _reset_model_view(self, *, preserve_presets: bool = False) -> None:
self.blockmodel = None
self.state = None
self.threshold = None
self._ensure_startup_threshold_range()
self.filter_enabled = False
if not preserve_presets:
self._reset_filter_slot(0)
self._reset_filter_slot(1)
else:
self._apply_startup_filter_presets_without_data()
self._filter_attribute_options = []
self._clear_filter_cache()
self._view_initialized = False
self.set_picking_active(False)
self._active_display_mesh = None
self.plotter.clear()
if self._remote_view is not None:
self._remote_view.update()
if self._server is not None:
state = self._server.state
state.attribute_options = self._startup_filter_attribute_options() if preserve_presets else []
state.active_attribute = self._initial_scalar if preserve_presets else ""
state.threshold_min = self.threshold.minimum
state.threshold_max = self.threshold.maximum
state.threshold = self.threshold.value
state.threshold_display = self._format_threshold(self.threshold.value)
state.threshold_step = self.threshold.step
state.filter_active = False
state.picking_active = False
state.picking_dialog_open = False
state.picking_dialog_text = ""
state.picking_debug_count = int(self._pick_debug_count)
state.picking_debug_last = self._pick_debug_last
state.show_model_bounds = self.show_model_bounds
if preserve_presets:
self._refresh_filter_options()
self._sync_data_filter_state()
else:
state.data_filter_attribute_options = []
for idx in range(2):
prefix = self._filter_slot_name(idx)
setattr(state, f"{prefix}_attribute", "")
setattr(state, f"{prefix}_is_categorical", False)
setattr(state, f"{prefix}_range_min", 0.0)
setattr(state, f"{prefix}_range_max", 1.0)
setattr(state, f"{prefix}_range_step", 0.005)
setattr(state, f"{prefix}_range", [0.0, 1.0])
setattr(state, f"{prefix}_category_options", [])
setattr(state, f"{prefix}_selected_categories", [])
setattr(state, f"{prefix}_summary", "None")
state.model_name = ""
state.model_path = ""
def _load_plot_state(self, scalar: str) -> None:
self.state = prepare_plot_state(
self.blockmodel,
scalar=scalar,
grid_type=self.grid_type,
frame=self.frame,
enable_picking=False,
title=self.title,
)
self.threshold = self._build_threshold_range(self.state.scalar)
self.filter_enabled = True
def load_blockmodel(self, blockmodel: "ParquetBlockModel", *, preferred_scalar: Optional[str] = None, auto_select_scalar: bool = True) -> None:
self.blockmodel = blockmodel
if not self._title_override:
self.title = self.blockmodel.name
self._clear_filter_cache()
self._refresh_filter_options()
self._syncing_state = True
try:
# Only load plot state if auto_select_scalar is True (file mode)
# For hive mode, just populate attributes and wait for user selection
if auto_select_scalar:
scalar = self._resolve_initial_scalar(preferred_scalar)
self._initial_scalar = scalar
self._load_plot_state(scalar)
self._apply_initial_threshold_value()
self._apply_initial_data_filters()
self._refresh_plot(preserve_camera=False)
else:
# Hive mode: preserve any preset selections until a concrete asset is loaded.
self._apply_initial_data_filters()
if self._server is not None:
self._server.state.attribute_options = self.blockmodel.available_attributes
self._server.state.model_name = self.blockmodel.name
self._server.state.model_path = str(self.blockmodel.blockmodel_path)
if auto_select_scalar and self.threshold is not None and self.state is not None:
# File mode: populate all threshold/attribute state
self._server.state.active_attribute = self.state.scalar
self._server.state.threshold_min = self.threshold.minimum
self._server.state.threshold_max = self.threshold.maximum
self._server.state.threshold = self.threshold.value
self._server.state.threshold_step = self.threshold.step
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
self._server.state.filter_active = self.filter_enabled
else:
# Hive mode: preserve the preset scalar label until an asset is loaded.
self._server.state.active_attribute = self._initial_scalar
self._ensure_startup_threshold_range()
self._server.state.threshold_min = self.threshold.minimum
self._server.state.threshold_max = self.threshold.maximum
self._server.state.threshold = self.threshold.value
self._server.state.threshold_step = self.threshold.step
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
self._server.state.filter_active = False
self._sync_data_filter_state()
finally:
self._syncing_state = False
def _asset_selection(self) -> dict[str, str]:
return {
key: value
for key, value in self._asset_level_values.items()
if value not in ("", None)
}
def _refresh_asset_selector_values(self) -> tuple[list[list[str]], list[str]]:
if self.asset_catalog is None:
return [], []
level_options: list[list[str]] = []
selection_prefix: dict[str, str] = {}
for key in self._asset_level_keys:
options = self.asset_catalog.level_options(key, selection_prefix)
current_value = self._asset_level_values.get(key, "")
# IMPORTANT: Don't auto-fill to first option on initial load
# Only auto-fill if user has explicitly selected values AND an option is missing
if current_value and current_value not in options:
# Only auto-select if user has already interacted with this level
if self._asset_selectors_autofilled:
current_value = options[0] if options else ""
# Don't set to first option just because list is empty
self._asset_level_values[key] = current_value
if current_value:
selection_prefix[key] = current_value
level_options.append(options)
name_options = self.asset_catalog.pbm_name_options(self._asset_selection())
# Similar logic for asset name - don't auto-select first PBM on initial load
if self._selected_asset_name and self._selected_asset_name not in name_options:
if self._asset_selectors_autofilled:
self._selected_asset_name = name_options[0] if name_options else ""
return level_options, name_options
def _sync_asset_selector_state(self) -> None:
if self._server is None:
return
state = self._server.state
state.asset_selector_enabled = self.asset_catalog is not None
if self.asset_catalog is None:
state.asset_name_options = []
state.selected_asset_name = ""
state.asset_selected_path = ""
return
level_options, name_options = self._refresh_asset_selector_values()
self._syncing_state = True
try:
# IMPORTANT: Set values FIRST before options to prevent Vuetify from clearing v_model
# when items array is updated
for idx, key in enumerate(self._asset_level_keys):
setattr(state, f"asset_level_{idx}_value", self._asset_level_values.get(key, ""))
state.selected_asset_name = self._selected_asset_name
# Now set options after values are in place
for idx, key in enumerate(self._asset_level_keys):
setattr(state, f"asset_level_{idx}_options", level_options[idx])
state.asset_name_options = name_options
try:
asset = self.asset_catalog.select_asset(self._asset_selection(), self._selected_asset_name)
state.asset_selected_path = str(asset.path)
except LookupError:
state.asset_selected_path = ""
finally:
self._syncing_state = False
def _register_asset_selector_handlers(self) -> None:
if self._server is None:
return
state, ctrl = self._server.state, self._server.controller
for idx, _ in enumerate(self._asset_level_keys):
if idx in self._registered_asset_level_indices:
continue
self._registered_asset_level_indices.add(idx)
field_name = f"asset_level_{idx}_value"
def _make_asset_level_handler(level_index: int, watched_field: str):
@state.change(watched_field)
def _asset_level_changed(**kwargs):
# IMPORTANT: Only process if the watched field is actually in kwargs
# Trame sometimes calls handlers with empty kwargs for internal events
if watched_field not in kwargs:
return
if self._syncing_state:
return
if level_index >= len(self._asset_level_keys):
return
level_key = self._asset_level_keys[level_index]
new_value = str(kwargs.get(watched_field) or "")
old_value = self._asset_level_values.get(level_key, "")
# Only sync if value actually changed
if new_value == old_value:
return
# Update current level
self._asset_level_values[level_key] = new_value
# Clear all child levels and asset name when parent changes
for idx in range(level_index + 1, len(self._asset_level_keys)):
self._asset_level_values[self._asset_level_keys[idx]] = ""
self._selected_asset_name = ""
# Clear model view immediately when selection changes
self._reset_model_view(preserve_presets=True)
self._asset_selectors_autofilled = True
# Let _sync_asset_selector_state manage _syncing_state
self._sync_asset_selector_state()
# NOTE: Don't call _load_selected_asset() - only load when asset name is selected
return _asset_level_changed
setattr(ctrl, f"update_asset_level_{idx}", _make_asset_level_handler(idx, field_name))
if not self._asset_name_handler_registered:
@state.change("selected_asset_name")
def _asset_name_changed(selected_asset_name=None, **_):
# IMPORTANT: Only process if selected_asset_name is not None
# Trame sometimes calls handlers with None for internal events
if selected_asset_name is None:
return
if self._syncing_state:
return
new_name = str(selected_asset_name or "")
old_name = self._selected_asset_name
# Only proceed if value actually changed
if new_name == old_name:
return
self._selected_asset_name = new_name
self._asset_selectors_autofilled = True
# Only load when asset name is explicitly selected (not empty)
if new_name:
self._reset_model_view(preserve_presets=True)
self._load_selected_asset()
# Sync state AFTER loading asset so attribute_options are available
self._sync_asset_selector_state()
else:
# Clear canvas if asset name is deselected
self._reset_model_view(preserve_presets=True)
self._sync_asset_selector_state()
ctrl.update_asset_name = _asset_name_changed
self._asset_name_handler_registered = True
def _set_asset_selection_from_current_model(self) -> None:
if self.asset_catalog is None:
return
found = self.asset_catalog.find_by_path(self.blockmodel.blockmodel_path)
if found is None:
return
level_map = found.level_map
self._asset_level_values = {key: level_map.get(key, "") for key in self._asset_level_keys}
self._selected_asset_name = found.name
def _load_selected_asset(self) -> None:
if self.asset_catalog is None or not self._selected_asset_name:
return
try:
selected = self.asset_catalog.select_asset(self._asset_selection(), self._selected_asset_name)
except LookupError:
return
# In file mode, skip if asset is already loaded
# In hive mode with None blockmodel, always load (first time)
if self._source_mode == "file" and self.blockmodel is not None:
if selected.path.resolve() == self.blockmodel.blockmodel_path.resolve():
return
from parq_blockmodel.blockmodel import ParquetBlockModel
self.load_blockmodel(
ParquetBlockModel(blockmodel_path=selected.path),
preferred_scalar=self._initial_scalar,
)
self._set_asset_selection_from_current_model()
self._sync_asset_selector_state()
def load_source_path(self, source_path: str | Path) -> None:
from parq_blockmodel.blockmodel import ParquetBlockModel
path = Path(source_path).expanduser()
if not path.is_absolute():
path = path.resolve()
if not path.exists():
raise FileNotFoundError(f"Path does not exist: {path}")
if path.is_file():
if path.suffix.lower() != ".pbm":
raise ValueError(f"Selected file must be a .pbm file: {path}")
self.asset_catalog = None
self.asset_catalog_root = None
self._asset_level_keys = []
self._asset_level_values = {}
self._selected_asset_name = ""
self._asset_selectors_autofilled = False
self._source_mode = "file"
self._source_path = str(path)
if self._server is not None:
self._server.state.source_mode = self._source_mode
self._server.state.source_path_input = self._source_path
self.load_blockmodel(ParquetBlockModel(blockmodel_path=path))
if self._server is not None:
self._server.state.source_status = f"Loaded PBM: {path.name}"
self._sync_asset_selector_state()
return
if not path.is_dir():
raise ValueError(f"Selected path is not a file or directory: {path}")
catalog = HivePbmCatalog.discover(path)
self.asset_catalog = catalog
self.asset_catalog_root = path.resolve()
self._asset_level_keys = list(catalog.level_keys)
self._asset_level_values = {key: "" for key in self._asset_level_keys}
self._selected_asset_name = ""
self._asset_selectors_autofilled = False
self._source_mode = "hive"
self._source_path = str(path.resolve())
# Don't auto-load first asset - let user select from DDLs
if self._server is not None:
self._server.state.source_mode = self._source_mode
self._server.state.source_path_input = self._source_path
self._server.state.source_status = f"Loaded hive directory: {self._source_path}"
self._register_asset_selector_handlers()
self._sync_asset_selector_state()
def set_attribute(self, attribute: str) -> None:
if self.blockmodel is None:
self._initial_scalar = str(attribute or "")
self._ensure_startup_threshold_range()
if self._server is not None and self.threshold is not None:
self._syncing_state = True
try:
self._server.state.active_attribute = self._initial_scalar
self._server.state.threshold_min = self.threshold.minimum
self._server.state.threshold_max = self.threshold.maximum
self._server.state.threshold = self.threshold.value
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
self._server.state.threshold_step = self.threshold.step
self._sync_data_filter_state()
finally:
self._syncing_state = False
return
self._syncing_state = True
try:
self._load_plot_state(attribute)
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None and self.threshold is not None:
self._server.state.active_attribute = attribute
self._server.state.threshold_min = self.threshold.minimum
self._server.state.threshold_max = self.threshold.maximum
self._server.state.threshold = self.threshold.value
self._server.state.filter_active = self.filter_enabled
self._sync_data_filter_state()
finally:
self._syncing_state = False
def set_threshold(self, value: float) -> None:
if self.threshold is None:
raise RuntimeError("Threshold range has not been initialised.")
self.threshold.value = float(value)
self.filter_enabled = True
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._server.state.threshold = self.threshold.value
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
self._server.state.filter_active = self.filter_enabled
def set_threshold_from_text(self, text_value: str) -> None:
if self.threshold is None:
raise RuntimeError("Threshold range has not been initialised.")
try:
value = float(text_value)
if not (self.threshold.minimum <= value <= self.threshold.maximum):
if self._server is not None:
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
return
self.set_threshold(value)
except (ValueError, TypeError):
if self._server is not None:
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
def reset_filter(self) -> None:
if self.threshold is None:
raise RuntimeError("Threshold range has not been initialised.")
self.filter_enabled = False
self.threshold.value = self.threshold.minimum
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._syncing_state = True
try:
self._server.state.threshold = self.threshold.value
self._server.state.threshold_display = self._format_threshold(self.threshold.value)
self._server.state.filter_active = False
finally:
self._syncing_state = False
def set_colormap(self, colormap: str) -> None:
self.colormap = colormap
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._server.state.selected_colormap = colormap
def set_discretize_deciles(self, enabled: bool) -> None:
self.discretize_deciles = bool(enabled)
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._server.state.discretize_deciles = self.discretize_deciles
def set_picking_active(self, enabled: bool) -> None:
enabled = bool(enabled)
if not hasattr(self.plotter, "picking_enabled"):
return
if enabled and not self.plotter.picking_enabled:
self._setup_picking_callback()
self.plotter.picking_enabled = True
self._pick_debug_last = "Picking enabled"
elif not enabled and self.plotter.picking_enabled:
self.plotter.disable_picking()
self.plotter.picking_enabled = False
if self._server is not None:
self._server.state.picking_dialog_open = False
self._server.state.picking_dialog_text = ""
self._pick_debug_last = "Picking disabled"
if self._server is not None:
self._server.state.picking_active = self.plotter.picking_enabled
self._server.state.picking_debug_count = int(self._pick_debug_count)
self._server.state.picking_debug_last = self._pick_debug_last
self.plotter.render()
if self._remote_view is not None:
self._remote_view.update()
def set_show_model_bounds(self, enabled: bool) -> None:
self.show_model_bounds = bool(enabled)
self._refresh_plot(preserve_camera=self._view_initialized)
if self._server is not None:
self._server.state.show_model_bounds = self.show_model_bounds
def _get_available_colormaps(self) -> list[str]:
try:
import matplotlib.pyplot as plt
all_cmaps = list(plt.colormaps())
except (ImportError, AttributeError):
all_cmaps = ["viridis", "plasma", "inferno", "cool", "hot"]
useful_cmaps = [
"viridis", "plasma", "inferno", "magma", "cividis",
"cool", "hot", "spring", "summer", "autumn", "winter",
"jet", "RdYlBu", "RdYlGn", "Spectral", "coolwarm", "seismic",
"twilight", "twilight_shifted", "hsv", "bone", "copper",
"gray", "hot_r", "cool_r", "Greys", "Blues", "Greens", "Oranges",
"Reds", "Purples", "RdPu", "BuGn", "BuPu", "GnBu", "OrRd", "PuBu"
]
available = [c for c in useful_cmaps if c in all_cmaps]
if not available:
available = all_cmaps[:10]
return sorted(available)
def _build_launch_kwargs(self, port: Optional[int], host: Optional[str]) -> dict[str, object]:
start_kwargs: dict[str, object] = {"open_browser": True, "show_connection_info": True}
if port is not None:
start_kwargs["port"] = port
if host is not None:
start_kwargs["host"] = host
return start_kwargs
[docs]
def launch(
self,
server_name: str = "parq-blockmodel-trame",
port: Optional[int] = None,
host: Optional[str] = None,
):
"""Launch the Trame visualization app.
Parameters
----------
server_name : str, optional
Base name for the Trame server, by default "parq-blockmodel-trame"
A UUID suffix is appended to ensure unique server instances.
port : int, optional
Port number for the server. If None, Trame will auto-assign a port.
Useful for running multiple instances without server state conflicts.
Example: launch(port=8080), launch(port=8081), etc.
host : str, optional
Host interface to bind to. Use "0.0.0.0" to accept external connections.
"""
try:
from trame.app import get_server
from trame.ui.vuetify import SinglePageWithDrawerLayout
from trame.widgets import vtk, vuetify
except ImportError as exc: # pragma: no cover
raise ImportError(
"Trame dependencies are required for the interactive app. "
"Install the 'trame', 'trame-vtk', and 'trame-vuetify' packages."
) from exc
try: # Optional: used for browser key-capture hotkeys.
from trame.widgets import trame as trame_widgets
except ImportError: # pragma: no cover
trame_widgets = None
# Reset all instance state to ensure fresh launch (not stale from previous launch)
self._asset_level_values = {}
self._selected_asset_name = ""
self._asset_selectors_autofilled = False
# Use unique server name to force fresh Trame server instance (prevents caching)
unique_server_name = f"{server_name}-{uuid4().hex[:8]}"
server = get_server(unique_server_name, client_type="vue2")
self._server = server
state, ctrl = server.state, server.controller
self.plotter.clear()
self._syncing_state = True
try:
state.source_mode = self._source_mode
state.source_path_input = self._source_path
state.picking_dialog_open = False
state.picking_dialog_text = ""
state.picking_debug_count = int(self._pick_debug_count)
state.picking_debug_last = self._pick_debug_last
state.picking_active = bool(getattr(self.plotter, "picking_enabled", False))
state.show_model_bounds = self.show_model_bounds
# For hive mode, skip initial blockmodel load - user will select via dropdowns
if not self._skip_initial_blockmodel_load:
self.load_blockmodel(self.blockmodel, preferred_scalar=self._initial_scalar)
# Initialize available colormaps
self.available_colormaps = self._get_available_colormaps()
self._refresh_filter_options()
# Set state only if blockmodel was loaded
if self.state is not None:
state.attribute_options = self.blockmodel.available_attributes
state.active_attribute = self.state.scalar
state.threshold_min = self.threshold.minimum
state.threshold_max = self.threshold.maximum
state.threshold = self.threshold.value
state.threshold_display = self._format_threshold(self.threshold.value)
state.threshold_step = self.threshold.step
state.filter_active = self.filter_enabled
else:
# Hive mode before user selection: show presets, but keep the canvas empty.
self._ensure_startup_threshold_range()
self._apply_startup_filter_presets_without_data()
state.attribute_options = self._startup_filter_attribute_options()
state.active_attribute = self._initial_scalar
state.threshold_min = self.threshold.minimum
state.threshold_max = self.threshold.maximum
state.threshold = self.threshold.value
state.threshold_display = self._format_threshold(self.threshold.value)
state.threshold_step = self.threshold.step
state.filter_active = False
self._sync_data_filter_state()
state.colormap_options = self.available_colormaps
state.selected_colormap = self.colormap
state.discretize_deciles = self.discretize_deciles
state.control_panel = 0
if self.state is not None:
state.model_name = self.blockmodel.name
state.model_path = str(self.blockmodel.blockmodel_path)
else:
state.model_name = ""
state.model_path = ""
state.source_mode = self._source_mode
state.source_path_input = self._source_path
if self._source_mode == "duckdb_query":
state.source_status = "DuckDB query mode is scaffolded; execution is not implemented yet."
elif self._source_mode == "file" and self.blockmodel is not None:
state.source_status = f"Loaded PBM: {self.blockmodel.blockmodel_path.name}"
elif self._source_mode == "hive" and self._source_path:
state.source_status = f"Loaded hive directory: {self._source_path}"
else:
state.source_status = ""
state.asset_selector_enabled = self.asset_catalog is not None
if self.asset_catalog is not None:
self._asset_level_keys = list(self.asset_catalog.level_keys)
self._asset_level_values = {key: "" for key in self._asset_level_keys}
finally:
self._syncing_state = False
@state.change("active_attribute")
def _attribute_changed(active_attribute=None, **_):
if self._syncing_state or active_attribute is None or not active_attribute:
return
if self.state is not None and str(active_attribute) == self.state.scalar:
return
if self.blockmodel is None:
self._initial_scalar = str(active_attribute)
return
self.set_attribute(active_attribute)
@state.change("threshold")
def _threshold_changed(threshold=None, **_):
if self._syncing_state or threshold is None:
return
if self.threshold is not None and float(threshold) == float(self.threshold.value):
return
self.set_threshold(threshold)
@state.change("filter_active")
def _filter_active_changed(filter_active=None, **_):
if self._syncing_state or filter_active is None:
return
self.filter_enabled = bool(filter_active)
self._refresh_plot(preserve_camera=self._view_initialized)
@state.change("data_filter_1_attribute")
def _data_filter_1_attribute_changed(data_filter_1_attribute=None, **_):
if self._syncing_state or data_filter_1_attribute is None:
return
if str(data_filter_1_attribute) == self._data_filters[0].attribute:
return
self.set_data_filter_attribute(0, str(data_filter_1_attribute))
@state.change("data_filter_2_attribute")
def _data_filter_2_attribute_changed(data_filter_2_attribute=None, **_):
if self._syncing_state or data_filter_2_attribute is None:
return
if str(data_filter_2_attribute) == self._data_filters[1].attribute:
return
self.set_data_filter_attribute(1, str(data_filter_2_attribute))
@state.change("data_filter_1_range")
def _data_filter_1_range_changed(data_filter_1_range=None, **_):
if self._syncing_state or data_filter_1_range is None:
return
self.set_data_filter_range(0, data_filter_1_range)
@state.change("data_filter_2_range")
def _data_filter_2_range_changed(data_filter_2_range=None, **_):
if self._syncing_state or data_filter_2_range is None:
return
self.set_data_filter_range(1, data_filter_2_range)
@state.change("data_filter_1_selected_categories")
def _data_filter_1_categories_changed(data_filter_1_selected_categories=None, **_):
if self._syncing_state or data_filter_1_selected_categories is None:
return
self.set_data_filter_categories(0, list(data_filter_1_selected_categories))
@state.change("data_filter_2_selected_categories")
def _data_filter_2_categories_changed(data_filter_2_selected_categories=None, **_):
if self._syncing_state or data_filter_2_selected_categories is None:
return
self.set_data_filter_categories(1, list(data_filter_2_selected_categories))
@state.change("selected_colormap")
def _colormap_changed(selected_colormap=None, **_):
if self._syncing_state or selected_colormap is None or not selected_colormap:
return
self.set_colormap(selected_colormap)
@state.change("discretize_deciles")
def _discretize_deciles_changed(discretize_deciles=None, **_):
if self._syncing_state or discretize_deciles is None:
return
self.set_discretize_deciles(bool(discretize_deciles))
@state.change("picking_active")
def _picking_active_changed(picking_active=None, **_):
if self._syncing_state or picking_active is None:
return
if bool(picking_active) == bool(getattr(self.plotter, "picking_enabled", False)):
return
self.set_picking_active(bool(picking_active))
@state.change("show_model_bounds")
def _show_model_bounds_changed(show_model_bounds=None, **_):
if self._syncing_state or show_model_bounds is None:
return
if bool(show_model_bounds) == self.show_model_bounds:
return
self.set_show_model_bounds(bool(show_model_bounds))
ctrl.update_attribute = _attribute_changed
ctrl.update_threshold = _threshold_changed
ctrl.update_colormap = _colormap_changed
ctrl.update_discretize_deciles = _discretize_deciles_changed
ctrl.update_data_filter_1_attribute = _data_filter_1_attribute_changed
ctrl.update_data_filter_2_attribute = _data_filter_2_attribute_changed
ctrl.update_data_filter_1_range = _data_filter_1_range_changed
ctrl.update_data_filter_2_range = _data_filter_2_range_changed
ctrl.update_data_filter_1_categories = _data_filter_1_categories_changed
ctrl.update_data_filter_2_categories = _data_filter_2_categories_changed
ctrl.update_picking_active = _picking_active_changed
ctrl.update_show_model_bounds = _show_model_bounds_changed
def _apply_threshold_text(**_):
if self._syncing_state or self._server is None:
return
self.set_threshold_from_text(str(self._server.state.threshold_display))
ctrl.apply_threshold_text = _apply_threshold_text
ctrl.reset_filter = self.reset_filter
ctrl.reset_data_filter_1 = lambda **_: self.reset_data_filter(0)
ctrl.reset_data_filter_2 = lambda **_: self.reset_data_filter(1)
ctrl.update_asset_name = lambda **_: None
ctrl.close_picking_dialog = lambda **_: setattr(state, "picking_dialog_open", False)
def _push_pick_state(
*,
dialog_open: Optional[bool] = None,
dialog_text: Optional[str] = None,
debug_text: Optional[str] = None,
) -> None:
changed_keys: list[str] = []
if dialog_open is not None:
state.picking_dialog_open = bool(dialog_open)
changed_keys.append("picking_dialog_open")
if dialog_text is not None:
state.picking_dialog_text = str(dialog_text)
changed_keys.append("picking_dialog_text")
if debug_text is not None:
state.picking_debug_last = str(debug_text)
changed_keys.append("picking_debug_last")
state.picking_debug_count = int(self._pick_debug_count)
changed_keys.append("picking_debug_count")
if hasattr(state, "dirty"):
for key in changed_keys:
try:
state.dirty(key)
except Exception:
break
if hasattr(state, "flush"):
try:
state.flush()
except Exception:
pass
def _extract_exact_cell_id_from_event(event_payload=None, **kwargs) -> Optional[int]:
def _to_int(value: object) -> Optional[int]:
try:
arr = np.asarray(value).ravel()
if arr.size == 0:
return None
cell_id = int(np.rint(float(arr[0])))
except (TypeError, ValueError):
return None
return cell_id if cell_id >= 0 else None
def _walk(obj):
if isinstance(obj, dict):
yield obj
for v in obj.values():
yield from _walk(v)
elif isinstance(obj, (list, tuple)):
for item in obj:
yield from _walk(item)
id_keys = (
"vtkOriginalCellIds",
"vtkOriginalCellId",
"vtkCellIds",
"cell_ids",
"cellIds",
"cell_id",
"cellId",
"id",
)
containers = []
if event_payload is not None:
containers.append(event_payload)
if kwargs:
containers.append(kwargs)
for container in containers:
for mapping in _walk(container):
for key in id_keys:
if key in mapping:
maybe_id = _to_int(mapping[key])
if maybe_id is not None:
return maybe_id
return None
def _extract_world_position_from_event(event_payload=None, **kwargs) -> Optional[np.ndarray]:
def _to_xyz(value: object) -> Optional[np.ndarray]:
if isinstance(value, dict):
if {"x", "y", "z"} <= set(value.keys()):
try:
return np.asarray(
[float(value["x"]), float(value["y"]), float(value["z"])],
dtype=float,
)
except (TypeError, ValueError):
return None
try:
arr = np.asarray(value, dtype=float).ravel()
except (TypeError, ValueError):
return None
if arr.size < 3:
return None
return arr[:3]
def _walk(obj):
if isinstance(obj, dict):
yield obj
for v in obj.values():
yield from _walk(v)
elif isinstance(obj, (list, tuple)):
for item in obj:
yield from _walk(item)
position_keys = (
"worldPosition",
"world_position",
"position",
"coords",
"point",
"xyz",
)
containers = []
if event_payload is not None:
containers.append(event_payload)
if kwargs:
containers.append(kwargs)
for container in containers:
for mapping in _walk(container):
for key in position_keys:
if key in mapping:
pos = _to_xyz(mapping[key])
if pos is not None:
return pos
if {"x", "y", "z"} <= set(mapping.keys()):
pos = _to_xyz(mapping)
if pos is not None:
return pos
return None
def _extract_display_position_from_event(event_payload=None, **kwargs) -> Optional[tuple[float, float]]:
def _to_xy(value: object) -> Optional[tuple[float, float]]:
if isinstance(value, dict):
if {"x", "y"} <= set(value.keys()):
try:
return (float(value["x"]), float(value["y"]))
except (TypeError, ValueError):
return None
try:
arr = np.asarray(value, dtype=float).ravel()
except (TypeError, ValueError):
return None
if arr.size < 2:
return None
return (float(arr[0]), float(arr[1]))
def _to_rect(value: object) -> Optional[dict[str, float]]:
if not isinstance(value, dict):
return None
required = ("left", "top", "width", "height")
if not all(k in value for k in required):
return None
try:
return {
"left": float(value["left"]),
"top": float(value["top"]),
"width": float(value["width"]),
"height": float(value["height"]),
}
except (TypeError, ValueError):
return None
def _walk(obj):
if isinstance(obj, dict):
yield obj
for v in obj.values():
yield from _walk(v)
elif isinstance(obj, (list, tuple)):
for item in obj:
yield from _walk(item)
pos_keys = (
"position",
"displayPosition",
"display_position",
"mouse",
"xy",
"display_xy",
"displayXY",
"canvasXY",
)
x_keys = ("offsetX", "clientX", "x", "screenX", "pageX")
y_keys = ("offsetY", "clientY", "y", "screenY", "pageY")
containers = []
if event_payload is not None:
containers.append(event_payload)
if kwargs:
containers.append(kwargs)
for container in containers:
for mapping in _walk(container):
for key in pos_keys:
if key in mapping:
maybe = _to_xy(mapping[key])
if maybe is not None:
return maybe
rect = None
for rect_key in ("targetRect", "target_rect", "rect", "boundingRect", "bounding_rect"):
if rect_key in mapping:
rect = _to_rect(mapping[rect_key])
if rect is not None:
break
x_val = None
y_val = None
for k in x_keys:
if k in mapping:
x_val = mapping[k]
break
for k in y_keys:
if k in mapping:
y_val = mapping[k]
break
if x_val is not None and y_val is not None:
if (
rect is not None
and ("offsetX" not in mapping and "offsetY" not in mapping)
and ("clientX" in mapping or "pageX" in mapping)
):
try:
x_val = float(x_val) - rect["left"]
y_val = float(y_val) - rect["top"]
except (TypeError, ValueError):
pass
maybe = _to_xy((x_val, y_val))
if maybe is not None:
return maybe
return None
def _summarize_pick_payload(event_payload=None, **kwargs) -> str:
keys: set[str] = set()
if isinstance(event_payload, dict):
keys.update(str(k) for k in event_payload.keys())
if kwargs:
keys.update(str(k) for k in kwargs.keys())
if not keys:
return "keys=none"
top = ",".join(sorted(keys)[:8])
return f"keys={top}"
def _resolve_cell_id_from_world_position(world_pos: np.ndarray) -> Optional[int]:
mesh = self._active_display_mesh
if mesh is None or mesh.n_cells <= 0:
return None
centers = mesh.cell_centers().points
if centers.size == 0:
return None
deltas = centers - np.asarray(world_pos, dtype=float)
distances = np.einsum("ij,ij->i", deltas, deltas)
nearest_local = int(np.argmin(distances))
if "vtkOriginalCellIds" in mesh.cell_data:
try:
return int(np.rint(float(mesh.cell_data["vtkOriginalCellIds"][nearest_local])))
except (TypeError, ValueError, IndexError):
return None
return nearest_local
def _resolve_cell_id_from_display_position(display_xy: tuple[float, float]) -> Optional[int]:
try:
vtk_mod = pv._vtk
picker = vtk_mod.vtkCellPicker()
except Exception:
return None
renderer = getattr(self.plotter, "renderer", None)
if renderer is None:
try:
renderers = getattr(self.plotter, "renderers", None)
if renderers is not None and len(renderers) > 0:
renderer = renderers[0]
except Exception:
renderer = None
if renderer is None:
return None
x, y = float(display_xy[0]), float(display_xy[1])
h = None
try:
h = int(getattr(self.plotter, "window_size", [0, 0])[1])
except Exception:
h = None
candidates = [(x, y)]
if h and h > 0:
candidates.append((x, float(max(h - y, 0))))
for cx, cy in candidates:
try:
success = int(picker.Pick(cx, cy, 0.0, renderer))
except Exception:
continue
if success <= 0:
continue
local_id = int(picker.GetCellId())
if local_id < 0:
continue
mesh = self._active_display_mesh
if mesh is not None and "vtkOriginalCellIds" in mesh.cell_data:
try:
return int(np.rint(float(mesh.cell_data["vtkOriginalCellIds"][local_id])))
except (TypeError, ValueError, IndexError):
return None
return local_id
return None
def _extract_display_position_from_interactor() -> Optional[tuple[float, float]]:
iren = getattr(self.plotter, "iren", None)
if iren is None:
return None
try:
pos = iren.GetEventPosition()
except Exception:
return None
if pos is None:
return None
try:
arr = np.asarray(pos, dtype=float).ravel()
except (TypeError, ValueError):
return None
if arr.size < 2:
return None
return (float(arr[0]), float(arr[1]))
def _build_pick_message(cell_id: int) -> Optional[str]:
if self.state is None:
return None
if cell_id < 0 or cell_id >= self.state.mesh.n_cells:
return None
cell_centers = self.state.mesh.cell_centers().points
centroid = cell_centers[cell_id]
centroid_str = f"({centroid[0]:.1f}, {centroid[1]:.1f}, {centroid[2]:.1f})"
values: dict[str, object] = {}
for attr in self.state.attributes:
raw_value = self.state.mesh.cell_data[attr][cell_id]
if attr in self.state.categorical_mappings:
if pd.isna(raw_value):
values[attr] = "<NA>"
else:
code = int(np.rint(float(raw_value)))
values[attr] = self.state.categorical_mappings[attr].get(code, f"<unknown:{code}>")
else:
values[attr] = raw_value
return f"Cell ID: {cell_id}, {centroid_str}, " + ", ".join(f"{k}: {v}" for k, v in values.items())
def _extract_client_key(event_payload=None, **kwargs) -> str:
values = []
if isinstance(event_payload, dict):
values.extend(
[
event_payload.get("key"),
event_payload.get("keySym"),
event_payload.get("keysym"),
event_payload.get("code"),
event_payload.get("keyCode"),
]
)
values.extend(
[
kwargs.get("key"),
kwargs.get("keySym"),
kwargs.get("keysym"),
kwargs.get("code"),
kwargs.get("keyCode"),
]
)
for value in values:
if value is None:
continue
if isinstance(value, (int, float)):
code = int(value)
if 0 <= code <= 255:
return chr(code).lower()
key_text = str(value).strip().lower()
if key_text.startswith("key") and len(key_text) == 4:
return key_text[-1]
if key_text:
return key_text
return ""
def _on_zup_key_down(event=None, **kwargs):
key = _extract_client_key(event, **kwargs)
if key and key != self.z_up_hotkey:
return
if hasattr(self.plotter, 'hotkey_pressed'):
self.plotter.hotkey_pressed['z'] = True
self.plotter.enforce_z_up()
self.plotter.render()
if self._remote_view is not None:
self._remote_view.update()
def _on_zup_key_up(event=None, **kwargs):
key = _extract_client_key(event, **kwargs)
if key and key != self.z_up_hotkey:
return
if hasattr(self.plotter, 'hotkey_pressed'):
self.plotter.hotkey_pressed['z'] = False
def _on_key_down(event=None, **kwargs):
key = _extract_client_key(event, **kwargs)
if not key:
return
if key == self.z_up_hotkey:
_on_zup_key_down(event, **kwargs)
def _on_pick_click(event=None, _source: str = "bridge", **kwargs):
if not bool(getattr(self.plotter, "picking_enabled", False)):
return
self._pick_debug_count += 1
cell_id = _extract_exact_cell_id_from_event(event, **kwargs)
world_pos = _extract_world_position_from_event(event, **kwargs)
if cell_id is None and world_pos is not None:
cell_id = _resolve_cell_id_from_world_position(world_pos)
display_xy = _extract_display_position_from_event(event, **kwargs)
if display_xy is None:
display_xy = _extract_display_position_from_interactor()
if cell_id is None and display_xy is not None:
cell_id = _resolve_cell_id_from_display_position(display_xy)
payload_summary = _summarize_pick_payload(event, **kwargs)
self._pick_debug_last = (
f"bridge count={self._pick_debug_count}, source={_source}, cell_id={cell_id}, "
f"world_pos={None if world_pos is None else np.round(world_pos, 3).tolist()}, "
f"display_xy={None if display_xy is None else [round(display_xy[0], 2), round(display_xy[1], 2)]}, "
f"{payload_summary}"
)
if cell_id is None:
_push_pick_state(dialog_open=False, dialog_text="", debug_text=self._pick_debug_last)
return
msg = _build_pick_message(cell_id)
if not msg:
_push_pick_state(dialog_open=False, dialog_text="", debug_text=self._pick_debug_last)
return
_push_pick_state(dialog_open=True, dialog_text=msg, debug_text=self._pick_debug_last)
def _on_pick_click_client(event=None, **kwargs):
_on_pick_click(event, _source="client", **kwargs)
ctrl.zup_key_down = _on_zup_key_down
ctrl.zup_key_up = _on_zup_key_up
ctrl.key_down = _on_key_down
ctrl.pick_click = _on_pick_click
ctrl.pick_click_client = _on_pick_click_client
if self.asset_catalog is not None:
self._register_asset_selector_handlers()
self._syncing_state = True
try:
self._sync_asset_selector_state()
finally:
self._syncing_state = False
def _load_data_source(**_):
if self._syncing_state:
return
if self._source_mode == "duckdb_query":
state.source_status = "DuckDB query mode is scaffolded; execution is not implemented yet."
return
try:
self.load_source_path(state.source_path_input)
except (FileNotFoundError, ValueError) as exc:
state.source_status = str(exc)
ctrl.load_data_source = _load_data_source
brand_logo_src = self._branding_logo_data_url()
with SinglePageWithDrawerLayout(server, show_drawer=self._drawer_default_open, width=340) as layout:
layout.toolbar.color = "grey lighten-3"
layout.toolbar.dense = True
layout.toolbar.dark = False
if hasattr(layout, "title") and hasattr(layout.title, "set_text"):
layout.title.set_text("")
with layout.toolbar:
vuetify.VImg(src=brand_logo_src, max_width=50, contain=True, classes="mr-2")
vuetify.VToolbarTitle(self.app_name)
vuetify.VSpacer()
vuetify.VChip(
"{{ model_name }}",
label=True,
small=True,
outlined=True,
classes="mr-2",
)
with layout.drawer:
with vuetify.VSheet(classes="pa-2 fill-height"):
with vuetify.VExpansionPanels(
v_model=("source_panel", [1]),
multiple=True,
focusable=True,
flat=True,
classes="mb-2",
):
with vuetify.VExpansionPanel():
vuetify.VExpansionPanelHeader(
"Data source",
dense=True,
classes="py-1 px-2 text-subtitle-2",
)
with vuetify.VExpansionPanelContent(classes="pt-1 pb-2 px-2"):
vuetify.VTextField(
v_model=("source_path_input", self._source_path),
label="Server path (.pbm file or hive directory)",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
)
vuetify.VBtn(
"Load",
click=ctrl.load_data_source,
block=True,
color="primary",
classes="mt-2",
)
vuetify.VChip(
"{{ source_mode }}",
label=True,
small=True,
outlined=True,
classes="mt-2",
)
vuetify.VChip(
"{{ source_status }}",
label=True,
small=True,
outlined=True,
classes="mt-2 text-caption text-truncate",
style="max-width: 100%;",
)
with vuetify.VExpansionPanel(v_if=("asset_selector_enabled", False)):
vuetify.VExpansionPanelHeader(
"Asset selector",
dense=True,
classes="py-1 px-2 text-subtitle-2",
)
with vuetify.VExpansionPanelContent(classes="pt-1 pb-2 px-2"):
for idx, key in enumerate(self._asset_level_keys):
vuetify.VSelect(
v_model=(
f"asset_level_{idx}_value",
self._asset_level_values.get(key, ""),
),
items=(f"asset_level_{idx}_options", []),
label=key,
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
change=getattr(ctrl, f"update_asset_level_{idx}"),
)
vuetify.VSelect(
v_model=("selected_asset_name", self._selected_asset_name),
items=("asset_name_options", []),
label="PBM name",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
change=ctrl.update_asset_name,
)
vuetify.VChip(
"{{ asset_selected_path }}",
label=True,
small=True,
outlined=True,
classes="text-caption text-truncate mt-2",
style="max-width: 100%;",
)
with vuetify.VExpansionPanels(
v_model=("control_panel", 0),
multiple=False,
focusable=True,
flat=True,
):
with vuetify.VExpansionPanel():
vuetify.VExpansionPanelHeader(
"Data Filter",
dense=True,
classes="py-1 px-2 text-subtitle-2",
)
with vuetify.VExpansionPanelContent(classes="pt-1 pb-2 px-2"):
vuetify.VSelect(
v_model=("data_filter_1_attribute", self._data_filters[0].attribute),
items=("data_filter_attribute_options", self._filter_attribute_options),
label="Filter 1 attribute",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
clearable=True,
change=ctrl.update_data_filter_1_attribute,
)
vuetify.VRangeSlider(
v_if=("!data_filter_1_is_categorical",),
v_model=("data_filter_1_range", list(self._data_filters[0].range_values)),
min=("data_filter_1_range_min", self._data_filters[0].minimum),
max=("data_filter_1_range_max", self._data_filters[0].maximum),
step=("data_filter_1_range_step", self._data_filters[0].step),
label="Filter 1 range",
dense=True,
hide_details=True,
thumb_label=False,
classes="mt-2",
disabled=("data_filter_1_attribute === ''",),
change=ctrl.update_data_filter_1_range,
)
vuetify.VSelect(
v_if=("data_filter_1_is_categorical",),
v_model=("data_filter_1_selected_categories", list(self._data_filters[0].selected_categories)),
items=("data_filter_1_category_options", list(self._data_filters[0].category_options)),
label="Filter 1 categories",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
multiple=True,
chips=True,
deletable_chips=True,
change=ctrl.update_data_filter_1_categories,
)
vuetify.VBtn(
"Reset filter 1",
click=ctrl.reset_data_filter_1,
block=True,
color="primary",
classes="mt-2",
)
vuetify.VChip(
"{{ data_filter_1_summary }}",
label=True,
small=True,
outlined=True,
classes="mt-2 text-caption text-truncate",
style="max-width: 100%;",
)
vuetify.VSelect(
v_model=("data_filter_2_attribute", self._data_filters[1].attribute),
items=("data_filter_attribute_options", self._filter_attribute_options),
label="Filter 2 attribute",
dense=True,
hide_details=True,
outlined=True,
classes="mt-4",
clearable=True,
change=ctrl.update_data_filter_2_attribute,
)
vuetify.VRangeSlider(
v_if=("!data_filter_2_is_categorical",),
v_model=("data_filter_2_range", list(self._data_filters[1].range_values)),
min=("data_filter_2_range_min", self._data_filters[1].minimum),
max=("data_filter_2_range_max", self._data_filters[1].maximum),
step=("data_filter_2_range_step", self._data_filters[1].step),
label="Filter 2 range",
dense=True,
hide_details=True,
thumb_label=False,
classes="mt-2",
disabled=("data_filter_2_attribute === ''",),
change=ctrl.update_data_filter_2_range,
)
vuetify.VSelect(
v_if=("data_filter_2_is_categorical",),
v_model=("data_filter_2_selected_categories", list(self._data_filters[1].selected_categories)),
items=("data_filter_2_category_options", list(self._data_filters[1].category_options)),
label="Filter 2 categories",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
multiple=True,
chips=True,
deletable_chips=True,
change=ctrl.update_data_filter_2_categories,
)
vuetify.VBtn(
"Reset filter 2",
click=ctrl.reset_data_filter_2,
block=True,
color="primary",
classes="mt-2",
)
vuetify.VChip(
"{{ data_filter_2_summary }}",
label=True,
small=True,
outlined=True,
classes="mt-2 text-caption text-truncate",
style="max-width: 100%;",
)
with vuetify.VExpansionPanel():
vuetify.VExpansionPanelHeader(
"Controls",
dense=True,
classes="py-1 px-2 text-subtitle-2",
)
with vuetify.VExpansionPanelContent(classes="pt-1 pb-2 px-2"):
vuetify.VSelect(
v_model=("active_attribute", self.state.scalar if self.state else ""),
items=("attribute_options", self.blockmodel.available_attributes if self.state else []),
label="Attribute",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
change=ctrl.update_attribute,
)
vuetify.VSelect(
v_model=("selected_colormap", self.colormap),
items=("colormap_options", self.available_colormaps),
label="Colormap",
dense=True,
hide_details=True,
outlined=True,
classes="mt-2",
change=ctrl.update_colormap,
)
vuetify.VCheckbox(
v_model=("discretize_deciles", self.discretize_deciles),
label="Discretise to deciles",
dense=True,
hide_details=True,
classes="mt-2",
change=ctrl.update_discretize_deciles,
)
vuetify.VSlider(
v_model=("threshold", self.threshold.value),
min=("threshold_min", self.threshold.minimum),
max=("threshold_max", self.threshold.maximum),
step=("threshold_step", self.threshold.step),
label="Threshold",
dense=True,
hide_details=True,
thumb_label=False,
outlined=True,
classes="mt-2",
change=ctrl.update_threshold,
)
vuetify.VTextField(
v_model=("threshold_display", self._format_threshold(self.threshold.value)),
label="Threshold value",
dense=True,
outlined=True,
hide_details=True,
classes="mt-2",
)
vuetify.VBtn(
"Apply",
click=ctrl.apply_threshold_text,
block=True,
color="primary",
classes="mt-2",
)
vuetify.VBtn(
"Reset threshold",
click=ctrl.reset_filter,
block=True,
color="primary",
classes="mt-2",
)
with vuetify.VExpansionPanel():
vuetify.VExpansionPanelHeader(
"View status",
dense=True,
classes="py-1 px-2 text-subtitle-2",
)
with vuetify.VExpansionPanelContent(classes="pt-1 pb-2 px-2"):
vuetify.VChip("Read-on-demand", color="success", small=True, outlined=True)
vuetify.VSpacer()
vuetify.VCheckbox(
v_model=("filter_active", self.filter_enabled),
label="Filter active",
readonly=True,
dense=True,
hide_details=True,
)
vuetify.VCheckbox(
v_model=("picking_active", bool(getattr(self.plotter, "picking_enabled", False))),
label="Picking active",
dense=True,
hide_details=True,
change=ctrl.update_picking_active,
)
vuetify.VCheckbox(
v_model=("show_model_bounds", self.show_model_bounds),
label="Show model bounds",
dense=True,
hide_details=True,
change=ctrl.update_show_model_bounds,
)
vuetify.VChip(
"Pick events: {{ picking_debug_count }}",
label=True,
small=True,
outlined=True,
classes="mt-2",
)
vuetify.VChip(
"{{ picking_debug_last }}",
label=True,
small=True,
outlined=True,
classes="mt-2 text-caption text-truncate",
style="max-width: 100%;",
)
with layout.content:
layout.content.classes = "pa-0 ma-0"
layout.content.style = "height: calc(100vh - 64px); overflow: hidden;"
if trame_widgets is not None:
key_trap = trame_widgets.MouseTrap(
ZUpKeyDown=ctrl.zup_key_down,
ZUpKeyUp=ctrl.zup_key_up,
)
key_trap.bind(self.z_up_hotkey, "ZUpKeyDown", listen_to="keydown")
key_trap.bind(self.z_up_hotkey, "ZUpKeyUp", listen_to="keyup")
with vuetify.VDialog(v_model=("picking_dialog_open", False), max_width=700):
with vuetify.VCard():
vuetify.VCardTitle("Picked Cell")
vuetify.VCardText(
"{{ picking_dialog_text }}",
style="white-space: pre-wrap; word-break: break-word;",
)
with vuetify.VSheet(classes="pa-2 d-flex justify-end"):
vuetify.VBtn(
"Close",
click=ctrl.close_picking_dialog,
color="primary",
text=True,
)
remote_view_kwargs = {
"interactive_ratio": 1,
"style": "width: 100%; height: 100%; min-height: 600px;",
"interactor_events": (
"pbm_interactor_events",
[
"KeyDown",
"KeyPress",
"KeyUp",
"LeftButtonPress",
"LeftButtonRelease",
],
),
"KeyDown": ctrl.key_down,
"KeyPress": ctrl.zup_key_down,
"KeyUp": ctrl.zup_key_up,
"LeftButtonPress": ctrl.pick_click,
}
with vuetify.VSheet(
style="width: 100%; height: 100%;",
click=(
ctrl.pick_click_client,
"""[{
clientX: $event.clientX,
clientY: $event.clientY,
offsetX: $event.offsetX,
offsetY: $event.offsetY,
targetRect: (
$event.target && $event.target.getBoundingClientRect
) ? {
left: $event.target.getBoundingClientRect().left,
top: $event.target.getBoundingClientRect().top,
width: $event.target.getBoundingClientRect().width,
height: $event.target.getBoundingClientRect().height
} : null
}]""",
),
):
self._remote_view = vtk.VtkRemoteView(
self.plotter.ren_win,
**remote_view_kwargs,
)
self._remote_view.update()
state.ready()
server.start(**self._build_launch_kwargs(port, host))
return server
