valknar
aad042650e
Covers all major Live API objects with get/set/listen/method handlers: song, track, return_track, clip, clip_slot, device (incl. rack chains and drum pads), scene, view, application, browser, groove. Zero external runtime dependencies — OSC encoded/decoded in osc_server.py. Wildcard * support for track/scene indices. Listener callbacks fire to matching /get/ addresses for bidirectional state sync. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""Handles /live/song/* OSC addresses."""
|
|
import logging
|
|
from typing import Optional
|
|
from .handler import AbletonOSCHandler
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Properties that are readable and writable
|
|
RW_PROPS = [
|
|
"tempo",
|
|
"time_signature_numerator",
|
|
"time_signature_denominator",
|
|
"loop",
|
|
"loop_start",
|
|
"loop_length",
|
|
"current_song_time",
|
|
"metronome",
|
|
"record_mode",
|
|
"arrangement_overdub",
|
|
"session_record",
|
|
"overdub",
|
|
"groove_amount",
|
|
"swing_amount",
|
|
"clip_trigger_quantization",
|
|
"midi_recording_quantization",
|
|
"punch_in",
|
|
"punch_out",
|
|
"exclusive_arm",
|
|
"exclusive_solo",
|
|
]
|
|
|
|
# Properties that are read-only
|
|
RO_PROPS = [
|
|
"is_playing",
|
|
"can_undo",
|
|
"can_redo",
|
|
"song_length",
|
|
"session_record_status",
|
|
]
|
|
|
|
|
|
class SongHandler(AbletonOSCHandler):
|
|
def init_api(self) -> None:
|
|
self.clear_listeners()
|
|
song = self.song
|
|
|
|
# --- get ---
|
|
for prop in RW_PROPS + RO_PROPS:
|
|
self._add(f"/live/song/get/{prop}", self._make_prop_getter(prop))
|
|
|
|
# --- set ---
|
|
for prop in RW_PROPS:
|
|
self._add(f"/live/song/set/{prop}", self._make_prop_setter(prop))
|
|
|
|
# --- listeners ---
|
|
for prop in RW_PROPS + RO_PROPS:
|
|
self._add(f"/live/song/start_listen/{prop}",
|
|
self._make_start_listen(prop))
|
|
self._add(f"/live/song/stop_listen/{prop}",
|
|
self._make_stop_listen(prop))
|
|
|
|
# --- aggregate queries ---
|
|
self._add("/live/song/get/num_tracks", self._get_num_tracks)
|
|
self._add("/live/song/get/num_scenes", self._get_num_scenes)
|
|
self._add("/live/song/get/num_return_tracks", self._get_num_return_tracks)
|
|
self._add("/live/song/get/track_names", self._get_track_names)
|
|
self._add("/live/song/get/scene_names", self._get_scene_names)
|
|
self._add("/live/song/get/return_track_names", self._get_return_track_names)
|
|
self._add("/live/song/get/cue_points", self._get_cue_points)
|
|
|
|
# --- method calls ---
|
|
self._add("/live/song/start_playing", lambda p: self._call("start_playing"))
|
|
self._add("/live/song/stop_playing", lambda p: self._call("stop_playing"))
|
|
self._add("/live/song/continue_playing", lambda p: self._call("continue_playing"))
|
|
self._add("/live/song/stop_all_clips", lambda p: self._call("stop_all_clips"))
|
|
self._add("/live/song/undo", lambda p: self._call("undo"))
|
|
self._add("/live/song/redo", lambda p: self._call("redo"))
|
|
self._add("/live/song/tap_tempo", lambda p: self._call("tap_tempo"))
|
|
self._add("/live/song/trigger_session_record",
|
|
lambda p: self._call("trigger_session_record"))
|
|
self._add("/live/song/re_enable_automation",
|
|
lambda p: self._call("re_enable_automation"))
|
|
self._add("/live/song/jump_by", self._jump_by)
|
|
self._add("/live/song/jump_to_next_cue", lambda p: self._call("jump_to_next_cue"))
|
|
self._add("/live/song/jump_to_prev_cue", lambda p: self._call("jump_to_prev_cue"))
|
|
self._add("/live/song/capture_midi", lambda p: self._call("capture_midi"))
|
|
|
|
# --- track/scene creation and deletion ---
|
|
self._add("/live/song/create_audio_track", self._create_audio_track)
|
|
self._add("/live/song/create_midi_track", self._create_midi_track)
|
|
self._add("/live/song/create_return_track",
|
|
lambda p: self._call("create_return_track"))
|
|
self._add("/live/song/create_scene", self._create_scene)
|
|
self._add("/live/song/delete_track", self._delete_track)
|
|
self._add("/live/song/delete_scene", self._delete_scene)
|
|
self._add("/live/song/delete_return_track", self._delete_return_track)
|
|
self._add("/live/song/duplicate_track", self._duplicate_track)
|
|
self._add("/live/song/duplicate_scene", self._duplicate_scene)
|
|
|
|
# --- beat listener (special) ---
|
|
self._add("/live/song/start_listen/beat", self._start_beat_listen)
|
|
self._add("/live/song/stop_listen/beat", self._stop_beat_listen)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Factories
|
|
# ------------------------------------------------------------------
|
|
|
|
def _make_prop_getter(self, prop: str):
|
|
def handler(params: tuple) -> Optional[tuple]:
|
|
val = self._get_prop(self.song, prop)
|
|
return (val,) if val is not None else None
|
|
return handler
|
|
|
|
def _make_prop_setter(self, prop: str):
|
|
def handler(params: tuple) -> None:
|
|
if params:
|
|
self._set_prop(self.song, prop, params[0])
|
|
return handler
|
|
|
|
def _make_start_listen(self, prop: str):
|
|
def handler(params: tuple) -> None:
|
|
self._register_listener(
|
|
f"song.{prop}", self.song, prop, f"/live/song/get/{prop}"
|
|
)
|
|
return handler
|
|
|
|
def _make_stop_listen(self, prop: str):
|
|
def handler(params: tuple) -> None:
|
|
self._remove_listener(f"song.{prop}")
|
|
return handler
|
|
|
|
# ------------------------------------------------------------------
|
|
# Aggregate queries
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get_num_tracks(self, params: tuple) -> tuple:
|
|
return (len(self.song.tracks),)
|
|
|
|
def _get_num_scenes(self, params: tuple) -> tuple:
|
|
return (len(self.song.scenes),)
|
|
|
|
def _get_num_return_tracks(self, params: tuple) -> tuple:
|
|
return (len(self.song.return_tracks),)
|
|
|
|
def _get_track_names(self, params: tuple) -> tuple:
|
|
start = int(params[0]) if len(params) > 0 else 0
|
|
end = int(params[1]) if len(params) > 1 else len(self.song.tracks)
|
|
tracks = list(self.song.tracks)[start:end]
|
|
return tuple(t.name for t in tracks)
|
|
|
|
def _get_scene_names(self, params: tuple) -> tuple:
|
|
start = int(params[0]) if len(params) > 0 else 0
|
|
end = int(params[1]) if len(params) > 1 else len(self.song.scenes)
|
|
scenes = list(self.song.scenes)[start:end]
|
|
return tuple(s.name for s in scenes)
|
|
|
|
def _get_return_track_names(self, params: tuple) -> tuple:
|
|
return tuple(t.name for t in self.song.return_tracks)
|
|
|
|
def _get_cue_points(self, params: tuple) -> tuple:
|
|
result = []
|
|
for cp in self.song.cue_points:
|
|
result.append(cp.name)
|
|
result.append(float(cp.time))
|
|
return tuple(result)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Methods
|
|
# ------------------------------------------------------------------
|
|
|
|
def _call(self, method: str) -> None:
|
|
try:
|
|
getattr(self.song, method)()
|
|
except Exception as e:
|
|
logger.warning("song.%s(): %s", method, e)
|
|
return None
|
|
|
|
def _jump_by(self, params: tuple) -> None:
|
|
if params:
|
|
try:
|
|
self.song.jump_by(float(params[0]))
|
|
except Exception as e:
|
|
logger.warning("song.jump_by: %s", e)
|
|
return None
|
|
|
|
# --- track/scene CRUD ---
|
|
|
|
def _create_audio_track(self, params: tuple) -> tuple:
|
|
idx = int(params[0]) if params else -1
|
|
try:
|
|
self.song.create_audio_track(idx)
|
|
except Exception as e:
|
|
logger.warning("create_audio_track: %s", e)
|
|
return (len(self.song.tracks),)
|
|
|
|
def _create_midi_track(self, params: tuple) -> tuple:
|
|
idx = int(params[0]) if params else -1
|
|
try:
|
|
self.song.create_midi_track(idx)
|
|
except Exception as e:
|
|
logger.warning("create_midi_track: %s", e)
|
|
return (len(self.song.tracks),)
|
|
|
|
def _create_scene(self, params: tuple) -> tuple:
|
|
idx = int(params[0]) if params else -1
|
|
try:
|
|
self.song.create_scene(idx)
|
|
except Exception as e:
|
|
logger.warning("create_scene: %s", e)
|
|
return (len(self.song.scenes),)
|
|
|
|
def _delete_track(self, params: tuple) -> None:
|
|
if params:
|
|
try:
|
|
self.song.delete_track(int(params[0]))
|
|
except Exception as e:
|
|
logger.warning("delete_track: %s", e)
|
|
return None
|
|
|
|
def _delete_scene(self, params: tuple) -> None:
|
|
if params:
|
|
try:
|
|
self.song.delete_scene(int(params[0]))
|
|
except Exception as e:
|
|
logger.warning("delete_scene: %s", e)
|
|
return None
|
|
|
|
def _delete_return_track(self, params: tuple) -> None:
|
|
if params:
|
|
try:
|
|
self.song.delete_return_track(int(params[0]))
|
|
except Exception as e:
|
|
logger.warning("delete_return_track: %s", e)
|
|
return None
|
|
|
|
def _duplicate_track(self, params: tuple) -> None:
|
|
if params:
|
|
try:
|
|
self.song.duplicate_track(int(params[0]))
|
|
except Exception as e:
|
|
logger.warning("duplicate_track: %s", e)
|
|
return None
|
|
|
|
def _duplicate_scene(self, params: tuple) -> None:
|
|
if params:
|
|
try:
|
|
self.song.duplicate_scene(int(params[0]))
|
|
except Exception as e:
|
|
logger.warning("duplicate_scene: %s", e)
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Beat listener
|
|
# ------------------------------------------------------------------
|
|
|
|
def _start_beat_listen(self, params: tuple) -> None:
|
|
self._remove_listener("song.beat")
|
|
self._beat_count = -1
|
|
|
|
def on_beat():
|
|
beat = int(self.song.get_current_beats_song_time().beats)
|
|
if beat != self._beat_count:
|
|
self._beat_count = beat
|
|
self._send("/live/song/get/beat", (beat,))
|
|
|
|
try:
|
|
self.song.add_current_song_time_listener(on_beat)
|
|
self._listener_store["song.beat"] = (
|
|
self.song.remove_current_song_time_listener, on_beat
|
|
)
|
|
except Exception as e:
|
|
logger.warning("beat listener: %s", e)
|
|
|
|
def _stop_beat_listen(self, params: tuple) -> None:
|
|
self._remove_listener("song.beat")
|