Clip Module: Dictionary Registry + Notes/Warp Markers/Views/All Properties

abletonosc/__init__.py

@@ -6,7 +6,7 @@
 from .osc_server import OSCServer
 from .application import ApplicationHandler
 from .song import SongHandler
-from .clip import ClipHandler
+from .clip import ClipHandler, ClipViewHandler
 from .clip_slot import ClipSlotHandler
 from .track import TrackHandler
 from .device import DeviceHandler

abletonosc/clip_slot.py

@@ -7,61 +7,109 @@ def __init__(self, manager):
         self.class_identifier = "clip_slot"
 
     def init_api(self):
-        def create_clip_slot_callback(func, *args, pass_clip_index=False):
+        def create_clip_slot_callback(func, *args, pass_clip_index=False, **kwargs):
             def clip_slot_callback(params: Tuple[Any]):
                 track_index, clip_index = int(params[0]), int(params[1])
                 track = self.song.tracks[track_index]
                 clip_slot = track.clip_slots[clip_index]
 
                 if pass_clip_index:
-                    rv = func(clip_slot, *args, tuple(params[0:]))
+                    rv = func(clip_slot, *args, tuple(params[0:]), **kwargs)
                 else:
-                    rv = func(clip_slot, *args, tuple(params[2:]))
+                    rv = func(clip_slot, *args, tuple(params[2:]), **kwargs)
 
-                self.logger.info(track_index, clip_index, rv)
+                self.logger.info("clip_slot %s,%s -> %s", track_index, clip_index, rv)
                 if rv is not None:
                     return (track_index, clip_index, *rv)
 
             return clip_slot_callback
 
-        methods = [
-            "fire",
-            "stop",
-            "create_clip",
-            "delete_clip"
-        ]
-        properties_r = [
-            "has_clip",
-            "controls_other_clips",
-            "is_group_slot",
-            "is_playing",
-            "is_triggered",
-            "playing_status",
-            "will_record_on_start",
-        ]
-        properties_rw = [
-            "has_stop_button"
-        ]
+        methods = {
+            "fire":              {"alias": 0, "caller": 1},
+            "stop":              {"alias": 0, "caller": 1},
+            "create_clip":       {"alias": 0, "caller": 1},   # Back-compat
+            "create/midi_clip":  {"alias": 1, "caller": "create_clip"},
+            "create/audio_clip": {"alias": 1, "caller": "create_audio_clip"},
+            "delete_clip":       {"alias": 0, "caller": 1},   # Back-compat
+            "delete/clip":       {"alias": 1, "caller": "delete_clip"},
+            "duplicate_to":      {"alias": 0, "caller": "clip_slot_duplicate_to"},
+            "set/fire_button":   {"alias": 1, "caller": "set_fire_button_state"},
+        }
 
-        for method in methods:
-            self.osc_server.add_handler("/live/clip_slot/%s" % method,
-                                        create_clip_slot_callback(self._call_method, method))
+        properties = {
+            # TODO: returns objects, needs serialization
+            # "canonical_parent":      {"get": 0, "set": 0, "listen": 0},
+            # "clip":                  {"get": 0, "set": 0, "listen": 0},
 
-        for prop in properties_r + properties_rw:
-            self.osc_server.add_handler("/live/clip_slot/get/%s" % prop,
-                                        create_clip_slot_callback(self._get_property, prop))
-            self.osc_server.add_handler("/live/clip_slot/start_listen/%s" % prop,
-                                        create_clip_slot_callback(self._start_listen, prop, pass_clip_index=True))
-            self.osc_server.add_handler("/live/clip_slot/stop_listen/%s" % prop,
-                                        create_clip_slot_callback(self._stop_listen, prop, pass_clip_index=True))
-        for prop in properties_rw:
-            self.osc_server.add_handler("/live/clip_slot/set/%s" % prop,
-                                        create_clip_slot_callback(self._set_property, prop))
+            # All clip slots
+            "has_clip":              {"get": 1, "set": 0, "listen": 1},
+            "has_stop_button":       {"get": 1, "set": 1, "listen": 1},
+            "is_group_slot":         {"get": 1, "set": 0, "listen": 0},
+            "is_playing":            {"get": 1, "set": 0, "listen": 0},
+            "is_recording":          {"get": 1, "set": 0, "listen": 0},
+            "is_triggered":          {"get": 1, "set": 0, "listen": 1},
+            "will_record_on_start":  {"get": 1, "set": 0, "listen": 0},
+
+            # Group Track slots only
+            "color":                 {"get": 1, "set": 0, "listen": 1},
+            "color_index":           {"get": 1, "set": 0, "listen": 1},
+            "controls_other_clips":  {"get": 1, "set": 0, "listen": 1},
+            "playing_status":        {"get": 1, "set": 0, "listen": 1},
 
-        def duplicate_clip_slot(clip_slot, args):
+
+        }
+
+        def clip_slot_duplicate_to(clip_slot, args):
             target_track_index, target_clip_index = tuple(args)
             track = self.song.tracks[target_track_index]
             target_clip_slot = track.clip_slots[target_clip_index]
             clip_slot.duplicate_clip_to(target_clip_slot)
 
-        self.osc_server.add_handler("/live/clip_slot/duplicate_clip_to", create_clip_slot_callback(duplicate_clip_slot))
+        # Add Handlers
+        local_funcs = locals()
+        for method, spec in methods.items():
+            alias = spec.get("alias")
+            caller = spec.get("caller")
+            if not caller:
+                continue
+            if not alias and isinstance(caller, str):
+                caller = local_funcs[caller]
+                self.osc_server.add_handler("/live/clip_slot/%s" % method,
+                                            create_clip_slot_callback(caller))
+            else:
+                if not alias:
+                    caller = method
+                self.osc_server.add_handler("/live/clip_slot/%s" % method,
+                                            create_clip_slot_callback(self._call_method, caller))
+
+        for prop, spec in properties.items():
+            getter_func = spec.get("get")
+            if isinstance(getter_func, str):
+                getter = local_funcs[getter_func]
+                self.osc_server.add_handler("/live/clip_slot/get/%s" % prop,
+                                            create_clip_slot_callback(getter))
+            elif getter_func:
+                self.osc_server.add_handler("/live/clip_slot/get/%s" % prop,
+                                            create_clip_slot_callback(self._get_property, prop))
+
+            setter_func = spec.get("set")
+            if isinstance(setter_func, str):
+                setter = local_funcs[setter_func]
+                self.osc_server.add_handler("/live/clip_slot/set/%s" % prop,
+                                            create_clip_slot_callback(setter))
+            elif setter_func:
+                self.osc_server.add_handler("/live/clip_slot/set/%s" % prop,
+                                            create_clip_slot_callback(self._set_property, prop))
+
+            observable = spec.get("listen")
+            if isinstance(observable, str):
+                getter = "bang" if observable == "bang" else local_funcs[observable]
+                self.osc_server.add_handler("/live/clip_slot/start_listen/%s" % prop,
+                                            create_clip_slot_callback(self._start_listen, prop, pass_clip_index=True, getter=getter))
+                self.osc_server.add_handler("/live/clip_slot/stop_listen/%s" % prop,
+                                            create_clip_slot_callback(self._stop_listen, prop, pass_clip_index=True))
+            elif observable:
+                self.osc_server.add_handler("/live/clip_slot/start_listen/%s" % prop,
+                                            create_clip_slot_callback(self._start_listen, prop, pass_clip_index=True))
+                self.osc_server.add_handler("/live/clip_slot/stop_listen/%s" % prop,
+                                            create_clip_slot_callback(self._stop_listen, prop, pass_clip_index=True))

abletonosc/handler.py

@@ -61,8 +61,13 @@ def _start_listen(self, target, prop, params: Optional[Tuple] = (), getter = Non
         def property_changed_callback():
             if getter is None:
                 value = getattr(target, prop)
+            elif getter == "bang":
+                value = 1
             else:
-                value = getter(params)
+                try:
+                    value = getter(target, params)
+                except TypeError:
+                    value = getter(params)
             if type(value) is not tuple:
                 value = (value,)
             self.logger.info("Property %s changed of %s %s: %s" % (prop, self.class_identifier, str(params), value))
@@ -80,9 +85,10 @@ def property_changed_callback():
         self.listener_functions[listener_key] = property_changed_callback
         self.listener_objects[listener_key] = target
         #--------------------------------------------------------------------------------
-        # Immediately send the current value
+        # Immediately send the current value (skip for bang-only listeners)
         #--------------------------------------------------------------------------------
-        property_changed_callback()
+        if getter != "bang":
+            property_changed_callback()
 
     def _stop_listen(self, target, prop, params: Optional[Tuple[Any]] = ()) -> None:
         listener_key = (prop, tuple(params))

manager.py

@@ -94,6 +94,7 @@ def show_message_callback(params):
                 abletonosc.SongHandler(self),
                 abletonosc.ApplicationHandler(self),
                 abletonosc.ClipHandler(self),
+                abletonosc.ClipViewHandler(self),
                 abletonosc.ClipSlotHandler(self),
                 abletonosc.TrackHandler(self),
                 abletonosc.DeviceHandler(self),
@@ -155,4 +156,4 @@ def build_midi_map(self, midi_map_handle):
         for channel, cc in self.midi_mappings.keys():
             parameter = self.midi_mappings[(channel, cc)]
             Live.MidiMap.map_midi_cc(midi_map_handle, parameter, channel, cc, Live.MidiMap.MapMode.absolute, 1)
-            logger.debug("Mapped CC %d on channel %d to parameter %s" % (cc, channel, parameter.name))
+            logger.debug("Mapped CC %d on channel %d to parameter %s" % (cc, channel, parameter.name))

tests/__init__.py

@@ -8,6 +8,10 @@
 import sys
 sys.path.append(".")
 
+from pathlib import Path
+import wave
+import struct
+
 from ..client import AbletonOSCClient, TICK_DURATION
 
 # Live tick is 100ms. Wait for this long plus a short additional buffer.
@@ -19,12 +23,45 @@ def client() -> AbletonOSCClient:
     yield client
     client.stop()
 
+@pytest.fixture(scope="function")
+def silent_audio_file() -> Path:
+    """
+    Create a silent WAV file in the tests directory for audio-clip tests.
+    """
+    path = Path(__file__).resolve().parent / "silent_8s.wav"
+    if not path.exists():
+        duration_s = 8.0
+        sample_rate = 48000
+        channels = 1
+        sample_width = 2  # 16-bit
+        total_frames = int(duration_s * sample_rate)
+        chunk_frames = 4096
+        silence_chunk = struct.pack("<h", 0) * chunk_frames
+        with wave.open(str(path), "wb") as wf:
+            wf.setnchannels(channels)
+            wf.setsampwidth(sample_width)
+            wf.setframerate(sample_rate)
+            frames_remaining = total_frames
+            while frames_remaining > 0:
+                frames = min(chunk_frames, frames_remaining)
+                wf.writeframes(silence_chunk[: frames * sample_width])
+                frames_remaining -= frames
+    yield path
+    remove_audio_file(path)
+
 def wait_one_tick():
     """
     Sleep for one Ableton Live tick (100ms).
     """
     time.sleep(TICK_DURATION)
 
+def remove_audio_file(path: Path) -> None:
+    for target in (path, Path(str(path) + ".asd")):
+        try:
+            target.unlink()
+        except FileNotFoundError:
+            pass
+
 c = AbletonOSCClient()
 c.send_message("/live/api/reload")
 c.stop()