Merge branch 'master' into feature/listen_scene

steeltrack · Nov 19, 2023 · 58dce85 · 58dce85
2 parents bf323af + c908fc9
commit 58dce85
Show file tree

Hide file tree

Showing 11 changed files with 444 additions and 230 deletions.
diff --git a/README.md b/README.md
diff --git a/abletonosc/application.py b/abletonosc/application.py
@@ -12,3 +12,9 @@ def get_version(_) -> Tuple:
             return application.get_major_version(), application.get_minor_version()
         self.osc_server.add_handler("/live/application/get/version", get_version)
         self.osc_server.send("/live/startup")
+
+        def get_average_process_usage(_) -> Tuple:
+            application = Live.Application.get_application()
+            return application.average_process_usage,
+        self.osc_server.add_handler("/live/application/get/average_process_usage", get_average_process_usage)
+        self.osc_server.send("/live/appli
8000
cation/get/average_process_usage")
diff --git a/abletonosc/clip.py b/abletonosc/clip.py
@@ -3,10 +3,32 @@
 from .handler import AbletonOSCHandler
 import Live
 
+def note_name_to_midi(name):
+    """ Maps a MIDI note name (D3, C#6) to a value.
+    Assumes that middle C is C4. """
+    note_names = [["C"],
+                  ["C#", "Db"],
+                  ["D"],
+                  ["D#", "Eb"],
+                  ["E"],
+                  ["F"],
+                  ["F#", "Gb"],
+                  ["G"],
+                  ["G#", "Ab"],
+                  ["A"],
+                  ["A#", "Bb"],
+                  ["B"]]
+
+    for index, names in enumerate(note_names):
+        if name in names:
+            return index
+    return None
+
 class ClipHandler(AbletonOSCHandler):
     def __init__(self, manager):
         super().__init__(manager)
         self.class_identifier = "clip"
+        self._clip_notes_cache = []
 
     def init_api(self):
         def create_clip_callback(func, *args, pass_clip_index=False):
@@ -66,7 +88,9 @@ def clip_callback(params: Tuple[Any]) -> Tuple:
             "looping",
             "loop_start",
             "loop_end",
-            "warping"
+            "warping",
+            "start_marker",
+            "end_marker",
         ]
 
         for method in methods:
@@ -85,8 +109,17 @@ def clip_callback(params: Tuple[Any]) -> Tuple:
                                         create_clip_callback(self._set_property, prop))
 
         def clip_get_notes(clip, params: Tuple[Any] = ()):
-            notes = clip.get_notes(0, 0, clip.length, 127)
-            return tuple(item for sublist in notes for item in sublist)
+            if len(params) == 4:
+                pitch_start, pitch_span, time_start, time_span = params
+            elif len(params) == 0:
+                pitch_start, pitch_span, time_start, time_span = 0, 127, -8192, 16384
+            else:
+                raise ValueError("Invalid number of arguments for /clip/get/notes. Either 0 or 4 arguments must be passed.")
+            notes = clip.get_notes_extended(pitch_start, pitch_span, time_start, time_span)
+            all_note_attributes = []
+            for note in notes:
+                all_note_attributes += [note.pitch, note.start_time, note.duration, note.velocity, note.mute]
+            return tuple(all_note_attributes)
 
         def clip_add_notes(clip, params: Tuple[Any] = ()):
             notes = []
@@ -101,8 +134,13 @@ def clip_add_notes(clip, params: Tuple[Any] = ()):
             clip.add_new_notes(tuple(notes))
 
         def clip_remove_notes(clip, params: Tuple[Any] = ()):
-            start_pitch, pitch_span, start_time, time_span = params
-            clip.remove_notes_extended(start_pitch, pitch_span, start_time, time_span)
+            if len(params) == 4:
+                pitch_start, pitch_span, time_start, time_span = params
+            elif len(params) == 0:
+                pitch_start, pitch_span, time_start, time_span = 0, 127, -8192, 16384
+            else:
+                raise ValueError("Invalid number of arguments for /clip/remove/notes. Either 0 or 4 arguments must be passed.")
+            clip.remove_notes_extended(pitch_start, pitch_span, time_start, time_span)
 
         self.osc_server.add_handler("/live/clip/get/notes", create_clip_callback(clip_get_notes))
         self.osc_server.add_handler("/live/clip/add/notes", create_clip_callback(clip_add_notes))
@@ -136,48 +174,25 @@ def clip_remove_playing_position_listener(track_clip_index, params: Tuple[Any] =
         self.osc_server.add_handler("/live/clip/stop_listen/playing_position",
                                     create_clip_callback(clip_remove_playing_position_listener, pass_clip_index=True))
 
-        def note_name_to_midi(name):
-            """ Maps a MIDI note name (D3, C#6) to a value.
-            Assumes that middle C is C4. """
-            note_names = [["C"],
-                          ["C#", "Db"],
-                          ["D"],
-                          ["D#", "Eb"],
-                          ["E"],
-                          ["F"],
-                          ["F#", "Gb"],
-                          ["G"],
-                          ["G#", "Ab"],
-                          ["A"],
-                          ["A#", "Bb"],
-                          ["B"]]
-
-            for index, names in enumerate(note_names):
-                if name in names:
-                    return index
-            return None
-
         def clips_filter_handler(params: Tuple):
             # TODO: Pre-cache clip notes
+            if len(self._clip_notes_cache) == 0:
+                self.logger.warning("Building clip notes cache...")
+                self._build_clip_name_cache()
+            else:
+                self.logger.warning("Found existing clip notes cache (len = %d)" % len(self._clip_notes_cache))
             note_indices = [note_name_to_midi(name) for name in params]
-            regex = "([_-])([A-G][A-G#b1-9-]*)$"
 
-            self.logger.info("Got note indices: %s" % note_indices)
-            for track in self.song.tracks:
-                for clip_slot in track.clip_slots:
-                    if clip_slot.has_clip:
+            self.logger.warning("Got note indices: %s" % note_indices)
+            for track_index, track in enumerate(self.song.tracks):
+                for clip_slot_index, clip_slot in enumerate(track.clip_slots):
+                    clip_notes_list = self._clip_notes_cache[track_index][clip_slot_index]
+                    if clip_notes_list:
                         clip = clip_slot.clip
-                        clip_name = clip.name
-                        match = re.search(regex, clip_name)
-                        if match:
-                            clip_notes_str = match.group(2)
-                            clip_notes_str = re.sub("[1-9]", "", clip_notes_str)
-                            clip_notes_list = clip_notes_str.split("-")
-                            clip_notes_list = [note_name_to_midi(name) for name in clip_notes_list]
-                            if all(note in note_indices for note in clip_notes_list):
-                                clip.muted = False
-                            else:
-                                clip.muted = True
+                        if all(note in note_indices for note in clip_notes_list):
+                            clip.muted = False
+                        else:
+                            clip.muted = True
 
         self.osc_server.add_handler("/live/clips/filter", clips_filter_handler)
 
@@ -194,3 +209,19 @@ def clips_unfilter_handler(params: Tuple):
 
         self.osc_server.add_handler("/live/clips/unfilter", clips_unfilter_handler)
 
+    def _build_clip_name_cache(self):
+        regex = "([_-])([A-G][A-G#b1-9-]*)$"
+        for track_index, track in enumerate(self.song.tracks):
+            self._clip_notes_cache.append([])
+            for clip_slot_index, clip_slot in enumerate(track.clip_slots):
+                self._clip_notes_cache[-1].append([])
+                if clip_slot.has_clip:
+                    clip = clip_slot.clip
+                    clip_name = clip.name
+                    match = re.search(regex, clip_name)
+                    if match:
+                        clip_notes_str = match.group(2)
+                        clip_notes_str = re.sub("[1-9]", "", clip_notes_str)
+                        clip_notes_list = clip_notes_str.split("-")
+                        clip_notes_list = [note_name_to_midi(name) for name in clip_notes_list]
+                        self._clip_notes_cache[-1][-1] = clip_notes_list
diff --git a/abletonosc/clip_slot.py b/abletonosc/clip_slot.py
@@ -57,3 +57,11 @@ def clip_slot_callback(params: Tuple[Any]):
         for prop in properties_rw:
             self.osc_server.add_handler("/live/clip_slot/set/%s" % prop,
                                         create_clip_slot_callback(self._set_property, prop))
+
+        def duplicate_clip_slot(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))
diff --git a/abletonosc/osc_server.py b/abletonosc/osc_server.py
@@ -1,6 +1,7 @@
 from typing import Tuple, Any, Callable
 from .constants import OSC_LISTEN_PORT, OSC_RESPONSE_PORT
 from ..pythonosc.osc_message import OscMessage, ParseError
+from ..pythonosc.osc_bundle import OscBundle
 from ..pythonosc.osc_message_builder import OscMessageBuilder, BuildError
 
 import re
@@ -83,11 +84,70 @@ def send(self,
         except BuildError:
             self.logger.error("AbletonOSC: OSC build error: %s" % (traceback.format_exc()))
 
+    def process_message(self, message, remote_addr):
+        if message.address in self._callbacks:
+            callback = self._callbacks[message.address]
+            rv = callback(message.params)
+
+            if rv is not None:
+                assert isinstance(rv, tuple)
+                remote_hostname, _ = remote_addr
+                response_addr = (remote_hostname, self._response_port)
+                self.send(address=message.address,
+                          params=rv,
+                          remote_addr=response_addr)
+        elif "*" in message.address:
+            regex = message.address.replace("*", "[^/]+")
+            for callback_address, callback in self._callbacks.items():
+                if re.match(regex, callback_address):
+                    try:
+                        rv = callback(message.params)
+                    except ValueError:
+                        #--------------------------------------------------------------------------------
+                        # Don't throw errors for queries that require more arguments
+                        # (e.g. /live/track/get/send with no args)
+                        #--------------------------------------------------------------------------------
+                        continue
+                    except AttributeError:
+                        #--------------------------------------------------------------------------------
+                        # Don't throw errors when trying to create listeners for properties that can't
+                        # be listened for (e.g. can_be_armed, is_foldable)
+                        #--------------------------------------------------------------------------------
+                        continue
+                    if rv is not None:
+                        assert isinstance(rv, tuple)
+                        remote_hostname, _ = remote_addr
+                        response_addr = (remote_hostname, self._response_port)
+                        self.send(address=callback_address,
+                                  params=rv,
+                                  remote_addr=response_addr)
+        else:
+            self.logger.error("AbletonOSC: Unknown OSC address: %s" % message.address)
+
+    def process_bundle(self, bundle, remote_addr):
+        for i in bundle:
+            if OscBundle.dgram_is_bundle(i.dgram):
+                self.process_bundle(i, remote_addr)
+            else:
+                self.process_message(i, remote_addr)
+
+    def parse_bundle(self, data, remote_addr):
+        bundle = OscBundle(data)
+        if bundle.num_contents == 0:
+            try:
+                message = OscMessage(data)
+                self.process_message(message, remote_addr)
+            except ParseError:
+                self.logger.error("AbletonOSC: OSC parse error: %s" % (traceback.format_exc()))
+        else:
+            self.process_bundle(bundle, remote_addr)
+
     def process(self) -> None:
         """
         Synchronously process all data queued on the OSC socket.
         """
         try:
+            repeats = 0
             while True:
                 #--------------------------------------------------------------------------------
                 # Loop until no more data is available.
@@ -100,49 +160,7 @@ def process(self) -> None:
                 # This is slightly ugly and prevents registering listeners from different IPs.
                 #--------------------------------------------------------------------------------
                 self._remote_addr = (remote_addr[0], OSC_RESPONSE_PORT)
-                try:
-                    message = OscMessage(data)
-
-                    if message.address in self._callbacks:
-                        callback = self._callbacks[message.address]
-                        rv = callback(message.params)
-
-                        if rv is not None:
-                            assert isinstance(rv, tuple)
-                            remote_hostname, _ = remote_addr
-                            response_addr = (remote_hostname, self._response_port)
-                            self.send(address=message.address,
-                                      params=rv,
-                                      remote_addr=response_addr)
-                    elif "*" in message.address:
-                        regex = message.address.replace("*", "[^/]+")
-                        for callback_address, callback in self._callbacks.items():
-                            if re.match(regex, callback_address):
-                                try:
-                                    rv = callback(message.params)
-                                except ValueError:
-                                    #--------------------------------------------------------------------------------
-                                    # Don't throw errors for queries that require more arguments
-                                    # (e.g. /live/track/get/send with no args)
-                                    #--------------------------------------------------------------------------------
-                                    continue
-                                except AttributeError:
-                                    #--------------------------------------------------------------------------------
-                                    # Don't throw errors when trying to create listeners for properties that can't
-                                    # be listened for (e.g. can_be_armed, is_foldable)
-                                    #--------------------------------------------------------------------------------
-                                    continue
-                                if rv is not None:
-                                    assert isinstance(rv, tuple)
-                                    remote_hostname, _ = remote_addr
-                                    response_addr = (remote_hostname, self._response_port)
-                                    self.send(address=callback_address,
-                                              params=rv,
-                                              remote_addr=response_addr)
-                    else:
-                        self.logger.error("AbletonOSC: Unknown OSC address: %s" % message.address)
-                except ParseError:
-                    self.logger.error("AbletonOSC: OSC parse error: %s" % (traceback.format_exc()))
+                self.parse_bundle(data, remote_addr)
 
         except socket.error as e:
             if e.errno == errno.ECONNRESET:

diff --git a/abletonosc/song.py b/abletonosc/song.py
@@ -1,4 +1,7 @@
+import os
+import tempfile
 import Live
+import json
 from functools import partial
 from typing import Tuple, Any
 
@@ -65,12 +68,13 @@ def init_api(self):
         ]
 
         #--------------------------------------------------------------------------------
-        # Callbacks for Song: properties (read-only)
+        # Callbacks for Songi: properties (read-only)
         #--------------------------------------------------------------------------------
         properties_r = [
             "can_redo",
             "can_undo",
-            "is_playing"
+            "is_playing",
+            "song_length",
         ]
 
         for prop in properties_r + properties_rw:
@@ -144,6 +148,58 @@ def song_get_track_data(params):
             return tuple(rv)
         self.osc_server.add_handler("/live/song/get/track_data", song_get_track_data)
 
+
+        def song_export_structure(params):
+            tracks = []
+            for track_index, track in enumerate(self.song.tracks):
+                group_track = None
+                if track.group_track is not None:
+                    group_track = list(self.song.tracks).index(track.group_track)
+                track_data = {
+                    "index": track_index,
+                    "name": track.name,
+                    "is_foldable": track.is_foldable,
+                    "group_track": group_track,
+                    "clips": [],
+                    "devices": []
+                }
+                for clip_index, clip_slot in enumerate(track.clip_slots):
+                    if clip_slot.clip:
+                        clip_data = {
+                            "index": clip_index,
+                            "name": clip_slot.clip.name,
+                            "length": clip_slot.clip.length,
+                        }
+                        track_data["clips"].append(clip_data)
+
+                for device_index, device in enumerate(track.devices):
+                    device_data = {
+                        "class_name": device.class_name,
+                        "type": device.type,
+                        "name": device.name,
+                        "parameters": []
+                    }
+                    for parameter in device.parameters:
+                        device_data["parameters"].append({
+                            "name": parameter.name,
+                            "value": parameter.value,
+                            "min": parameter.min,
+                            "max": parameter.max,
+                            "is_quantized": parameter.is_quantized,
+                        })
+                    track_data["devices"].append(device_data)
+
+                tracks.append(track_data)
+            song = {
+                "tracks": tracks
+            }
+
+            fd = open(os.path.join(tempfile.gettempdir(), "abletonosc-song-structure.json"), "w")
+            json.dump(song, fd)
+            fd.close()
+            return (1,)
+        self.osc_server.add_handler("/live/song/export/structure", song_export_structure)
+
         #--------------------------------------------------------------------------------
         # Callbacks for Song: Scene properties
         #--------------------------------------------------------------------------------