in databao/core/thread.py [0:0]
def _materialize_visualization(self, request: str | None, rows_limit: int | None) -> "VisualisationResult":
"""Materialize latest visualization for the given request and current data."""
data = self._materialize_data(rows_limit)
if self._visualization_result is None or request != self._visualization_request:
# TODO Cache visualization results as in Executor.execute()?
stream = self._stream_plot if self._stream_plot is not None else self._default_stream_plot
self._visualization_result = self._agent.visualizer.visualize(request, data, stream=stream)
self._visualization_request = request
self._meta.update(self._visualization_result.meta)
self._meta["plot_code"] = self._visualization_result.code # maybe worth to expand as a property later
if self._visualization_result is None:
raise RuntimeError("_visualization_result is None after materialization")
return self._visualization_result