"""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")