21  10. Mixing with a human in the room

Levels, pans and sends across eight tracks, a glue compressor and a limiter on the main track — and, unlike every other lesson, a loop that waits for you. Observers are the one capability here that nothing else in the field has: they report what the user does with their hands in Live’s GUI, which is what makes this a conversation rather than a script.

Prerequisites: lesson 9 executed — eight tracks, 155 arrangement clips, seven locators.

The ask: “Watch the mixer. I’m going to set the drums and bass by ear; then balance everything else around them, and tell me every time I change something.”

from sideman import Live

live = Live()
TRACKS = {i: live.get(f"live_set tracks {i}", "name")["value"] for i in range(8)}
TRACKS
{0: 'Drums',
 1: 'Sub',
 2: 'Mid',
 3: 'Pad',
 4: 'Lead',
 5: 'Arp',
 6: 'Noise',
 7: 'Vocal'}

21.1 Availability is per instance

Before watching anything, ask what each track actually has. The main track has no mute and no arm; a MIDI track has no input meters. describe reports these under unavailable, and an observer on a property an object does not have is a failed round trip.

{name: sorted(live.describe(f"live_set tracks {i}",
                            include_values=False)["unavailable"])
 for i, name in ((0, "Drums (MIDI)"), (7, "Vocal (audio)"))} \
 | {"Main": sorted(live.describe("live_set master_track",
                                 include_values=False)["unavailable"])}
{'Drums (MIDI)': ['fold_state', 'input_meter_left', 'input_meter_right'],
 'Vocal (audio)': ['fold_state'],
 'Main': ['arm',
  'arrangement_clips',
  'fired_slot_index',
  'fold_state',
  'implicit_arm',
  'mute',
  'muted_via_solo',
  'playing_slot_index',
  'solo']}

21.2 A baseline, before any observer

The observer registry lives inside Live and is shared across clients — the MCP server, a shell session and this notebook all see the same buffer. So take the current latest_seq first and poll with since=: it is cheaper than re-reading, and it means you never claim someone else’s events as your own.

For the same reason, unobserve_all is a blunt instrument. It removes observers you did not create.

baseline = live.poll_events()["latest_seq"]
baseline, live.observers()["active_listeners"]
(0, 0)

21.3 Watch the faders

for i in range(8):
    live.observe(f"live_set tracks {i} mixer_device volume", "value")
live.observers()["active_listeners"]
8

Now the hand on the fader. In your session this is you: drag the Drums and Sub faders in Live until the kick and the sub sit where you want them. The outputs frozen in this notebook were produced with nobody at the keyboard, so a second client stands in — a separate Live() connection, which is exactly what a GUI drag looks like from here: a change this notebook did not make.

hands = Live()
hands.set("live_set tracks 0 mixer_device volume", "value", 0.88)   # Drums
hands.set("live_set tracks 1 mixer_device volume", "value", 0.74)   # Sub
{'path': 'live_set tracks 1 mixer_device volume',
 'property': 'value',
 'value': 0.7400000095367432}
ev = live.poll_events(since=baseline)
ev["count"], ev["dropped_events"], [(TRACKS[int(e["path"].split()[2])],
                                     round(e["value"], 3)) for e in ev["events"]]
(2, 0, [('Drums', 0.88), ('Sub', 0.74)])
Figure 21.1: The path those two events took: a second client moves a fader, Live notifies one listener per property, the queue buffers what comes back, and the poll hands it over with a sequence number to resume from.

Two events, and they name the tracks the user touched. That is the whole interaction loop: they move something, you read what moved, you respond to it.

21.4 Balance around what they set

Drums at 0.88 and Sub at 0.74 are now fixed points. Everything else is set relative to them in one round trip — volumes, two pans to open up the middle, and sends for the vocal and the arp.

mix = {2: 0.62, 3: 0.55, 4: 0.60, 5: 0.48, 6: 0.42, 7: 0.58}
live.set_batch(
    [{"path": f"live_set tracks {i} mixer_device volume",
      "property": "value", "value": v} for i, v in mix.items()]
    + [{"path": "live_set tracks 5 mixer_device panning", "property": "value",
        "value": -0.25},                                   # Arp left
       {"path": "live_set tracks 4 mixer_device panning", "property": "value",
        "value": 0.15},                                    # Lead right
       {"path": "live_set tracks 7 mixer_device sends 0", "property": "value",
        "value": 0.35},                                    # Vocal -> reverb
       {"path": "live_set tracks 5 mixer_device sends 0", "property": "value",
        "value": 0.20}]                                    # Arp -> reverb
)["applied"]
10
[(TRACKS[i], round(live.get(f"live_set tracks {i} mixer_device volume",
                            "value")["value"], 2)) for i in range(8)]
[('Drums', 0.88),
 ('Sub', 0.74),
 ('Mid', 0.62),
 ('Pad', 0.55),
 ('Lead', 0.6),
 ('Arp', 0.48),
 ('Noise', 0.42),
 ('Vocal', 0.58)]
Figure 21.2: The mixer those numbers describe: eight track strips, then the two returns and Main, with the meters moving on whatever is sounding.

Those writes fired the observers too — an observer reports a change, not a source of change. Anything reacting to user input has to compare against what it just did, or it will answer itself.

after_write = live.poll_events(since=ev["latest_seq"])
after_write["count"]
6

21.5 Meters are a different kind of signal

output_meter_level is observable on every track and it changes while the transport runs. Watch all eight, play four seconds of the drop, and count what comes back — a fader move produced two events, so the question is how much noise a meter puts in the same buffer.

import time
from collections import Counter

for i in range(8):
    live.observe(f"live_set tracks {i}", "output_meter_level")

mark = live.poll_events()["latest_seq"]
live.set("live_set", "current_song_time", 192.0)     # Drop 1
live.call("live_set", "start_playing")
playing = live.get("live_set", "is_playing")["value"]
time.sleep(4)
live.call("live_set", "stop_playing")

ev = live.poll_events(since=mark, limit=500)
playing, ev["latest_seq"] - mark, ev["dropped_events"],     Counter(e["property"] for e in ev["events"])
(True, 92, 0, Counter({'output_meter_level': 92}))

Those four seconds in Live: Drop 1 playing, every meter moving, and a fadermoved by the second client while the notebook watches.

Measured, not assumed. Four seconds of eight meters is a couple of hundred events — Live coalesces meter updates rather than firing per audio frame, so this is noisy but not a flood. The buffer holds 2000 and drops the oldest first, so roughly a minute of playback with these eight left on would begin losing events, and dropped_events would say so. A nonzero dropped_events while you are watching for gestures means a fader move was thrown away, which is why the meters come off before the conversation continues.

for i in range(8):
    live.unobserve(f"live_set tracks {i}", "output_meter_level")
live.observers()["active_listeners"]
8

21.6 The main track

Same as lesson 6: no track index, so select it first, then load. Glue Compressor to hold the mix together, Limiter to stop it clipping — in that order, because devices load at the end of the chain and the limiter must be last.

live.set("live_set view", "selected_track", {"__path__": "live_set master_track"})
live.browser_load("audio_effects/Glue Compressor")
live.browser_load("audio_effects/Limiter")
[live.get(f"live_set master_track devices {i}", "name")["value"]
 for i in range(live.count("live_set master_track", "devices")["count"])]
['Pro-Q 3', 'Glue Compressor', 'Limiter']

Pro-Q 3 from lesson 6 is still first in the chain, which is where an EQ belongs.

21.7 The guard

The Set is not in version control and nothing here has been saved. Every function whose name says it destroys something refuses to run unless you pass confirm=True — the refusal is a PermissionError from inside Live, so it is distinguishable from a genuine failure.

try:
    live.call("live_set", "delete_track", [7])
except Exception as e:
    print(type(e).__name__, e.type, "|", e)
LiveError PermissionError | PermissionError: 'delete_track' is destructive; re-send with confirm=true to proceed

21.8 Clean up the observers

A listener Live still holds after you lose track of it fires forever. leaked must be 0.

live.observers()
{'active_listeners': 8,
 'buffered_events': 188,
 'dropped_events': 0,
 'max_events': 2000,
 'listeners': [{'path': 'live_set tracks 0 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 1 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 2 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 3 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 4 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 5 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 6 mixer_device volume',
   'property': 'value',
   'object_valid': True},
  {'path': 'live_set tracks 7 mixer_device volume',
   'property': 'value',
   'object_valid': True}]}
live.unobserve_all()
{'attempted': 8, 'outcomes': {'removed': 8}, 'failures': [], 'leaked': 0}

21.9 The Set, finished

Eight tracks, five and a half minutes, and nothing saved — saving is yours, and ⌘S is the one thing in this book that has to be done by hand on purpose.

rows = []
for i in range(8):
    rows.append((TRACKS[i],
                 round(live.get(f"live_set tracks {i} mixer_device volume",
                                "value")["value"], 2),
                 round(live.get(f"live_set tracks {i} mixer_device panning",
                                "value")["value"], 2),
                 live.arrangement_list(f"live_set tracks {i}")["count"]))
rows
[('Drums', 0.88, 0.0, 38),
 ('Sub', 0.74, 0.0, 32),
 ('Mid', 0.62, 0.0, 16),
 ('Pad', 0.55, 0.0, 28),
 ('Lead', 0.6, 0.15, 20),
 ('Arp', 0.48, -0.25, 16),
 ('Noise', 0.42, 0.0, 2),
 ('Vocal', 0.58, 0.0, 3)]
The lesson inside the lesson

Availability is per instance, not per type: the main track has no mute, a MIDI track has no input meters, and describe says so under unavailable rather than making you find out from a traceback.

Observers are a pull with a ring buffer behind them, and they catch what the user does in the GUI — which nothing else in this field reports. The registry is shared across every client talking to this Live, so take a latest_seq baseline, poll with since=, and think twice before unobserve_all removes a listener another session is depending on.

Destructive calls refuse without confirm=true. Saving is not automated at all: the Set you just built exists only in memory until you press ⌘S.

21.10 Checkpoint

obs = live.observers()
assert obs["active_listeners"] == 0, obs
assert live.count("live_set master_track", "devices")["count"] == 3
assert live.count("live_set", "tracks")["count"] == 8
assert live.count("live_set", "cue_points")["count"] == 7
assert [round(live.get(f"live_set tracks {i} mixer_device volume",
                       "value")["value"], 2) for i in range(8)] == \
       [0.88, 0.74, 0.62, 0.55, 0.60, 0.48, 0.42, 0.58]
assert live.get("live_set", "tempo")["value"] == 122.0
assert live.get("live_set", "is_playing")["value"] is False
assert live.get("live_set", "file_path")["value"] == "", "now you can save it"
print("lesson 10 ok — mix set, 0 observers leaked, 3 devices on the main track")
lesson 10 ok — mix set, 0 observers leaked, 3 devices on the main track