ndv.controllers #

def __init__(
    self,
    data: Any | DataWrapper = None,
    /,
    *,
    viewer_options: ArrayViewerModel | ArrayViewerModelKwargs | None = None,
    display_model: ArrayDisplayModel | None = None,
    **kwargs: Unpack[ArrayDisplayModelKwargs],
) -> None:
    wrapper = None if data is None else DataWrapper.create(data)
    if display_model is None:
        display_model = self._default_display_model(wrapper, **kwargs)
    elif kwargs:
        warnings.warn(
            "When display_model is provided, kwargs are be ignored.",
            stacklevel=2,
        )

    self._display_model = display_model
    self._data_wrapper: DataWrapper | None = wrapper
    self._resolved = EMPTY_STATE

    self._connect_datawrapper(None, wrapper)

    self._viewer_model = ArrayViewerModel.model_validate(viewer_options or {})
    self._viewer_model.events.interaction_mode.connect(
        self._on_interaction_mode_changed
    )
    self._roi_model: RectangularROIModel | None = None

    app = _app.gui_frontend()

    # whether to fetch data asynchronously.  Not publicly exposed yet...
    # but can use 'NDV_SYNCHRONOUS' env var to set globally
    # jupyter doesn't need async because it's already async (in that the
    # GUI is already running in JS)
    NDV_SYNCHRONOUS = os.getenv("NDV_SYNCHRONOUS", "0") in {"1", "True", "true"}
    self._async = not NDV_SYNCHRONOUS and app != _app.GuiFrontend.JUPYTER
    # maps pending futures to their request generation (for stale detection)
    self._gen_counter = count()
    self._current_gen: int = 0
    self._futures: dict[Future[DataResponse], int] = {}

    # mapping of channel keys to their respective controllers
    # where None is the default channel
    self._lut_controllers: dict[ChannelKey, ChannelController] = {}

    # get and create the front-end and canvas classes
    frontend_cls = _app.get_array_view_class()
    canvas_cls = _app.get_array_canvas_class()
    self._canvas = canvas_cls(self._viewer_model)

    # TODO: Is this necessary?
    self._histograms: dict[ChannelKey, HistogramCanvas] = {}
    self._shared_histogram: SharedHistogramCanvas | None = None
    self._shared_histogram_links: dict[ChannelKey, _SharedHistogramLink] = {}
    self._view = frontend_cls(self._canvas.frontend_widget(), self._viewer_model)

    self._roi_view: RectangularROIHandle | None = None

    self._set_model_connected(self._display_model)
    self._canvas.set_ndim(self._display_model.n_visible_axes)

    self._view.currentIndexChanged.connect(self._on_view_current_index_changed)
    self._view.resetZoomClicked.connect(self._on_view_reset_zoom_clicked)
    self._view.histogramRequested.connect(self._add_histogram)
    self._view.sharedHistogramRequested.connect(self._add_shared_histogram)
    self._view.channelModeChanged.connect(self._on_view_channel_mode_changed)
    self._view.ndimToggleRequested.connect(self._on_view_ndim_toggle_requested)

    self._highlight_pos: tuple[float, float] | None = None
    self._canvas.mouseMoved.connect(self._on_canvas_mouse_moved)
    self._canvas.mouseLeft.connect(self._on_canvas_mouse_left)

    self._focused_slider_axis: AxisKey | None = None
    self._disconnect_key_events = _app.filter_key_events(
        self._view.frontend_widget(), self._view
    )
    self._view.keyPressed.connect(self._on_key_pressed)

    if self._data_wrapper is not None:
        self._fully_synchronize_view()