Skip to content

Referring to another box

A box can point at another box, the way an attribute of one Python object can hold another object. In this tutorial you make a Stage that points at a Motor, connect a callback that hears about changes inside whichever motor the stage points at, and then switch the stage to a second motor.

Before you start

What you need

The script motor.py from Reacting to changes.

1. Define the stage

Add this class below react:

class Stage(SharedBox):
    motor: Motor | None = None

The motor field doesn't copy a motor's values into the stage. It is a reference field: it stores which Motor the stage points at, and the motor keeps its values in its own shared memory. The | None lets the field be empty, which is also its default here. Reference fields explains what exactly the field stores.

2. Follow the motor

Add this function below Stage:

def follow() -> None:
    with (
        Motor.create("tutorial-x") as x,
        Motor.create("tutorial-y") as y,
        Stage.create("tutorial-stage", x) as stage,
    ):
        seen: queue.Queue[int] = queue.Queue()
        stage.events.follow("motor").position.connect(lambda new, old: seen.put(new))
        x.position = 1
        print("motor position", seen.get(timeout=5))
        stage.motor = y
        # forwarding moves to the new box on a background thread, shortly after
        time.sleep(0.5)
        y.position = 7
        print("motor position", seen.get(timeout=5))
    Stage.unlink("tutorial-stage")
    Motor.unlink("tutorial-x")
    Motor.unlink("tutorial-y")

The function creates two motors, x and y, and a stage that points at x. events.follow("motor") gives you signals like a motor's own events, and each one fires when the motor that stage.motor currently points at changes. The callback runs on a background thread, and printing from two threads at once can mix their lines, so the callback only puts each new position on a queue. The main thread takes the positions from there and prints them.

Assigning y to stage.motor makes the callback follow y instead of x.

The switch to a new box takes a moment

After you assign y to stage.motor, a background thread moves the following from x to y, shortly after the assignment returns. A write to y made before that is missed by the callback. That is why the script waits half a second before it writes to y; in your own code, leave the same gap or don't rely on that first write.

3. Run it

In the if block at the end of the file, add the highlighted line:

if __name__ == "__main__":
    share()
    react()
    follow()

Run the script:

uv run motor.py

The last two lines are new: the callback heard the write to x, then the write to y:

position set by the other process: 10
position 0 -> 20
watched 20
motor position 1
motor position 7

What you built

You have a stage that points at a motor, and a callback that keeps hearing about the stage's motor even after you point the stage at a different one.

Next steps

The whole script 
"""The script the sharedbox tutorial builds, one page at a time."""

import multiprocessing as mp
import queue
import time
from typing import Annotated

from sharedbox import Capacity, SharedBox



class Motor(SharedBox):
    position: int = 0
    enabled: bool = False
    label: Annotated[str, Capacity(32)] = ""




def move(position: int) -> None:
    motor = Motor.attach("tutorial-motor")
    motor.position = position
    motor.close()




def share() -> None:
    with Motor.create("tutorial-motor", label="x-axis") as motor:
        other = mp.Process(target=move, args=(10,))
        other.start()
        other.join()
        print("position set by the other process:", motor.position)
    Motor.unlink("tutorial-motor")




def react() -> None:
    with Motor.create("tutorial-motor", label="x-axis") as motor:
        motor.events.position.connect(
            lambda new, old: print(f"position {old} -> {new}")
        )
        positions = motor.watch("position")
        other = mp.Process(target=move, args=(20,))
        other.start()
        watched = next(iter(positions))
        other.join()
    print("watched", watched)
    Motor.unlink("tutorial-motor")




class Stage(SharedBox):
    motor: Motor | None = None




def follow() -> None:
    with (
        Motor.create("tutorial-x") as x,
        Motor.create("tutorial-y") as y,
        Stage.create("tutorial-stage", x) as stage,
    ):
        seen: queue.Queue[int] = queue.Queue()
        stage.events.follow("motor").position.connect(lambda new, old: seen.put(new))
        x.position = 1
        print("motor position", seen.get(timeout=5))
        stage.motor = y
        # forwarding moves to the new box on a background thread, shortly after
        time.sleep(0.5)
        y.position = 7
        print("motor position", seen.get(timeout=5))
    Stage.unlink("tutorial-stage")
    Motor.unlink("tutorial-x")
    Motor.unlink("tutorial-y")




if __name__ == "__main__":
    share()
    react()
    follow()