"""Element wiring manager for callback and state binding.
This module provides the ElementWiringManager class which handles wiring
callbacks for interactive elements, state bindings, and menu shortcuts.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from wijjit.autocomplete.resolver import resolve_autocomplete
from wijjit.core.events import HandlerScope
from wijjit.elements.display.link import Link
from wijjit.elements.display.tree import Tree
from wijjit.elements.input.button import Button
from wijjit.elements.input.checkbox import Checkbox, CheckboxGroup
from wijjit.elements.input.code_editor import CodeEditor
from wijjit.elements.input.datagrid import DataGrid
from wijjit.elements.input.radio import Radio, RadioGroup
from wijjit.elements.input.select import Select
from wijjit.elements.input.slider import Slider
from wijjit.elements.input.text import TextArea, TextInput
from wijjit.elements.input.toggle import Toggle
from wijjit.elements.menu import ContextMenu, DropdownMenu, MenuElement
from wijjit.logging_config import get_logger
from wijjit.tags.layout import resolve_bind_key
if TYPE_CHECKING:
from wijjit.core.app import Wijjit
from wijjit.core.state import State
from wijjit.elements.base import Element, ScrollableElement
logger = get_logger(__name__)
[docs]
class ElementWiringManager:
"""Manages element callback wiring and state binding.
This manager is responsible for:
- Wiring action callbacks for buttons
- Binding state to inputs (two-way data binding)
- Registering menu shortcuts
- Managing scroll position persistence
- Managing highlight state for selectable elements
Parameters
----------
app : Wijjit
Reference to the main application
Attributes
----------
app : Wijjit
Application reference
_registered_menuitem_shortcuts : set[str]
Set of registered menu item shortcut handler IDs
_registered_menu_shortcuts : set[str]
Set of registered menu trigger shortcut handler IDs
"""
[docs]
def __init__(self, app: Wijjit) -> None:
"""Initialize the element wiring manager.
Parameters
----------
app : Wijjit
Reference to the main application
"""
self.app = app
self._registered_menuitem_shortcuts: set[str] = set()
self._registered_menu_shortcuts: set[str] = set()
[docs]
def clear_view_shortcuts(self) -> None:
"""Clear shortcut tracking sets when navigating away from view.
This method clears the internal tracking sets used to prevent
duplicate handler registration within a single view render.
Note: With VIEW-scoped handlers, the actual handler cleanup is
handled automatically by handler_registry.clear_view(). This method
primarily resets the duplicate-detection sets for the next view.
"""
self._registered_menuitem_shortcuts.clear()
self._registered_menu_shortcuts.clear()
[docs]
def wire_elements(
self,
elements: list[Element],
state: State,
) -> None:
"""Wire callbacks for all elements.
This method wires up action callbacks, state bindings, and
menu shortcuts for all positioned elements.
Parameters
----------
elements : list[Element]
List of positioned elements to wire
state : State
Application state for bindings
"""
for elem in elements:
self._wire_element(elem, state)
# Group standalone radios by name so each knows its siblings. This
# drives mutual exclusion (select() deselects same-group siblings) and
# Up/Down arrow navigation, both of which rely on Radio.radio_group.
self._wire_radio_groups(elements)
def _wire_radio_groups(self, elements: list[Element]) -> None:
"""Link standalone radios that share a ``name`` into a group.
Parameters
----------
elements : list[Element]
Flat list of positioned elements for the current render.
"""
groups: dict[str, list[Radio]] = {}
for elem in elements:
if isinstance(elem, Radio) and elem.name:
groups.setdefault(elem.name, []).append(elem)
for siblings in groups.values():
for radio in siblings:
radio.radio_group = siblings
@staticmethod
def _bind_key(
elem: Any,
default_key: str | None = None,
*,
id_is_default: bool = True,
) -> str | None:
"""Resolve the state key this element binds to, or None.
The write half of binding. It delegates to the same
:func:`~wijjit.tags.layout.resolve_bind_key` the tags use for the read
half, so an element can never read one key and write another.
Parameters
----------
elem : Element
The element being wired. Its ``bind`` prop is a bool (bind to the
default key / do not bind) or a state key name.
default_key : str or None, optional
The default key when it is not the element's id. Radio and
RadioGroup pass their group ``name``.
id_is_default : bool, optional
Whether the id may serve as the default key. Radio and RadioGroup
pass False: a radio's key is its *group*, so an ungrouped radio
binds to nothing rather than falling back to an id that would give
each radio its own key and break the group's mutual exclusion.
Returns
-------
str or None
The state key, or None when this element does not bind.
"""
elem_id = getattr(elem, "id", None) if id_is_default else None
return resolve_bind_key(elem.bind, elem_id, default_key)
def _wire_element(self, elem: Element, state: State) -> None:
"""Wire callbacks for a single element.
Parameters
----------
elem : Element
Element to wire
state : State
Application state
"""
# Wire up action callbacks for buttons
if isinstance(elem, Button) and elem.action:
action_id = elem.action
# Pass the ActionEvent through, but use the action from the element
elem.on_activate = lambda event, aid=action_id: self.app._dispatch_action(
aid, event=event
)
# Wire up action callbacks for links
if isinstance(elem, Link) and elem.action:
action_id = elem.action
# Wire on_click to dispatch the action
elem.on_click = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up TextInput callbacks
if isinstance(elem, TextInput):
self._wire_textinput(elem, state)
# Wire up TextArea callbacks
if isinstance(elem, TextArea):
self._wire_textarea(elem, state)
# Wire up CodeEditor render callback (CodeEditor extends TextArea)
if isinstance(elem, CodeEditor):
self._wire_code_editor(elem)
# Wire up Select callbacks
if isinstance(elem, Select):
self._wire_select(elem, state)
# Wire up scroll position persistence for all ScrollableElement elements
from wijjit.elements.base import ScrollableElement
if isinstance(elem, ScrollableElement):
self._wire_scrollable(elem, state)
# Wire up Tree callbacks
if isinstance(elem, Tree):
self._wire_tree(elem, state)
# Wire up Checkbox callbacks
if isinstance(elem, Checkbox):
self._wire_checkbox(elem, state)
# Wire up Radio callbacks
if isinstance(elem, Radio):
self._wire_radio(elem, state)
# Wire up CheckboxGroup callbacks
if isinstance(elem, CheckboxGroup):
self._wire_checkbox_group(elem, state)
# Wire up RadioGroup callbacks
if isinstance(elem, RadioGroup):
self._wire_radio_group(elem, state)
# Wire up Slider callbacks
if isinstance(elem, Slider):
self._wire_slider(elem, state)
# Wire up Toggle callbacks
if isinstance(elem, Toggle):
self._wire_toggle(elem, state)
# Wire up DataGrid callbacks. DataGrid subclasses ScrollableElement, so
# this runs after _wire_scrollable above and must not disturb its
# scroll-persistence callbacks - it only claims on_data_change.
if isinstance(elem, DataGrid):
self._wire_datagrid(elem, state)
def _wire_textinput(self, elem: TextInput, state: State) -> None:
"""Wire TextInput callbacks.
Parameters
----------
elem : TextInput
TextInput element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
# Initialize element value from state if key exists
# Note: cursor_pos is now preserved by reconciliation (ephemeral state)
if bind_key in state:
elem.value = str(state[bind_key])
# Set up two-way binding
def on_change_handler(old_val, new_val, key=bind_key):
# Update state when element changes
state[key] = new_val
elem.on_change = on_change_handler
# Resolve autocomplete spec to completer if needed
if elem._autocomplete_spec is not None and elem.completer is None:
elem.completer = resolve_autocomplete(
elem._autocomplete_spec,
elem.id,
self.app,
)
# Clear the spec after resolution
elem._autocomplete_spec = None
# Inject overlay manager for autocomplete popup display
if elem.completer is not None:
elem._overlay_manager = self.app.overlay_manager
def _wire_textarea(self, elem: TextArea, state: State) -> None:
"""Wire TextArea callbacks.
Parameters
----------
elem : TextArea
TextArea element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
# Note: Cursor, selection, and scroll state are now preserved by
# reconciliation (ephemeral state) - no need to save/restore manually
# Set up two-way binding for content only
def on_change_handler(old_val, new_val, key=bind_key):
# Update content state
state[key] = new_val
elem.on_change = on_change_handler
def _wire_code_editor(self, elem: CodeEditor) -> None:
"""Wire CodeEditor render callback.
Parameters
----------
elem : CodeEditor
CodeEditor element to wire
Notes
-----
This wires the _request_render callback so that after debounced
retokenization completes, the app triggers a re-render to display
the updated syntax highlighting.
"""
# Wire up render request callback for syntax highlighting updates
def request_render():
self.app.needs_render = True
elem._request_render = request_render
def _wire_select(self, elem: Select, state: State) -> None:
"""Wire Select callbacks.
Parameters
----------
elem : Select
Select element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
if elem.multiple:
# Multi-select mode: state holds a list
if bind_key in state:
state_value = state[bind_key]
if isinstance(state_value, (list, set, tuple)):
elem.selected_values = set(state_value)
elif state_value is not None:
elem.selected_values = {str(state_value)}
# Set up two-way binding for multi-select
def on_change_handler_multi(old_val, new_val, key=bind_key):
# Update state when element changes (new_val is a list)
state[key] = new_val
elem.on_change = on_change_handler_multi
else:
# Single-select mode: state holds a single value
if bind_key in state:
elem.value = state[bind_key]
# Update selected_index to match the value
elem.selected_index = elem._find_option_index(elem.value)
# Set up two-way binding for single-select
def on_change_handler_single(old_val, new_val, key=bind_key):
# Update state when element changes
state[key] = new_val
elem.on_change = on_change_handler_single
# Wire up highlighted_index persistence if element has the state key
if elem.highlight_state_key:
highlight_key = elem.highlight_state_key
def on_highlight_handler(new_index, hkey=highlight_key):
# Update state when highlight changes
state[hkey] = new_index
elem.on_highlight_change = on_highlight_handler
def _wire_scrollable(self, elem: ScrollableElement, state: State) -> None:
"""Wire scroll position persistence for ScrollableElement elements.
Parameters
----------
elem : ScrollableElement
Scrollable element to wire
state : State
Application state
Notes
-----
Handles both vertical and horizontal scroll callbacks for explicit
state tracking. Scroll position is now preserved automatically by
reconciliation (ephemeral state), so restore calls are not needed.
"""
# Wire vertical scroll callback for explicit state tracking
if elem.scroll_state_key:
scroll_key = elem.scroll_state_key
def on_scroll_handler(position, skey=scroll_key):
# Update state when scroll position changes
state[skey] = position
elem.on_scroll = on_scroll_handler
# Note: restore_scroll_position no longer needed - reconciliation
# preserves scroll position via ephemeral state
# Wire horizontal scroll callback for explicit state tracking
if elem.scroll_state_key_x:
scroll_key_x = elem.scroll_state_key_x
def on_scroll_x_handler(position, skey_x=scroll_key_x):
# Update state when horizontal scroll position changes
state[skey_x] = position
elem.on_scroll_x = on_scroll_x_handler
# Note: restore_scroll_position_x no longer needed - reconciliation
# preserves scroll position via ephemeral state
def _wire_tree(self, elem: Tree, state: State) -> None:
"""Wire Tree callbacks.
Parameters
----------
elem : Tree
Tree element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
def on_select_handler(node, aid=action_id):
# Dispatch action with node data
self.app._dispatch_action(aid, data=node)
elem.on_select = on_select_handler
def _wire_checkbox(self, elem: Checkbox, state: State) -> None:
"""Wire Checkbox callbacks.
Parameters
----------
elem : Checkbox
Checkbox element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
# Initialize element checked state from state if key exists
if bind_key in state:
elem.checked = bool(state[bind_key])
# Set up two-way binding
def on_change_handler(old_val, new_val, key=bind_key):
# Update state when element changes
state[key] = new_val
elem.on_change = on_change_handler
def _wire_radio(self, elem: Radio, state: State) -> None:
"""Wire Radio callbacks.
Parameters
----------
elem : Radio
Radio element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled. A radio's default key is its group
# name, never its id: every radio in a group reads and writes the one
# key holding the group's selection. id_is_default=False keeps an
# ungrouped radio unbound rather than giving it a key of its own.
bind_key = self._bind_key(elem, elem.name, id_is_default=False)
if bind_key:
# Initialize element checked state from the group's key
if bind_key in state:
elem.checked = state[bind_key] == elem.value
# Set up two-way binding
radio_value = elem.value
def on_change_handler(old_val, new_val, key=bind_key, rval=radio_value):
# Update state when radio is selected
if new_val: # Only update state when radio is selected (not deselected)
state[key] = rval
elem.on_change = on_change_handler
def _wire_checkbox_group(self, elem: CheckboxGroup, state: State) -> None:
"""Wire CheckboxGroup callbacks.
Parameters
----------
elem : CheckboxGroup
CheckboxGroup element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
# Initialize element selected values from state if key exists
if bind_key in state:
elem.selected_values = set(state[bind_key])
# Set up two-way binding
def on_change_handler(old_val, new_val, key=bind_key):
# Update state when element changes
state[key] = new_val
elem.on_change = on_change_handler
# Wire up highlighted_index persistence if element has the state key
if elem.highlight_state_key:
highlight_key = elem.highlight_state_key
def on_highlight_handler(new_index, hkey=highlight_key):
# Update state when highlight changes
state[hkey] = new_index
elem.on_highlight_change = on_highlight_handler
def _wire_radio_group(self, elem: RadioGroup, state: State) -> None:
"""Wire RadioGroup callbacks.
Parameters
----------
elem : RadioGroup
RadioGroup element to wire
state : State
Application state
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled. Like Radio, the group's key is its
# name (which the tag already defaults to the id), never the id itself.
bind_key = self._bind_key(elem, elem.name, id_is_default=False)
if bind_key:
# Initialize element selected value from the group's key
if bind_key in state:
elem.selected_value = state[bind_key]
elem.selected_index = elem._find_option_index(elem.selected_value)
# Set up two-way binding
def on_change_handler(old_val, new_val, key=bind_key):
# Update state when element changes
state[key] = new_val
elem.on_change = on_change_handler
# Wire up highlighted_index persistence if element has the state key
if elem.highlight_state_key:
highlight_key = elem.highlight_state_key
def on_highlight_handler(new_index, hkey=highlight_key):
# Update state when highlight changes
state[hkey] = new_index
elem.on_highlight_change = on_highlight_handler
def _wire_slider(self, elem: Slider, state: State) -> None:
"""Wire Slider callbacks.
Parameters
----------
elem : Slider
Slider element to wire
state : State
Application state
Notes
-----
Slider advertised ``bind=True`` and its tag read ``state[id]`` at
render, but no wiring ever subscribed to the ``on_change`` the element
was already firing - so dragging a slider moved the bar on screen and
left state untouched, forever. The read side alone is not a binding.
Slider has no ``on_action`` slot (unlike Toggle), so its ``action``
attribute stays unwired. That gap is pre-existing and separate from the
binding one closed here.
"""
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
if bind_key in state:
try:
elem.value = float(state[bind_key])
except (TypeError, ValueError) as e:
logger.warning(f"Failed to restore slider '{bind_key}': {e}")
def on_change_handler(old_val, new_val, key=bind_key):
# Update state when the slider moves
state[key] = new_val
elem.on_change = on_change_handler
def _wire_toggle(self, elem: Toggle, state: State) -> None:
"""Wire Toggle callbacks.
Parameters
----------
elem : Toggle
Toggle element to wire
state : State
Application state
Notes
-----
Same gap as Slider: the element fired ``on_change`` with nobody
listening, so a toggle never wrote its new value back to state.
"""
# Wire up action callback if action is specified
if elem.action:
action_id = elem.action
elem.on_action = lambda aid=action_id: self.app._dispatch_action(aid)
# Wire up state binding if enabled
bind_key = self._bind_key(elem)
if bind_key:
if bind_key in state:
elem.checked = bool(state[bind_key])
def on_change_handler(old_val, new_val, key=bind_key):
# Update state when the toggle flips
state[key] = new_val
elem.on_change = on_change_handler
def _wire_datagrid(self, elem: DataGrid, state: State) -> None:
"""Wire DataGrid state binding.
Parameters
----------
elem : DataGrid
DataGrid element to wire
state : State
Application state
Notes
-----
The write-back **must copy the rows**. ``DataGrid.set_cell`` mutates
``self.data`` in place and then fires ``on_data_change(self.data)`` -
the element's own live list. Storing that reference would make
``state[key] is elem.data``, aliasing application state to the
element's working list.
``State`` no longer goes *silent* on that alias - since review 2.9 it
detects the same-object write and fires a change rather than comparing
a mutated list against itself - but the copy is still required. An
aliased value mutates under the app with no write going through
``State`` at all, so a reader sees edits before they are committed,
and the ``old_value``/``new_value`` handed to every callback are the
same object, which defeats any callback that diffs them. Copying keeps
state's snapshot distinct from the element's working list.
Note also that DataGrid normalizes every cell to ``str`` on load, so
the rows written back here are strings regardless of what was seeded.
"""
bind_key = self._bind_key(elem)
if not bind_key:
return
def on_data_change_handler(rows, key=bind_key):
# Copy - see the note above on aliasing.
state[key] = [list(row) for row in rows]
elem.on_data_change = on_data_change_handler