Observable#

The ngapp.observable module provides reactive state management for ngapp applications. An Observable holds a single value and notifies registered listeners whenever the value changes. Pass an Observable directly as a component prop (e.g. ui_model_value=my_observable) for automatic two-way binding.

Why Observable?#

A common challenge in interactive applications is keeping multiple concerns synchronised: a checkbox in the UI, a keyboard shortcut, a persistence layer, and the underlying domain logic may all read or write the same piece of state. Without a shared mechanism, each pair requires manual glue code and guard flags to prevent feedback loops.

Observable solves this by providing a single source of truth that any number of consumers can subscribe to. When the value changes — regardless of which consumer changed it — every other subscriber is notified.

Quick start #

from ngapp.observable import Observable
from ngapp.components import QCheckbox

# 1. Declare an observable
visible = Observable(True, "visible")

# 2. React to changes
visible.on_change(lambda new, old: print(f"visible: {old} -> {new}"))

# 3. Pass it directly as a prop — two-way binding is automatic
checkbox = QCheckbox("Show", ui_model_value=visible)

# 4. Any mutation is propagated everywhere
visible.value = False   # prints "visible: True -> False", unchecks the checkbox
visible.toggle()        # prints "visible: False -> True", checks the checkbox

Creating an Observable #

from ngapp.observable import Observable

# Required: initial value.  Optional: name for debugging / serialisation.
scale = Observable(1.0, "scale")
enabled = Observable(False, "enabled")
items = Observable([], "items")

The name parameter has no functional effect at runtime. It is used by snapshot() and restore() as the dictionary key, and appears in the repr() output.

Type coercion with `converter`#

Pass a converter function to automatically coerce incoming values. If the converter raises ValueError or TypeError, the assignment is silently ignored and the value stays unchanged:

scale = Observable(1.0, "scale", converter=float)

scale.value = "3.14"   # converted to 3.14
scale.value = "abc"    # silently ignored, value stays 3.14
scale.value = 5        # converted to 5.0

This is especially useful with NumberInput (or any QInput with ui_type="number"), where the widget emits string values:

scale = Observable(1.0, "scale", converter=float)
inp = NumberInput(ui_model_value=scale, ui_label="Scale")
# User types "2.5" → observable receives 2.5 as float

Display formatting with `formatter`#

Pass a formatter function to control how the value is presented in UI widgets. The stored value is not affected — only the display representation changes. This is the counterpart to converter:

converter: Widget → Observable (parse user input)
formatter: Observable → Widget (format for display)

colormap_min = Observable(
    0.000007957,
    "colormap_min",
    converter=float,
    formatter=lambda v: f"{v:.4g}",
)

colormap_min.value          # 7.957394756115198e-06 (raw float)
colormap_min.display_value  # "7.957e-06" (formatted string)

# Bind to a QInput — the widget shows "7.957e-06"
inp = QInput(ui_model_value=colormap_min, ui_type="number")

# When autoscale updates the value:
colormap_min.value = 0.00000000123
# Widget automatically shows "1.23e-09" (formatter applied)

# When the user types "0.5" in the input:
# converter parses it → stored as 0.5
# formatter presents it → widget shows "0.5"

The display_value property is also available for manual use:

label.ui_children = [f"Range: {obs_min.display_value} – {obs_max.display_value}"]

Reading and writing #

Read and write through the value property:

print(scale.value)    # 1.0
scale.value = 2.5     # listeners are called
scale.value = 2.5     # no notification (value unchanged)

The setter compares the new value to the current one with ==. If they are equal, no listeners are invoked.

Listening to changes #

Register a callback with on_change(). The callback receives (new_value, old_value):

def on_scale(new, old):
    print(f"scale changed from {old} to {new}")

dispose = scale.on_change(on_scale)

scale.value = 3.0     # prints "scale changed from 2.5 to 3.0"

dispose()             # removes the listener
scale.value = 4.0     # on_scale is NOT called

on_change returns a dispose function. Call it to unsubscribe. Calling dispose more than once is safe.

Multiple listeners are supported; they fire in registration order.

Toggling booleans #

For boolean observables, toggle() is a shorthand for inverting the current value:

enabled = Observable(False, "enabled")
enabled.toggle()      # enabled.value is now True
enabled.toggle()      # enabled.value is now False

Widget binding #

Direct prop binding #

Pass an Observable directly as any component prop. For ui_model_value, binding is two-way: changes to the observable update the widget, and user interaction updates the observable. For all other props, binding is one-way (observable → widget).

from ngapp.observable import Observable
from ngapp.components import QSlider, QBtn

opacity = Observable(1.0, "opacity")
color = Observable("primary", "color")

# Two-way: dragging the slider updates opacity.value, and vice versa
slider = QSlider(ui_model_value=opacity, ui_min=0.0, ui_max=1.0, ui_step=0.01)

# One-way: changing color.value updates the button color
btn = QBtn(ui_label="Click", ui_color=color)

opacity.value = 0.5   # slider moves to 0.5
color.value = "red"   # button turns red

Assigning an Observable to a prop after construction also works:

slider.ui_model_value = another_observable

The previous binding is automatically disposed and the new one takes effect.

Explicit binding with `bind()`#

For non-standard widget attributes or events, use bind() to wire an Observable to any attribute and event pair:

from ngapp.observable import bind

bind(my_obs, widget, widget_attr="ui_label", event="on_change")

bind returns a dispose function:

dispose = bind(opacity, slider)
# later…
dispose()

Batching changes #

When multiple observables must be updated together, observable_batch() defers all listener calls until the block exits. This avoids redundant intermediate work (for example, multiple render calls):

from ngapp.observable import observable_batch

with observable_batch():
    position.value = (1.0, 2.0, 3.0)
    scale.value = 0.5
    enabled.value = True
# All listeners are called here, once per changed observable.

Batches can be nested. Listeners fire only when the outermost batch completes.

Serialisation helpers #

Three functions make it easy to save and restore observable state without writing per-field boilerplate.

`snapshot()`#

Collect all Observable values on an object into a plain dictionary, keyed by their name:

class MyComponent:
    def __init__(self):
        self.scale = Observable(1.0, "scale")
        self.visible = Observable(True, "visible")

comp = MyComponent()
data = snapshot(comp)
# {"scale": 1.0, "visible": True}

`restore()`#

Write values from a dictionary back into the matching observables. Unknown keys are silently ignored:

restore(comp, {"scale": 2.5, "visible": False})
# comp.scale.value == 2.5
# comp.visible.value == False

Because restore sets each .value, all listeners fire as usual.

`collect_observables()`#

Return a {name: Observable} mapping for programmatic access:

obs = collect_observables(comp)
for name, o in obs.items():
    print(f"{name} = {o.value}")

Typical usage patterns #

Reactive side-effects #

Wire domain logic to observables so that any change — from the UI, a keyboard shortcut, or code — triggers the correct response:

class Viewer:
    def __init__(self):
        self.wireframe = Observable(True, "wireframe")
        self.wireframe.on_change(self._apply_wireframe)

    def _apply_wireframe(self, val, _old):
        self.renderer.show_wireframe = val
        self.scene.render()

    def toggle_wireframe(self):
        self.wireframe.toggle()   # one line — side-effects handled

Component with bound UI #

Declare observables in a component, bind them to widgets in a settings panel:

# Component
class SimView:
    def __init__(self, saved_settings):
        self.resolution = Observable(
            saved_settings.get("resolution", 256), "resolution"
        )
        self.resolution.on_change(self._on_resolution)

    def _on_resolution(self, val, _old):
        self.solver.set_resolution(val)
        self.solver.run()

# Settings panel
class SimSettings:
    def __init__(self, comp):
        slider = QSlider(
            ui_model_value=comp.resolution,
            ui_min=64, ui_max=1024, ui_step=64,
        )

Save and restore #

Persist the full observable state at save time, restore it when loading:

from ngapp.observable import snapshot, restore

# Save
settings = snapshot(comp)
storage.save("settings", settings)

# Restore (on next launch)
saved = storage.load("settings")
restore(comp, saved)

API reference #

Reactive observable values with change notification and two-way widget binding.

This module provides Observable, a generic container that holds a single typed value and notifies registered listeners whenever that value changes. It serves as a single source of truth for application state that may be read or written by multiple consumers (UI widgets, keyboard shortcuts, programmatic logic, persistence layers, etc.).

The companion bind() function establishes a two-way connection between an Observable and an ngapp Component widget, keeping the two in sync with a built-in re-entrancy guard.

observable_batch() allows grouping multiple value changes so that listeners are invoked only once per observable after all changes are applied.

Example #

from ngapp.observable import Observable, bind, observable_batch

visible = Observable(True, "visible")
visible.on_change(lambda new, old: scene.render())

checkbox = QCheckbox("Visible", ui_model_value=visible.value)
bind(visible, checkbox)

# Any source can change the value; all listeners and the widget
# are updated automatically.
visible.toggle()

class ngapp.observable.Observable( default: T, name: str = '', converter: Callable | None = None, formatter: Callable | None = None, )#

A single observable value with change notification.

Observable wraps a value of type T and maintains a list of listener callbacks. Assigning to value compares the new value against the current one (using ==); if they differ, every registered listener is called with (new_value, old_value).

Parameters#

defaultT: The initial value.
namestr, optional: A descriptive name used for debugging and serialisation keys.
convertercallable, optional: A function applied to every incoming value before it is stored. If the converter raises ValueError or TypeError, the assignment is silently ignored (the value stays unchanged). Useful for type coercion, e.g. converter=float.
formattercallable, optional: A function applied to the stored value when presenting it to UI widgets (via display_value). The raw value is unaffected. Useful for number formatting, e.g. formatter=lambda v: f"{v:.4g}".

Attributes#

valueT: The current value. Setting this property triggers change notification when the new value differs from the old one.
display_value: The formatted value for display. If no formatter is set, this is identical to value.

Example#

enabled = Observable(False, "enabled")

dispose = enabled.on_change(lambda new, old: print(f"{old} -> {new}"))
enabled.value = True   # prints: False -> True
enabled.value = True   # no output (value unchanged)
enabled.toggle()       # prints: True -> False

dispose()              # removes the listener

# Formatter example:
temp = Observable(0.000123, "temp", converter=float,
                  formatter=lambda v: f"{v:.4g}")
temp.display_value     # "0.000123" → "0.000123" wait no "1.23e-04"

property display_value#

The value formatted for display in UI widgets.

If a formatter was provided, returns formatter(value). Otherwise returns value unchanged.

on_change( cb: Callable[[T, T], None], ) → Callable[[], None]#

Register a listener that is called on every value change.

Parameters#

cbcallable(new_value, old_value): The callback to invoke when value changes.

Returns#

callable: A dispose function. Calling it removes cb from the listener list. Calling it more than once is safe.

Parameters:: cb (Callable[[T, T], None])
Return type:: Callable[[], None]

toggle() → None#

Invert the current value. Intended for boolean observables.

Return type:: None

property value: T#: The current value.

Parameters:

default (T)
name (str)
converter (Callable | None)
formatter (Callable | None)

ngapp.observable.bind( prop: Observable, widget, widget_attr: str = 'ui_model_value', event: str = 'on_update_model_value', ) → Callable[[], None]#

Establish a two-way binding between an Observable and a widget.

The binding synchronises the observable and the widget in both directions:

Observable -> Widget: When the observable value changes, the widget attribute widget_attr is updated.
Widget -> Observable: When the widget emits the event registered via event, the observable value is updated.

An internal re-entrancy guard ensures that a change originating on one side does not bounce back and cause an infinite loop.

Parameters#

propObservable: The observable to bind.
widgetComponent: An ngapp component instance (e.g. QCheckbox, QSlider).
widget_attrstr, optional: The widget attribute to read/write. Defaults to "ui_model_value".
eventstr, optional: The name of the widget method used to register an event handler. Defaults to "on_update_model_value".

Returns#

callable: A dispose function that removes the observable-to-widget listener.

Parameters:

prop (Observable)
widget_attr (str)
event (str)

Return type:

Callable[[], None]

ngapp.observable.collect_observables(obj) → dict[str, Observable]#

Return a {name: observable} dict of all Observable attributes on obj.

Inspects obj.__dict__ and returns every value that is an Observable instance, keyed by Observable._name.

Parameters#

objobject: The object to inspect.

Returns#

dict[str, Observable]: Mapping from observable name to instance.

Return type:: dict[str, Observable]

ngapp.observable.observable_batch()#

Defer Observable listener invocations until the block exits.

Within the managed block, setting Observable.value records pending notifications instead of dispatching them immediately. When the outermost block exits, every recorded callback is invoked exactly once in the order it was queued. Batches may be nested; listeners fire only when the outermost batch completes.

This is useful when multiple observables must be updated atomically to avoid redundant or intermediate side-effects.

Example#

with observable_batch():
    position.value = (1.0, 2.0, 3.0)
    scale.value = 0.5
# All listeners are called here, once per changed observable.

ngapp.observable.restore(obj, data: dict[str, Any]) → None#

Restore Observable values on obj from data.

For every observable on obj whose _name appears in data, the value is set (triggering listeners as usual). Unknown keys in data are silently ignored.

Parameters#

objobject: The object whose observables should be restored.
datadict[str, Any]: Mapping from observable name to value, as produced by snapshot().

Parameters:: data (dict[str, Any])
Return type:: None

ngapp.observable.snapshot(obj) → dict[str, Any]#

Return a plain {name: value} dict of all Observable values on obj.

This is the serialisation counterpart of restore().

Parameters#

objobject: The object whose observables should be snapshotted.

Returns#

dict[str, Any]: Mapping from observable name to its current value.

Return type:: dict[str, Any]