Signals.

A signal is how a node announces that something happened without knowing who cares. The emitter declares it, listeners connect to it, and emitting calls every connected listener in turn.

▶ Run in browser

Tags: basics signals events

Declare a signal as a class attribute, typed with what it carries:

class Emitter(Node):
    ticked = Signal(int)

    def tick(self):
        self.count += 1
        self.ticked.emit(self.count)   # fire the signal

Each instance gets its own signal on first access, so listeners on one Emitter never hear another.

  • signal.connect(fn) returns a Connection. Keep it to leave later: conn.disconnect() detaches the listener and conn.connected reports whether it is still attached.

  • signal.connect(fn, once=True) spends itself on the first emit and then reports connected False.

  • A bound method of a Node is held weakly. destroy() disconnects every connection made through the node’s methods, and one whose node was garbage collected is skipped and dropped on the next emit, so a freed listener never crashes the emitter.

The demo connects three listeners to one Emitter: a method on the root, a once-listener and the method of a separate Echo node. Every delivery is written to the log, and the status line reads each Connection back. Emit with Space or the Emit button, detach and re-attach the root listener with D, and free the Echo node with F; each button along the bottom presses the same action as its key.

Source

  1# /// simvx
  2# tags = ["basics", "signals", "events"]
  3# web = { width = 960, height = 540, responsive = true }
  4# ///
  5"""Signals.
  6
  7A signal is how a node announces that something happened without knowing who
  8cares. The emitter declares it, listeners connect to it, and emitting calls
  9every connected listener in turn.
 10
 11Declare a signal as a class attribute, typed with what it carries:
 12
 13```python
 14class Emitter(Node):
 15    ticked = Signal(int)
 16
 17    def tick(self):
 18        self.count += 1
 19        self.ticked.emit(self.count)   # fire the signal
 20```
 21
 22Each instance gets its own signal on first access, so listeners on one
 23`Emitter` never hear another.
 24
 25- `signal.connect(fn)` returns a `Connection`. Keep it to leave later:
 26  `conn.disconnect()` detaches the listener and `conn.connected` reports
 27  whether it is still attached.
 28- `signal.connect(fn, once=True)` spends itself on the first emit and then
 29  reports `connected` False.
 30- A bound method of a `Node` is held weakly. `destroy()` disconnects every
 31  connection made through the node's methods, and one whose node was garbage
 32  collected is skipped and dropped on the next emit, so a freed listener never
 33  crashes the emitter.
 34
 35The demo connects three listeners to one `Emitter`: a method on the root, a
 36once-listener and the method of a separate `Echo` node. Every delivery is
 37written to the log, and the status line reads each `Connection` back. Emit
 38with Space or the Emit button, detach and re-attach the root listener with D,
 39and free the `Echo` node with F; each button along the bottom presses the same
 40action as its key.
 41"""
 42
 43from collections import deque
 44
 45from simvx.core import Anchor, Button, CanvasLayer, HBoxContainer, Key, Label, Node, Signal, on_input
 46from simvx.graphics import App
 47
 48
 49class Emitter(Node):
 50    """Owns the signal and knows nothing about who listens."""
 51
 52    ticked = Signal(int)
 53    count = 0
 54
 55    def tick(self):
 56        self.count += 1
 57        self.ticked.emit(self.count)
 58
 59
 60class Echo(Node):
 61    """A separate listener node, whose bound method the signal holds weakly."""
 62
 63    def __init__(self, log, **kwargs):
 64        super().__init__(**kwargs)
 65        self.log = log
 66
 67    def on_tick(self, n: int):
 68        self.log(f"Echo.on_tick heard {n}")
 69
 70
 71class SignalsDemo(Node):
 72    input_context = {"emit": [Key.SPACE], "toggle": [Key.D], "free": [Key.F]}
 73
 74    def on_ready(self):
 75        hud = self.add_child(CanvasLayer())
 76        hud.add_child(Label("Signals", anchor=Anchor.TOP_LEFT, inset=16, font_size=28))
 77        self.status = hud.add_child(Label(anchor=Anchor.TOP_LEFT, inset=(16, 56), font_size=16))
 78        self.log_label = hud.add_child(Label(anchor=Anchor.TOP_LEFT, inset=(32, 96), font_size=16))
 79        self.lines = deque(maxlen=12)
 80        self.add_buttons(hud)
 81
 82        self.emitter = self.add_child(Emitter())
 83        self.root_conn = self.emitter.ticked.connect(self.on_tick)
 84        self.once_conn = self.emitter.ticked.connect(self.on_first_tick, once=True)
 85        self.echo = self.add_child(Echo(self.log))
 86        self.emitter.ticked.connect(self.echo.on_tick)
 87        self.log("connected: root listener, once listener, Echo.on_tick")
 88        self.refresh_status()
 89
 90    def on_tick(self, n: int):
 91        self.log(f"root listener heard {n}")
 92
 93    def on_first_tick(self, n: int):
 94        self.log(f"once listener heard {n} and disconnected itself")
 95
 96    @on_input("emit")
 97    def emit(self, _event):
 98        self.log(f"emit ticked({self.emitter.count + 1})")
 99        self.emitter.tick()
100        self.refresh_status()
101
102    @on_input("toggle")
103    def toggle_root_listener(self, _event):
104        if self.root_conn.connected:
105            self.root_conn.disconnect()
106            self.log("root listener detached with Connection.disconnect()")
107        else:
108            self.root_conn = self.emitter.ticked.connect(self.on_tick)
109            self.log("root listener reconnected")
110        self.refresh_status()
111
112    @on_input("free")
113    def free_echo(self, _event):
114        # Node has no public "was destroyed" state once the removal has run, so the reference is dropped by hand.
115        if self.echo is not None:
116            self.echo.destroy()
117            self.echo = None
118            self.log("Echo destroyed: its connection went with it, later emits skip it")
119        self.refresh_status()
120
121    def log(self, line: str):
122        self.lines.append(line)
123        self.log_label.text = "\n".join(self.lines)
124
125    def refresh_status(self):
126        # Read after the action has run: a once-listener is only disconnected once its callback has returned.
127        root = "connected" if self.root_conn.connected else "disconnected"
128        once = "connected" if self.once_conn.connected else "spent"
129        echo = "freed" if self.echo is None else "alive"
130        self.status.text = f"root listener: {root}    once listener: {once}    Echo: {echo}"
131
132    # Example help overlay: not part of the technique shown.
133    def add_buttons(self, hud):
134        bar = hud.add_child(HBoxContainer(anchor=Anchor.BOTTOM_LEFT, inset=12))
135        bar.add_child(Button("Emit", action="emit"))
136        bar.add_child(Button("Disconnect / reconnect root", action="toggle"))
137        bar.add_child(Button("Free Echo", action="free"))
138
139
140if __name__ == "__main__":
141    App(title="Signals", width=960, height=540).run(SignalsDemo())