How to follow a whole reference graph¶

Boxes can point at boxes that point at further boxes: a stage at a motor, the motor at an encoder. When a box's reference fields lead on like this, they form a graph. This guide shows how one callback can hear about a change anywhere in it.

Before you start¶

import queue
from typing import Any

from sharedbox import SharedBox


class Encoder(SharedBox, name="example-encoder"):
    count: int = 0


class Motor(SharedBox, name="example-motor"):
    position: int = 0
    encoder: Encoder | None = None


class Stage(SharedBox, name="example-stage"):
    motor: Motor | None = None

Referring to another box follows a single reference field; here you follow all of them at once.

1. Follow every reference¶

The script creates an encoder, a motor that points at it and a stage that points at the motor. Call follow on the stage's events without naming a field, then connect to the nested signal:

changes: queue.Queue[tuple[tuple[str, ...], Any]] = queue.Queue()
stage.events.follow()
stage.events.nested.connect(lambda path, new, old: changes.put((path, new)))
encoder.count = 3
print(changes.get(timeout=5))  # (('motor', 'encoder', 'count'), 3)
motor.position = 7
print(changes.get(timeout=5))  # (('motor', 'position'), 7)

Your callback gets (path, new, old), where path lists the field names on the way from the stage to the field that changed. When any process points a reference field at a different box, the following moves to the new box by itself.

The callbacks run on background threads, one for each box you follow. Hand the values over to your own thread, as the queue does here, or connect with thread="main", as the docstring of events describes.

2. Follow one path instead¶

If you only care about one box deep in the graph, call follow with a field at each level instead. Each call gives you the signals of the class that field points at:

stage.events.unfollow()
counts: queue.Queue[int] = queue.Queue()
encoder_events = stage.events.follow("motor").follow("encoder")
encoder_events.count.connect(lambda new, old: counts.put(new))
encoder.count = 4
print(counts.get(timeout=5))  # 4

3. Stop following¶

unfollow with a field stops what follow with that field started, and without a field it stops everything, as the script does before step 2. Closing the box stops all following too.

Following isn't free: each box you follow takes one waiter slot and one thread. Limits says what that adds up to for a large graph.

"""The script of the guide "How to follow a whole reference graph"."""

import queue
from typing import Any

from sharedbox import SharedBox


class Encoder(SharedBox, name="example-encoder"):
    count: int = 0


class Motor(SharedBox, name="example-motor"):
    position: int = 0
    encoder: Encoder | None = None


class Stage(SharedBox, name="example-stage"):
    motor: Motor | None = None



with Encoder() as encoder, Motor(0, encoder) as motor, Stage(motor) as stage:
    changes: queue.Queue[tuple[tuple[str, ...], Any]] = queue.Queue()
    stage.events.follow()
    stage.events.nested.connect(lambda path, new, old: changes.put((path, new)))
    encoder.count = 3
    print(changes.get(timeout=5))  # (('motor', 'encoder', 'count'), 3)
    motor.position = 7
    print(changes.get(timeout=5))  # (('motor', 'position'), 7)

    stage.events.unfollow()
    counts: queue.Queue[int] = queue.Queue()
    encoder_events = stage.events.follow("motor").follow("encoder")
    encoder_events.count.connect(lambda new, old: counts.put(new))
    encoder.count = 4
    print(counts.get(timeout=5))  # 4
Stage.unlink()
Motor.unlink()
Encoder.unlink()