renato97/ableton-mcp-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
6d080d43b3c869e644c709f0b13bead55fe66559
ableton-mcp-ai/AbletonMCP_AI/abletonmcp_runtime.py

2957 lines
128 KiB
Python
Raw Blame History

# AbletonMCP/init.py
from __future__ import absolute_import, print_function, unicode_literals
from _Framework.ControlSurface import ControlSurface
import socket
import json
import os
import threading
import time
import traceback
# Change queue import for Python 2
try:
import Queue as queue # Python 2
except ImportError:
import queue # Python 3
try:
string_types = basestring # Python 2
except NameError:
string_types = str # Python 3
# Constants for socket communication
DEFAULT_PORT = 9877
HOST = "localhost"
def create_instance(c_instance):
"""Create and return the AbletonMCP script instance"""
return AbletonMCP(c_instance)
class AbletonMCP(ControlSurface):
"""AbletonMCP Remote Script for Ableton Live"""
def __init__(self, c_instance):
"""Initialize the control surface"""
ControlSurface.__init__(self, c_instance)
self.log_message("AbletonMCP Remote Script initializing... [VERSION MODIFIED FOR DEBUG v2]")
# Socket server for communication
self.server = None
self.client_threads = []
self.server_thread = None
self.running = False
self._main_thread_tasks = queue.Queue()
self._recent_arrangement_clips = {}
# Cache the song reference for easier access
self._song = self.song()
# Start the socket server
self.start_server()
self.log_message("AbletonMCP initialized")
# Show a message in Ableton
self.show_message("AbletonMCP: Listening for commands on port " + str(DEFAULT_PORT))
def disconnect(self):
"""Called when Ableton closes or the control surface is removed"""
self.log_message("AbletonMCP disconnecting...")
self.running = False
# Stop the server
if self.server:
try:
self.server.close()
except:
pass
# Wait for the server thread to exit
if self.server_thread and self.server_thread.is_alive():
self.server_thread.join(1.0)
# Clean up any client threads
for client_thread in self.client_threads[:]:
if client_thread.is_alive():
# We don't join them as they might be stuck
self.log_message("Client thread still alive during disconnect")
ControlSurface.disconnect(self)
self.log_message("AbletonMCP disconnected")
def _enqueue_main_thread_task(self, callback):
"""Queue a task to be executed from Live's main thread."""
self._main_thread_tasks.put(callback)
def update_display(self):
"""Drain queued Live mutations from Ableton's main thread."""
processed = 0
while processed < 4:
try:
callback = self._main_thread_tasks.get_nowait()
except queue.Empty:
break
try:
callback()
except Exception as e:
self.log_message("Error in queued main thread task: " + str(e))
self.log_message(traceback.format_exc())
processed += 1
def start_server(self):
"""Start the socket server in a separate thread"""
try:
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server.bind((HOST, DEFAULT_PORT))
self.server.listen(5) # Allow up to 5 pending connections
self.running = True
self.server_thread = threading.Thread(target=self._server_thread)
self.server_thread.daemon = True
self.server_thread.start()
self.log_message("Server started on port " + str(DEFAULT_PORT))
except Exception as e:
self.log_message("Error starting server: " + str(e))
self.show_message("AbletonMCP: Error starting server - " + str(e))
def _server_thread(self):
"""Server thread implementation - handles client connections"""
try:
self.log_message("Server thread started")
# Set a timeout to allow regular checking of running flag
self.server.settimeout(1.0)
while self.running:
try:
# Accept connections with timeout
client, address = self.server.accept()
self.log_message("Connection accepted from " + str(address))
self.show_message("AbletonMCP: Client connected")
# Handle client in a separate thread
client_thread = threading.Thread(
target=self._handle_client,
args=(client,)
)
client_thread.daemon = True
client_thread.start()
# Keep track of client threads
self.client_threads.append(client_thread)
# Clean up finished client threads
self.client_threads = [t for t in self.client_threads if t.is_alive()]
except socket.timeout:
# No connection yet, just continue
continue
except Exception as e:
if self.running: # Only log if still running
self.log_message("Server accept error: " + str(e))
time.sleep(0.5)
self.log_message("Server thread stopped")
except Exception as e:
self.log_message("Server thread error: " + str(e))
def _handle_client(self, client):
"""Handle communication with a connected client"""
self.log_message("Client handler started")
client.settimeout(None) # No timeout for client socket
buffer = '' # Changed from b'' to '' for Python 2
try:
while self.running:
try:
# Receive data
data = client.recv(8192)
if not data:
# Client disconnected
self.log_message("Client disconnected")
break
# Accumulate data in buffer with explicit encoding/decoding
try:
# Python 3: data is bytes, decode to string
buffer += data.decode('utf-8')
except AttributeError:
# Python 2: data is already string
buffer += data
try:
# Try to parse command from buffer
command = json.loads(buffer) # Removed decode('utf-8')
buffer = '' # Clear buffer after successful parse
self.log_message("Received command: " + str(command.get("type", "unknown")))
# Process the command and get response
response = self._process_command(command)
# Send the response with explicit encoding
try:
# Python 3: encode string to bytes
client.sendall((json.dumps(response) + '\n').encode('utf-8'))
except AttributeError:
# Python 2: string is already bytes
client.sendall(json.dumps(response) + '\n')
except ValueError:
# Incomplete data, wait for more
continue
except Exception as e:
self.log_message("Error handling client data: " + str(e))
self.log_message(traceback.format_exc())
# Send error response if possible
error_response = {
"status": "error",
"message": str(e)
}
try:
# Python 3: encode string to bytes
client.sendall((json.dumps(error_response) + '\n').encode('utf-8'))
except AttributeError:
# Python 2: string is already bytes
client.sendall(json.dumps(error_response) + '\n')
except:
# If we can't send the error, the connection is probably dead
break
# For serious errors, break the loop
if not isinstance(e, ValueError):
break
except Exception as e:
self.log_message("Error in client handler: " + str(e))
finally:
try:
client.close()
except:
pass
self.log_message("Client handler stopped")
def _process_command(self, command):
"""Process a command from the client and return a response"""
command_type = command.get("type", "")
params = command.get("params", {})
# Initialize response
response = {
"status": "success",
"result": {}
}
try:
# Route the command to the appropriate handler
if command_type == "get_session_info":
response["result"] = self._get_session_info()
elif command_type == "get_track_info":
track_index = params.get("track_index", 0)
response["result"] = self._get_track_info(track_index)
# Commands that modify Live's state should be scheduled on the main thread
elif command_type in [
"create_midi_track", "create_audio_track", "create_return_track",
"set_track_name", "set_track_mute", "set_track_solo", "set_track_arm",
"set_track_volume", "set_track_pan", "set_track_send", "set_track_color",
"set_track_monitoring", "set_master_volume", "set_master_pan",
"create_clip", "delete_clip", "add_notes_to_clip", "set_clip_name",
"set_clip_loop", "set_tempo", "set_signature", "set_current_song_time",
"set_loop", "set_loop_region", "set_metronome", "set_overdub",
"set_record_mode", "fire_clip", "stop_clip", "stop_all_clips",
"start_playback", "stop_playback", "fire_scene", "create_scene",
"set_scene_name", "delete_scene", "load_instrument_or_effect",
"load_browser_item", "load_browser_item_by_name",
"load_browser_item_at_path", "set_device_parameter", "set_device_on",
"generate_track", "clear_all_tracks", "load_device",
"create_arrangement_audio_pattern",
"set_scene_color", "jump_to", "loop_selection",
"show_arrangement_view", "delete_track", "stop"
]:
# Use a thread-safe approach with a response queue
response_queue = queue.Queue()
# Define a function to execute on the main thread
def main_thread_task():
try:
result = None
if command_type == "create_midi_track":
index = params.get("index", -1)
result = self._create_midi_track(index)
elif command_type == "create_audio_track":
index = params.get("index", -1)
result = self._create_audio_track(index)
elif command_type == "create_return_track":
result = self._create_return_track()
elif command_type == "set_track_name":
track_index = params.get("track_index", 0)
name = params.get("name", "")
result = self._set_track_name(track_index, name)
elif command_type == "set_track_mute":
track_index = params.get("track_index", 0)
mute = params.get("mute", False)
result = self._set_track_mute(track_index, mute)
elif command_type == "set_track_solo":
track_index = params.get("track_index", 0)
solo = params.get("solo", False)
result = self._set_track_solo(track_index, solo)
elif command_type == "set_track_arm":
track_index = params.get("track_index", 0)
arm = params.get("arm", False)
result = self._set_track_arm(track_index, arm)
elif command_type == "set_track_volume":
track_index = params.get("track_index", 0)
volume = params.get("volume", 0.85)
result = self._set_track_volume(track_index, volume)
elif command_type == "set_track_pan":
track_index = params.get("track_index", 0)
pan = params.get("pan", 0.0)
result = self._set_track_pan(track_index, pan)
elif command_type == "set_track_send":
track_index = params.get("track_index", 0)
send_index = params.get("send_index", 0)
value = params.get("value", 0.0)
result = self._set_track_send(track_index, send_index, value)
elif command_type == "set_track_color":
track_index = params.get("track_index", 0)
color = params.get("color", 0)
result = self._set_track_color(track_index, color)
elif command_type == "set_track_monitoring":
track_index = params.get("track_index", 0)
state = params.get("state", 0)
result = self._set_track_monitoring(track_index, state)
elif command_type == "set_master_volume":
volume = params.get("volume", 0.85)
result = self._set_master_volume(volume)
elif command_type == "set_master_pan":
pan = params.get("pan", 0.0)
result = self._set_master_pan(pan)
elif command_type == "create_clip":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
length = params.get("length", 4.0)
result = self._create_clip(track_index, clip_index, length)
elif command_type == "create_arrangement_clip":
track_index = params.get("track_index", 0)
start_time = params.get("start_time", 0.0)
length = params.get("length", 4.0)
track_type = params.get("track_type", "track")
result = self._create_arrangement_clip(track_index, start_time, length, track_type)
elif command_type == "create_arrangement_audio_pattern":
track_index = params.get("track_index", 0)
file_path = params.get("file_path", "")
positions = params.get("positions", [])
name = params.get("name", "")
result = self._create_arrangement_audio_pattern(track_index, file_path, positions, name)
elif command_type == "delete_clip":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
result = self._delete_clip(track_index, clip_index)
elif command_type == "add_notes_to_clip":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
notes = params.get("notes", [])
result = self._add_notes_to_clip(track_index, clip_index, notes)
elif command_type == "add_notes_to_arrangement_clip":
track_index = params.get("track_index", 0)
start_time = params.get("start_time", 0.0)
notes = params.get("notes", [])
track_type = params.get("track_type", "track")
result = self._add_notes_to_arrangement_clip(track_index, start_time, notes, track_type)
elif command_type == "duplicate_clip_to_arrangement":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
start_time = params.get("start_time", 0.0)
track_type = params.get("track_type", "track")
result = self._duplicate_clip_to_arrangement(track_index, clip_index, start_time, track_type)
elif command_type == "set_clip_name":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
name = params.get("name", "")
result = self._set_clip_name(track_index, clip_index, name)
elif command_type == "set_clip_loop":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
loop_start = params.get("loop_start", None)
loop_end = params.get("loop_end", None)
loop_length = params.get("loop_length", None)
looping = params.get("looping", None)
result = self._set_clip_loop(
track_index,
clip_index,
loop_start,
loop_end,
loop_length,
looping
)
elif command_type == "set_tempo":
tempo = params.get("tempo", 120.0)
result = self._set_tempo(tempo)
elif command_type == "set_signature":
numerator = params.get("numerator", 4)
denominator = params.get("denominator", 4)
result = self._set_signature(numerator, denominator)
elif command_type == "set_current_song_time":
time_value = params.get("time", 0.0)
result = self._set_current_song_time(time_value)
elif command_type == "set_loop":
enabled = params.get("enabled", False)
result = self._set_loop(enabled)
elif command_type == "set_loop_region":
start = params.get("start", 0.0)
length = params.get("length", 4.0)
result = self._set_loop_region(start, length)
elif command_type == "set_metronome":
enabled = params.get("enabled", False)
result = self._set_metronome(enabled)
elif command_type == "set_overdub":
enabled = params.get("enabled", False)
result = self._set_overdub(enabled)
elif command_type == "set_record_mode":
enabled = params.get("enabled", False)
result = self._set_record_mode(enabled)
elif command_type == "fire_clip":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
result = self._fire_clip(track_index, clip_index)
elif command_type == "stop_clip":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
result = self._stop_clip(track_index, clip_index)
elif command_type == "stop_all_clips":
result = self._stop_all_clips()
elif command_type == "start_playback":
result = self._start_playback()
elif command_type == "stop_playback":
result = self._stop_playback()
elif command_type == "fire_scene":
scene_index = params.get("scene_index", 0)
result = self._fire_scene(scene_index)
elif command_type == "create_scene":
index = params.get("index", -1)
result = self._create_scene(index)
elif command_type == "set_scene_name":
scene_index = params.get("scene_index", 0)
name = params.get("name", "")
result = self._set_scene_name(scene_index, name)
elif command_type == "delete_scene":
scene_index = params.get("scene_index", 0)
result = self._delete_scene(scene_index)
elif command_type == "set_scene_color":
scene_index = params.get("scene_index", 0)
color = params.get("color", 0)
result = self._set_scene_color(scene_index, color)
elif command_type == "load_instrument_or_effect":
track_index = params.get("track_index", 0)
uri = params.get("uri", "")
result = self._load_instrument_or_effect(track_index, uri)
elif command_type == "load_device":
track_index = params.get("track_index", 0)
device_name = params.get("device_name", "")
track_type = params.get("track_type", "track")
result = self._load_device(track_index, device_name, track_type)
elif command_type == "load_browser_item":
track_index = params.get("track_index", 0)
item_uri = params.get("item_uri", "")
result = self._load_browser_item(track_index, item_uri)
elif command_type == "load_browser_item_by_name":
track_index = params.get("track_index", 0)
query = params.get("query", "")
category_type = params.get("category_type", "all")
max_depth = params.get("max_depth", 5)
result = self._load_browser_item_by_name(
track_index,
query,
category_type,
max_depth
)
elif command_type == "load_browser_item_at_path":
track_index = params.get("track_index", 0)
path = params.get("path", "")
item_name = params.get("item_name", None)
result = self._load_browser_item_at_path(
track_index,
path,
item_name
)
elif command_type == "set_device_parameter":
track_index = params.get("track_index", 0)
track_type = params.get("track_type", "track")
device_index = params.get("device_index", 0)
parameter_index = params.get("parameter_index", None)
parameter_name = params.get("parameter_name", params.get("parameter", None))
value = params.get("value", 0.0)
result = self._set_device_parameter(
track_index,
device_index,
parameter_index,
parameter_name,
value,
track_type
)
elif command_type == "set_device_on":
track_index = params.get("track_index", 0)
device_index = params.get("device_index", 0)
enabled = params.get("enabled", True)
result = self._set_device_on(track_index, device_index, enabled)
elif command_type == "jump_to":
time_value = params.get("time", 0.0)
result = self._jump_to(time_value)
elif command_type == "loop_selection":
start = params.get("start", 0.0)
length = params.get("length", 4.0)
enable = params.get("enable", None)
result = self._loop_selection(start, length, enable)
elif command_type == "show_arrangement_view":
result = self._show_arrangement_view()
elif command_type == "delete_track":
track_index = params.get("track_index", 0)
result = self._delete_track(track_index)
elif command_type == "stop":
result = self._stop_playback()
elif command_type == "generate_track":
self._generate_track_async(params, response_queue)
return
elif command_type == "clear_all_tracks":
result = self._clear_all_tracks()
# Put the result in the queue
response_queue.put({"status": "success", "result": result})
except Exception as e:
self.log_message("Error in main thread task: " + str(e))
self.log_message(traceback.format_exc())
response_queue.put({"status": "error", "message": str(e)})
# Queue the task to run on Ableton's main thread via update_display
self._enqueue_main_thread_task(main_thread_task)
# Determine timeout based on command type
if command_type == "generate_track":
timeout_seconds = 180.0 # Extended timeout for track generation
elif command_type in (
"create_arrangement_clip",
"add_notes_to_arrangement_clip",
"duplicate_clip_to_arrangement",
"create_arrangement_audio_pattern",
):
timeout_seconds = 60.0 # Session->Arrangement fallback records in real time
else:
timeout_seconds = 10.0
# Wait for the response with a timeout
try:
task_response = response_queue.get(timeout=timeout_seconds)
if task_response.get("status") == "error":
response["status"] = "error"
response["message"] = task_response.get("message", "Unknown error")
else:
response["result"] = task_response.get("result", {})
except queue.Empty:
response["status"] = "error"
response["message"] = "Timeout waiting for operation to complete"
elif command_type == "get_tracks":
response["result"] = self._get_tracks()
elif command_type == "get_clips":
track_index = params.get("track_index", 0)
track_type = params.get("track_type", "track")
response["result"] = self._get_clips_for_type(track_index, track_type)
elif command_type == "get_clip_info":
track_index = params.get("track_index", 0)
clip_index = params.get("clip_index", 0)
track_type = params.get("track_type", "track")
response["result"] = self._get_clip_info(track_index, clip_index, track_type)
elif command_type == "get_scenes":
response["result"] = self._get_scenes()
elif command_type == "get_track_devices":
track_index = params.get("track_index", 0)
response["result"] = self._get_track_devices(track_index)
elif command_type == "get_devices":
track_index = params.get("track_index", 0)
track_type = params.get("track_type", "track")
response["result"] = self._get_track_devices_for_type(track_index, track_type)
elif command_type == "get_all_tracks":
response["result"] = self._get_tracks()
elif command_type == "get_set_info":
response["result"] = self._get_session_info()
elif command_type == "get_master_info":
response["result"] = self._get_master_info()
elif command_type == "get_device_parameters":
track_index = params.get("track_index", 0)
track_type = params.get("track_type", "track")
device_index = params.get("device_index", 0)
response["result"] = self._get_device_parameters(track_index, device_index, track_type)
elif command_type == "search_browser_items":
query = params.get("query", "")
category_type = params.get("category_type", "all")
max_results = params.get("max_results", 25)
max_depth = params.get("max_depth", 5)
loadable_only = params.get("loadable_only", False)
response["result"] = self._search_browser_items(
query,
category_type,
max_results,
max_depth,
loadable_only
)
elif command_type == "get_browser_item":
uri = params.get("uri", None)
path = params.get("path", None)
response["result"] = self._get_browser_item(uri, path)
elif command_type == "get_browser_categories":
category_type = params.get("category_type", "all")
response["result"] = self._get_browser_categories(category_type)
elif command_type == "get_browser_items":
path = params.get("path", "")
item_type = params.get("item_type", "all")
response["result"] = self._get_browser_items(path, item_type)
# Add the new browser commands
elif command_type == "get_browser_tree":
category_type = params.get("category_type", "all")
max_depth = params.get("max_depth", 2)
response["result"] = self.get_browser_tree(category_type, max_depth)
elif command_type == "get_browser_items_at_path":
path = params.get("path", "")
response["result"] = self.get_browser_items_at_path(path)
else:
response["status"] = "error"
response["message"] = "Unknown command: " + command_type
except Exception as e:
self.log_message("Error processing command: " + str(e))
self.log_message(traceback.format_exc())
response["status"] = "error"
response["message"] = str(e)
return response
# Command implementations
def _get_session_info(self):
"""Get information about the current session"""
try:
result = {
"tempo": self._song.tempo,
"signature_numerator": self._song.signature_numerator,
"signature_denominator": self._song.signature_denominator,
"is_playing": self._song.is_playing,
"current_song_time": self._song.current_song_time,
"loop": self._song.loop,
"loop_start": self._song.loop_start,
"loop_length": self._song.loop_length,
"metronome": self._song.metronome,
"overdub": self._song.overdub,
"num_tracks": len(self._song.tracks),
"track_count": len(self._song.tracks),
"num_return_tracks": len(self._song.return_tracks),
"return_track_count": len(self._song.return_tracks),
"num_scenes": len(self._song.scenes),
"scene_count": len(self._song.scenes),
"master_track": {
"name": "Master",
"volume": self._song.master_track.mixer_device.volume.value,
"panning": self._song.master_track.mixer_device.panning.value
}
}
if hasattr(self._song, "record_mode"):
result["record_mode"] = self._song.record_mode
elif hasattr(self._song, "session_record"):
result["record_mode"] = self._song.session_record
return result
except Exception as e:
self.log_message("Error getting session info: " + str(e))
raise
def _get_track_info(self, track_index):
"""Get information about a track"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
track_type = "midi" if track.has_midi_input else "audio" if track.has_audio_input else "unknown"
# Get clip slots
clip_slots = []
for slot_index, slot in enumerate(track.clip_slots):
clip_info = None
if slot.has_clip:
clip = slot.clip
clip_info = {
"name": clip.name,
"length": clip.length,
"is_playing": clip.is_playing,
"is_recording": clip.is_recording
}
clip_slots.append({
"index": slot_index,
"has_clip": slot.has_clip,
"clip": clip_info
})
# Get devices
devices = []
for device_index, device in enumerate(track.devices):
devices.append({
"index": device_index,
"name": device.name,
"class_name": device.class_name,
"type": self._get_device_type(device)
})
sends = []
if hasattr(track.mixer_device, "sends"):
for send in track.mixer_device.sends:
sends.append(send.value)
color_value = None
if hasattr(track, "color"):
color_value = track.color
elif hasattr(track, "color_index"):
color_value = track.color_index
result = {
"index": track_index,
"name": track.name,
"track_type": track_type,
"is_audio_track": track.has_audio_input,
"is_midi_track": track.has_midi_input,
"mute": self._safe_getattr(track, "mute", False),
"solo": self._safe_getattr(track, "solo", False),
"arm": self._safe_getattr(track, "arm", False),
"volume": self._safe_mixer_value(track, "volume"),
"panning": self._safe_mixer_value(track, "panning"),
"sends": sends,
"clip_slots": clip_slots,
"devices": devices,
"device_count": len(track.devices)
}
if color_value is not None:
result["color"] = color_value
return result
except Exception as e:
self.log_message("Error getting track info: " + str(e))
raise
def _summarize_track(self, track, index, track_type):
"""Summarize a track for listing."""
info = {
"index": index,
"name": track.name,
"type": track_type
}
mute = self._safe_getattr(track, "mute")
if mute is not None:
info["mute"] = mute
solo = self._safe_getattr(track, "solo")
if solo is not None:
info["solo"] = solo
if track_type == "track":
arm = self._safe_getattr(track, "arm")
if arm is not None:
info["arm"] = arm
if hasattr(track, "mixer_device"):
volume = self._safe_mixer_value(track, "volume")
panning = self._safe_mixer_value(track, "panning")
if volume is not None:
info["volume"] = volume
if panning is not None:
info["panning"] = panning
if hasattr(track, "has_audio_input"):
info["is_audio_track"] = track.has_audio_input
if hasattr(track, "has_midi_input"):
info["is_midi_track"] = track.has_midi_input
if hasattr(track, "devices"):
info["device_count"] = len(track.devices)
if hasattr(track, "color"):
info["color"] = track.color
elif hasattr(track, "color_index"):
info["color"] = track.color_index
return info
def _get_tracks(self):
"""Get summary info for all tracks, return tracks, and master."""
try:
tracks = []
for index, track in enumerate(self._song.tracks):
tracks.append(self._summarize_track(track, index, "track"))
return_tracks = []
for index, track in enumerate(self._song.return_tracks):
return_tracks.append(self._summarize_track(track, index, "return"))
master = self._summarize_track(self._song.master_track, -1, "master")
return {
"tracks": tracks,
"return_tracks": return_tracks,
"master_track": master
}
except Exception as e:
self.log_message("Error getting tracks: " + str(e))
raise
def _safe_getattr(self, obj, attr_name, default=None):
"""Read Live API attributes without exploding on optional properties."""
try:
return getattr(obj, attr_name)
except Exception:
return default
def _safe_mixer_value(self, track, attr_name, default=None):
try:
mixer = getattr(track, "mixer_device", None)
if mixer is None:
return default
parameter = getattr(mixer, attr_name, None)
if parameter is None:
return default
return getattr(parameter, "value", default)
except Exception:
return default
def _create_midi_track(self, index):
"""Create a new MIDI track at the specified index"""
try:
# Create the track
self._song.create_midi_track(index)
# Get the new track
new_track_index = len(self._song.tracks) - 1 if index == -1 else index
new_track = self._song.tracks[new_track_index]
result = {
"index": new_track_index,
"name": new_track.name
}
return result
except Exception as e:
self.log_message("Error creating MIDI track: " + str(e))
raise
def _create_audio_track(self, index):
"""Create a new audio track at the specified index"""
try:
self._song.create_audio_track(index)
new_track_index = len(self._song.tracks) - 1 if index == -1 else index
new_track = self._song.tracks[new_track_index]
return {
"index": new_track_index,
"name": new_track.name
}
except Exception as e:
self.log_message("Error creating audio track: " + str(e))
raise
def _create_return_track(self):
"""Create a new return track"""
try:
if not hasattr(self._song, "create_return_track"):
raise RuntimeError("Return tracks are not available in this Live version")
self._song.create_return_track()
new_index = len(self._song.return_tracks) - 1
new_track = self._song.return_tracks[new_index]
return {
"index": new_index,
"name": new_track.name
}
except Exception as e:
self.log_message("Error creating return track: " + str(e))
raise
def _resolve_track_reference(self, track_index, track_type):
"""Resolve a regular, return, or master track reference."""
normalized = str(track_type or "track").lower()
if normalized in ["return", "return_track", "return_tracks"]:
if track_index < 0 or track_index >= len(self._song.return_tracks):
raise IndexError("Return track index out of range")
return self._song.return_tracks[track_index]
if normalized in ["master", "master_track"]:
return self._song.master_track
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
return self._song.tracks[track_index]
def _set_track_mute(self, track_index, mute):
"""Set track mute state"""
try:
track = self._song.tracks[track_index]
track.mute = bool(mute)
return {"mute": track.mute}
except Exception as e:
self.log_message("Error setting track mute: " + str(e))
raise
def _set_track_solo(self, track_index, solo):
"""Set track solo state"""
try:
track = self._song.tracks[track_index]
track.solo = bool(solo)
return {"solo": track.solo}
except Exception as e:
self.log_message("Error setting track solo: " + str(e))
raise
def _set_track_arm(self, track_index, arm):
"""Set track arm state"""
try:
track = self._song.tracks[track_index]
if not hasattr(track, "arm"):
raise RuntimeError("Track does not support arm")
track.arm = bool(arm)
return {"arm": track.arm}
except Exception as e:
self.log_message("Error setting track arm: " + str(e))
raise
def _set_track_volume(self, track_index, volume):
"""Set track volume"""
try:
track = self._song.tracks[track_index]
track.mixer_device.volume.value = float(volume)
return {"volume": track.mixer_device.volume.value}
except Exception as e:
self.log_message("Error setting track volume: " + str(e))
raise
def _set_track_pan(self, track_index, pan):
"""Set track panning"""
try:
track = self._song.tracks[track_index]
track.mixer_device.panning.value = float(pan)
return {"panning": track.mixer_device.panning.value}
except Exception as e:
self.log_message("Error setting track pan: " + str(e))
raise
def _set_track_send(self, track_index, send_index, value):
"""Set track send level"""
try:
track = self._song.tracks[track_index]
sends = track.mixer_device.sends
if send_index < 0 or send_index >= len(sends):
raise IndexError("Send index out of range")
sends[send_index].value = float(value)
return {"send_index": send_index, "value": sends[send_index].value}
except Exception as e:
self.log_message("Error setting track send: " + str(e))
raise
def _set_track_color(self, track_index, color):
"""Set track color index or value"""
try:
track = self._song.tracks[track_index]
if hasattr(track, "color"):
track.color = int(color)
return {"color": track.color}
if hasattr(track, "color_index"):
track.color_index = int(color)
return {"color": track.color_index}
raise RuntimeError("Track color is not supported")
except Exception as e:
self.log_message("Error setting track color: " + str(e))
raise
def _set_track_monitoring(self, track_index, state):
"""Set track monitoring state (0=off,1=auto,2=in)"""
try:
track = self._song.tracks[track_index]
if not hasattr(track, "current_monitoring_state"):
raise RuntimeError("Track does not support monitoring state")
track.current_monitoring_state = int(state)
return {"current_monitoring_state": track.current_monitoring_state}
except Exception as e:
self.log_message("Error setting track monitoring: " + str(e))
raise
def _set_master_volume(self, volume):
"""Set master volume"""
try:
self._song.master_track.mixer_device.volume.value = float(volume)
return {"volume": self._song.master_track.mixer_device.volume.value}
except Exception as e:
self.log_message("Error setting master volume: " + str(e))
raise
def _set_master_pan(self, pan):
"""Set master panning"""
try:
self._song.master_track.mixer_device.panning.value = float(pan)
return {"panning": self._song.master_track.mixer_device.panning.value}
except Exception as e:
self.log_message("Error setting master pan: " + str(e))
raise
def _set_track_name(self, track_index, name):
"""Set the name of a track"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
# Set the name
track = self._song.tracks[track_index]
track.name = name
result = {
"name": track.name
}
return result
except Exception as e:
self.log_message("Error setting track name: " + str(e))
raise
def _delete_track(self, track_index):
"""Delete a regular track."""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
deleted_name = self._song.tracks[track_index].name
self._song.delete_track(track_index)
return {"deleted": True, "name": deleted_name}
except Exception as e:
self.log_message("Error deleting track: " + str(e))
raise
def _create_clip(self, track_index, clip_index, length):
"""Create a new MIDI clip in the specified track and clip slot"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
# Check if the clip slot already has a clip
if clip_slot.has_clip:
raise Exception("Clip slot already has a clip")
# Create the clip
clip_slot.create_clip(length)
result = {
"name": clip_slot.clip.name,
"length": clip_slot.clip.length
}
return result
except Exception as e:
self.log_message("Error creating clip: " + str(e))
raise
def _find_or_create_empty_clip_slot(self, track):
"""Find an empty clip slot on a track, creating a new scene if needed."""
for slot_index, slot in enumerate(getattr(track, "clip_slots", [])):
if not getattr(slot, "has_clip", False):
return slot_index
if not hasattr(self._song, "create_scene"):
raise RuntimeError("No empty clip slots available and create_scene is unsupported")
self._song.create_scene(-1)
return len(getattr(track, "clip_slots", [])) - 1
def _locate_arrangement_clip(self, track, start_time, tolerance=0.05, expected_length=None):
"""Locate the closest arrangement clip near the requested start time."""
candidates = []
seen = set()
minimum_length = None
if expected_length is not None:
try:
expected_length = max(float(expected_length), 0.0)
minimum_length = 0.25 if expected_length <= 1.0 else max(1.0, expected_length * 0.25)
except Exception:
minimum_length = None
for attr_name in ("clips", "arrangement_clips"):
try:
arrangement_source = getattr(track, attr_name, None)
except Exception:
arrangement_source = None
if arrangement_source is None:
continue
try:
iterator = list(arrangement_source)
except Exception:
continue
for clip in iterator:
if clip is None or id(clip) in seen:
continue
seen.add(id(clip))
clip_start = self._safe_getattr(clip, "start_time", None)
if clip_start is None:
continue
clip_length = float(self._safe_getattr(clip, "length", 0.0) or 0.0)
if minimum_length is not None and clip_length < minimum_length:
continue
candidates.append((clip, float(clip_start), clip_length))
best_clip = None
best_score = None
max_window = max(float(tolerance), 1.5)
for clip, clip_start, clip_length in candidates:
diff = abs(float(clip_start) - float(start_time))
if diff > max_window:
continue
length_penalty = 0.0
if expected_length is not None and clip_length > 0:
length_penalty = abs(float(clip_length) - float(expected_length)) * 0.1
score = diff + length_penalty
if best_score is None or score < best_score:
best_score = score
best_clip = clip
return best_clip
def _record_session_clip_to_arrangement(self, track_index, clip_index, start_time, length, track_type="track"):
"""Record a session clip into Arrangement View when direct MIDI clip creation is unavailable."""
track = self._resolve_track_reference(track_index, track_type)
clip_slot = track.clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
bpm = float(getattr(self._song, "tempo", 120.0) or 120.0)
record_seconds = max(0.35, float(length) * 60.0 / max(1.0, bpm) + 0.35)
try:
self._stop_playback()
except Exception:
pass
try:
self._stop_all_clips()
except Exception:
pass
try:
self._show_arrangement_view()
except Exception:
pass
try:
if hasattr(self._song, "loop"):
self._song.loop = False
except Exception:
pass
previous_arm = self._safe_getattr(track, "arm", None)
try:
self._jump_to(float(start_time))
if previous_arm is not None and not bool(previous_arm):
try:
track.arm = True
except Exception:
pass
self._set_record_mode(True)
self._set_overdub(False)
clip_slot.fire()
time.sleep(0.12)
self._start_playback()
time.sleep(record_seconds)
self._stop_playback()
finally:
try:
self._set_record_mode(False)
except Exception:
pass
if previous_arm is not None:
try:
track.arm = bool(previous_arm)
except Exception:
pass
try:
clip_slot.stop()
except Exception:
pass
target_clip = None
for current_tolerance in (0.05, 0.1, 0.25, 0.5, 1.0, 1.5):
for _ in range(5):
target_clip = self._locate_arrangement_clip(
track,
start_time,
tolerance=current_tolerance,
expected_length=length,
)
if target_clip is not None:
break
time.sleep(0.12)
if target_clip is not None:
break
if target_clip is None:
raise RuntimeError("Arrangement clip was not materialized")
self._recent_arrangement_clips[(int(track_index), round(float(start_time), 3))] = target_clip
return target_clip
def _create_arrangement_clip(self, track_index, start_time, length, track_type="track"):
"""Create a new MIDI clip in Arrangement View at the specified time"""
try:
track = self._resolve_track_reference(track_index, track_type)
clip = None
if hasattr(track, "create_clip"):
try:
clip = track.create_clip(start_time, length)
self._recent_arrangement_clips[(int(track_index), round(float(start_time), 3))] = clip
except Exception as direct_error:
self.log_message("Direct arrangement clip creation failed, using session fallback: " + str(direct_error))
if clip is None:
temp_slot_index = self._find_or_create_empty_clip_slot(track)
clip_slot = track.clip_slots[temp_slot_index]
if clip_slot.has_clip:
clip_slot.delete_clip()
clip_slot.create_clip(length)
try:
clip = self._record_session_clip_to_arrangement(
track_index,
temp_slot_index,
start_time,
length,
track_type,
)
finally:
try:
if clip_slot.has_clip:
clip_slot.delete_clip()
except Exception:
pass
result = {
"name": clip.name,
"length": clip.length,
"start_time": start_time
}
return result
except Exception as e:
self.log_message("Error creating arrangement clip: " + str(e))
raise
def _create_arrangement_audio_pattern(self, track_index, file_path, positions, name=""):
"""Create one or more arrangement audio clips from an absolute file path."""
try:
if str(file_path).startswith('/mnt/'):
parts = str(file_path)[5:].split('/', 1)
file_path = parts[0].upper() + ":\\" + parts[1].replace('/', '\\')
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
resolved_path = os.path.abspath(str(file_path or ""))
if not resolved_path or not os.path.isfile(resolved_path):
raise IOError("Audio file not found: " + resolved_path)
if isinstance(positions, (int, float)):
positions = [positions]
elif not isinstance(positions, (list, tuple)):
positions = [0.0]
cleaned_positions = []
for position in positions:
try:
cleaned_positions.append(float(position))
except Exception:
continue
if not cleaned_positions:
cleaned_positions = [0.0]
created_positions = []
for index, position in enumerate(cleaned_positions):
success = False
created_clip = None
for attempt in range(3):
try:
# Find an empty session slot
temp_slot_index = self._find_or_create_empty_clip_slot(track)
clip_slot = track.clip_slots[temp_slot_index]
if clip_slot.has_clip:
clip_slot.delete_clip()
# Load audio into session slot
session_clip = None
if hasattr(clip_slot, "create_audio_clip"):
session_clip = clip_slot.create_audio_clip(resolved_path)
elif hasattr(track, "create_audio_clip"):
# Fallback if LOM uses track for this
session_clip = track.create_audio_clip(resolved_path, float(position))
if session_clip:
self.log_message("Warning: created audio clip directly on track (fallback)")
import time
time.sleep(0.1)
# Duplicate to arrangement
# If session_clip exists and we have the duplicate method
if hasattr(self._song, "duplicate_clip_to_arrangement") and hasattr(clip_slot, "create_audio_clip"):
self.log_message("Duplicating session audio clip to arrangement")
self._song.duplicate_clip_to_arrangement(track, temp_slot_index, float(position))
time.sleep(0.1)
if clip_slot.has_clip:
clip_slot.delete_clip()
clip_persisted = False
for clip in getattr(track, "arrangement_clips", getattr(track, "clips", [])):
if hasattr(clip, "start_time") and abs(float(clip.start_time) - float(position)) < 0.05:
clip_persisted = True
created_clip = clip
break
if clip_persisted:
success = True
break
self.log_message("Warning: Clip at " + str(position) + " not persisted on attempt " + str(attempt+1))
time.sleep(0.1)
except Exception as e:
self.log_message("Warning: Clip creation error at attempt " + str(attempt+1) + ": " + str(e))
try:
if 'clip_slot' in locals() and clip_slot.has_clip:
clip_slot.delete_clip()
except:
pass
time.sleep(0.1)
if not success:
self.log_message("Error: Failed to persist audio clip at " + str(position) + " after 3 attempts")
continue
clip_name = str(name or "").strip()
if clip_name:
if len(cleaned_positions) > 1:
clip_name = clip_name + " " + str(index + 1)
try:
if created_clip is not None and hasattr(created_clip, "name"):
created_clip.name = clip_name
except Exception:
pass
created_positions.append(float(position))
return {
"track_index": int(track_index),
"file_path": resolved_path,
"created_count": len(created_positions),
"positions": created_positions,
"name": str(name or "").strip(),
}
except Exception as e:
self.log_message("Error creating arrangement audio pattern: " + str(e))
raise
def _get_clip_info(self, track_index, clip_index, track_type="track"):
"""Get information about a clip in a track"""
try:
track = self._resolve_track_reference(track_index, track_type)
clip_slots = getattr(track, "clip_slots", [])
if clip_index < 0 or clip_index >= len(clip_slots):
raise IndexError("Clip index out of range")
clip_slot = clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
clip = clip_slot.clip
result = {
"name": clip.name,
"length": clip.length,
"is_playing": clip.is_playing,
"is_recording": clip.is_recording
}
if hasattr(clip, "is_audio_clip"):
result["is_audio_clip"] = clip.is_audio_clip
if hasattr(clip, "is_midi_clip"):
result["is_midi_clip"] = clip.is_midi_clip
if hasattr(clip, "looping"):
result["looping"] = clip.looping
if hasattr(clip, "loop_start"):
result["loop_start"] = clip.loop_start
if hasattr(clip, "loop_end"):
result["loop_end"] = clip.loop_end
if hasattr(clip, "loop_length"):
result["loop_length"] = clip.loop_length
if hasattr(clip, "start_marker"):
result["start_marker"] = clip.start_marker
if hasattr(clip, "end_marker"):
result["end_marker"] = clip.end_marker
return result
except Exception as e:
self.log_message("Error getting clip info: " + str(e))
raise
def _delete_clip(self, track_index, clip_index):
"""Delete a clip from a slot"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
clip_slot.delete_clip()
return {"deleted": True}
except Exception as e:
self.log_message("Error deleting clip: " + str(e))
raise
def _set_clip_loop(self, track_index, clip_index, loop_start, loop_end, loop_length, looping):
"""Set clip loop settings"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
clip = clip_slot.clip
if loop_start is not None and hasattr(clip, "loop_start"):
clip.loop_start = float(loop_start)
if loop_end is not None and hasattr(clip, "loop_end"):
clip.loop_end = float(loop_end)
if loop_length is not None and hasattr(clip, "loop_length") and loop_end is None:
clip.loop_length = float(loop_length)
if looping is not None and hasattr(clip, "looping"):
clip.looping = bool(looping)
return {
"looping": clip.looping if hasattr(clip, "looping") else None,
"loop_start": clip.loop_start if hasattr(clip, "loop_start") else None,
"loop_end": clip.loop_end if hasattr(clip, "loop_end") else None,
"loop_length": clip.loop_length if hasattr(clip, "loop_length") else None
}
except Exception as e:
self.log_message("Error setting clip loop: " + str(e))
raise
def _coerce_live_notes(self, notes):
"""Convert note data to Live's format, accepting 'start' or 'start_time' keys"""
live_notes = []
for note in notes:
pitch = int(note.get("pitch", 60))
start_time = float(note.get("start_time", note.get("start", 0.0)))
duration = float(note.get("duration", 0.25))
velocity = int(note.get("velocity", 100))
mute = bool(note.get("mute", False))
live_notes.append((pitch, start_time, duration, velocity, mute))
return tuple(live_notes)
def _add_notes_to_clip(self, track_index, clip_index, notes):
"""Add MIDI notes to a clip"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
clip = clip_slot.clip
# Convert note data to Live's format (accepts 'start' or 'start_time')
live_notes = self._coerce_live_notes(notes)
# Add the notes
clip.set_notes(live_notes)
result = {
"note_count": len(notes)
}
return result
except Exception as e:
self.log_message("Error adding notes to clip: " + str(e))
raise
def _add_notes_to_arrangement_clip(self, track_index, start_time, notes, track_type="track"):
"""Add MIDI notes to an Arrangement View clip at the specified start time"""
try:
track = self._resolve_track_reference(track_index, track_type)
clip_key = (int(track_index), round(float(start_time), 3))
target_clip = self._recent_arrangement_clips.get(clip_key)
if target_clip is None:
target_clip = self._locate_arrangement_clip(track, start_time, tolerance=0.05)
if target_clip is not None:
self._recent_arrangement_clips[clip_key] = target_clip
if target_clip is None:
raise Exception(f"No clip found at start_time {start_time}")
# Convert note data to Live's format
live_notes = []
for note in notes:
pitch = note.get("pitch", 60)
note_start = note.get("start_time", note.get("start", 0.0))
duration = note.get("duration", 0.25)
velocity = note.get("velocity", 100)
mute = note.get("mute", False)
live_notes.append((pitch, note_start, duration, velocity, mute))
# Add the notes
target_clip.set_notes(tuple(live_notes))
result = {
"note_count": len(notes),
"clip_name": target_clip.name
}
return result
except Exception as e:
self.log_message("Error adding notes to arrangement clip: " + str(e))
raise
def _set_clip_name(self, track_index, clip_index, name):
"""Set the name of a clip"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
clip = clip_slot.clip
clip.name = name
result = {
"name": clip.name
}
return result
except Exception as e:
self.log_message("Error setting clip name: " + str(e))
raise
def _set_tempo(self, tempo):
"""Set the tempo of the session"""
try:
self._song.tempo = tempo
result = {
"tempo": self._song.tempo
}
return result
except Exception as e:
self.log_message("Error setting tempo: " + str(e))
raise
def _set_signature(self, numerator, denominator):
"""Set the time signature"""
try:
self._song.signature_numerator = int(numerator)
self._song.signature_denominator = int(denominator)
return {
"signature_numerator": self._song.signature_numerator,
"signature_denominator": self._song.signature_denominator
}
except Exception as e:
self.log_message("Error setting signature: " + str(e))
raise
def _set_current_song_time(self, time_value):
"""Set the current song time"""
try:
self._song.current_song_time = float(time_value)
return {"current_song_time": self._song.current_song_time}
except Exception as e:
self.log_message("Error setting song time: " + str(e))
raise
def _jump_to(self, time_value):
"""Alias used by the MCP server."""
return self._set_current_song_time(time_value)
def _set_loop(self, enabled):
"""Enable or disable loop"""
try:
self._song.loop = bool(enabled)
return {"loop": self._song.loop}
except Exception as e:
self.log_message("Error setting loop: " + str(e))
raise
def _set_loop_region(self, start, length):
"""Set loop start and length"""
try:
self._song.loop_start = float(start)
self._song.loop_length = float(length)
return {
"loop_start": self._song.loop_start,
"loop_length": self._song.loop_length
}
except Exception as e:
self.log_message("Error setting loop region: " + str(e))
raise
def _loop_selection(self, start, length, enable=None):
"""Alias used by the MCP server for transport loop selection."""
result = self._set_loop_region(start, length)
if enable is not None:
result["loop"] = self._set_loop(enable).get("loop")
return result
def _set_metronome(self, enabled):
"""Enable or disable metronome"""
try:
self._song.metronome = bool(enabled)
return {"metronome": self._song.metronome}
except Exception as e:
self.log_message("Error setting metronome: " + str(e))
raise
def _set_overdub(self, enabled):
"""Enable or disable overdub"""
try:
self._song.overdub = bool(enabled)
return {"overdub": self._song.overdub}
except Exception as e:
self.log_message("Error setting overdub: " + str(e))
raise
def _set_record_mode(self, enabled):
"""Enable or disable record mode"""
try:
if hasattr(self._song, "record_mode"):
self._song.record_mode = bool(enabled)
return {"record_mode": self._song.record_mode}
if hasattr(self._song, "session_record"):
self._song.session_record = bool(enabled)
return {"record_mode": self._song.session_record}
raise RuntimeError("Record mode is not supported")
except Exception as e:
self.log_message("Error setting record mode: " + str(e))
raise
def _duplicate_clip_to_arrangement(self, track_index, clip_index, start_time, track_type="track"):
"""Duplicate a Session View clip to Arrangement View at the specified start time"""
try:
track = self._resolve_track_reference(track_index, track_type)
clip_slots = getattr(track, "clip_slots", [])
if clip_index < 0 or clip_index >= len(clip_slots):
raise IndexError("Clip index out of range")
clip_slot = clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
source_clip = clip_slot.clip
arrangement_clip = None
if hasattr(track, "create_clip"):
try:
arrangement_clip = track.create_clip(start_time, source_clip.length)
if hasattr(source_clip, 'get_notes'):
source_notes = source_clip.get_notes(1, 1)
arrangement_clip.set_notes(source_notes)
except Exception as direct_error:
self.log_message("Direct clip duplication to arrangement failed, using session fallback: " + str(direct_error))
if arrangement_clip is None:
arrangement_clip = self._record_session_clip_to_arrangement(
track_index,
clip_index,
start_time,
float(getattr(source_clip, "length", 4.0) or 4.0),
track_type,
)
# Copy other properties
if hasattr(source_clip, 'name') and source_clip.name:
try:
arrangement_clip.name = source_clip.name
except:
pass
if hasattr(source_clip, 'looping'):
try:
arrangement_clip.looping = source_clip.looping
except:
pass
result = {
"track_index": track_index,
"start_time": start_time,
"length": arrangement_clip.length,
"name": arrangement_clip.name
}
return result
except Exception as e:
self.log_message("Error duplicating clip to arrangement: " + str(e))
raise
def _fire_clip(self, track_index, clip_index):
"""Fire a clip"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
if not clip_slot.has_clip:
raise Exception("No clip in slot")
clip_slot.fire()
result = {
"fired": True
}
return result
except Exception as e:
self.log_message("Error firing clip: " + str(e))
raise
def _stop_clip(self, track_index, clip_index):
"""Stop a clip"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if clip_index < 0 or clip_index >= len(track.clip_slots):
raise IndexError("Clip index out of range")
clip_slot = track.clip_slots[clip_index]
clip_slot.stop()
result = {
"stopped": True
}
return result
except Exception as e:
self.log_message("Error stopping clip: " + str(e))
raise
def _stop_all_clips(self):
"""Stop all clips in the session"""
try:
self._song.stop_all_clips()
return {"stopped": True}
except Exception as e:
self.log_message("Error stopping all clips: " + str(e))
raise
def _get_scenes(self):
"""Get list of scenes"""
try:
scenes = []
for index, scene in enumerate(self._song.scenes):
scenes.append({
"index": index,
"name": scene.name
})
return {"scenes": scenes}
except Exception as e:
self.log_message("Error getting scenes: " + str(e))
raise
def _create_scene(self, index):
"""Create a new scene at index"""
try:
scene_index = len(self._song.scenes) if index == -1 else index
self._song.create_scene(scene_index)
scene = self._song.scenes[scene_index]
return {"index": scene_index, "name": scene.name}
except Exception as e:
self.log_message("Error creating scene: " + str(e))
raise
def _set_scene_name(self, scene_index, name):
"""Set a scene name"""
try:
if scene_index < 0 or scene_index >= len(self._song.scenes):
raise IndexError("Scene index out of range")
scene = self._song.scenes[scene_index]
scene.name = name
return {"name": scene.name}
except Exception as e:
self.log_message("Error setting scene name: " + str(e))
raise
def _set_scene_color(self, scene_index, color):
"""Set scene color when supported by the Live API."""
try:
if scene_index < 0 or scene_index >= len(self._song.scenes):
raise IndexError("Scene index out of range")
scene = self._song.scenes[scene_index]
if hasattr(scene, "color"):
scene.color = int(color)
return {"color": scene.color}
if hasattr(scene, "color_index"):
scene.color_index = int(color)
return {"color": scene.color_index}
return {"color": None, "supported": False}
except Exception as e:
self.log_message("Error setting scene color: " + str(e))
raise
def _fire_scene(self, scene_index):
"""Fire a scene"""
try:
if scene_index < 0 or scene_index >= len(self._song.scenes):
raise IndexError("Scene index out of range")
scene = self._song.scenes[scene_index]
scene.fire()
return {"fired": True}
except Exception as e:
self.log_message("Error firing scene: " + str(e))
raise
def _delete_scene(self, scene_index):
"""Delete a scene"""
try:
if scene_index < 0 or scene_index >= len(self._song.scenes):
raise IndexError("Scene index out of range")
if hasattr(self._song, "delete_scene"):
self._song.delete_scene(scene_index)
else:
raise RuntimeError("Scene deletion is not supported")
return {"deleted": True}
except Exception as e:
self.log_message("Error deleting scene: " + str(e))
raise
def _start_playback(self):
"""Start playing the session"""
try:
self._song.start_playing()
result = {
"playing": self._song.is_playing
}
return result
except Exception as e:
self.log_message("Error starting playback: " + str(e))
raise
def _stop_playback(self):
"""Stop playing the session"""
try:
self._song.stop_playing()
result = {
"playing": self._song.is_playing
}
return result
except Exception as e:
self.log_message("Error stopping playback: " + str(e))
raise
def _show_arrangement_view(self):
"""Best-effort request to focus Arrangement View."""
try:
app = self.application()
view = getattr(app, "view", None)
if view and hasattr(view, "show_view"):
try:
view.show_view("Arranger")
except Exception:
try:
view.show_view("Arrangement")
except Exception:
pass
return {"view": "arrangement"}
except Exception as e:
self.log_message("Error showing arrangement view: " + str(e))
raise
def _get_track_devices(self, track_index):
"""Get devices on a track"""
return self._get_track_devices_for_type(track_index, "track")
def _get_clips_for_type(self, track_index, track_type):
"""Get populated session clips plus arrangement clips for a track-like target."""
try:
track = self._resolve_track_reference(track_index, track_type)
session_clips = []
for slot_index, slot in enumerate(getattr(track, "clip_slots", [])):
if not getattr(slot, "has_clip", False):
continue
clip = slot.clip
clip_info = {
"slot_index": slot_index,
"name": self._safe_getattr(clip, "name", ""),
"length": float(self._safe_getattr(clip, "length", 0.0) or 0.0),
"is_playing": bool(self._safe_getattr(clip, "is_playing", False)),
"is_recording": bool(self._safe_getattr(clip, "is_recording", False)),
}
is_audio_clip = self._safe_getattr(clip, "is_audio_clip")
if is_audio_clip is not None:
clip_info["is_audio_clip"] = bool(is_audio_clip)
is_midi_clip = self._safe_getattr(clip, "is_midi_clip")
if is_midi_clip is not None:
clip_info["is_midi_clip"] = bool(is_midi_clip)
session_clips.append(clip_info)
arrangement_clips = []
arrangement_source = getattr(track, "clips", None)
if arrangement_source is None:
arrangement_source = getattr(track, "arrangement_clips", None)
for clip in arrangement_source or []:
start_time = self._safe_getattr(clip, "start_time", None)
if start_time is None:
continue
arrangement_clips.append({
"name": self._safe_getattr(clip, "name", ""),
"start_time": float(start_time),
"length": float(self._safe_getattr(clip, "length", 0.0) or 0.0),
"is_audio_clip": bool(self._safe_getattr(clip, "is_audio_clip", False)),
"is_midi_clip": bool(self._safe_getattr(clip, "is_midi_clip", False)),
})
arrangement_clips.sort(key=lambda item: (item["start_time"], item["name"]))
return {
"track_index": int(track_index),
"track_type": str(track_type or "track"),
"session_clip_count": len(session_clips),
"session_clips": session_clips,
"arrangement_clip_count": len(arrangement_clips),
"arrangement_clips": arrangement_clips[:512],
}
except Exception as e:
self.log_message("Error getting clips for track: " + str(e))
raise
def _get_track_devices_for_type(self, track_index, track_type):
"""Get devices on a track-like target."""
try:
track = self._resolve_track_reference(track_index, track_type)
devices = []
for device_index, device in enumerate(track.devices):
devices.append({
"index": device_index,
"name": device.name,
"class_name": device.class_name,
"type": self._get_device_type(device),
"parameter_count": len(device.parameters)
})
return {"devices": devices}
except Exception as e:
self.log_message("Error getting track devices: " + str(e))
raise
def _get_master_info(self):
"""Get basic info about the master track."""
master = self._song.master_track
return {
"name": master.name,
"volume": self._safe_mixer_value(master, "volume"),
"panning": self._safe_mixer_value(master, "panning"),
"device_count": len(getattr(master, "devices", []))
}
def _get_device_parameters(self, track_index, device_index, track_type="track"):
"""Get device parameters"""
try:
track = self._resolve_track_reference(track_index, track_type)
if device_index < 0 or device_index >= len(track.devices):
raise IndexError("Device index out of range")
device = track.devices[device_index]
parameters = []
for index, param in enumerate(device.parameters):
try:
is_quantized = bool(param.is_quantized)
except Exception:
is_quantized = False
param_info = {
"index": index,
"name": param.name,
"value": param.value,
"min": param.min,
"max": param.max,
"is_quantized": is_quantized
}
if is_quantized:
try:
param_info["value_items"] = list(param.value_items)
except Exception:
pass
parameters.append(param_info)
return {
"device_name": device.name,
"parameters": parameters
}
except Exception as e:
self.log_message("Error getting device parameters: " + str(e))
raise
def _set_device_parameter(self, track_index, device_index, parameter_index, parameter_name, value, track_type="track"):
"""Set a device parameter by index or name"""
try:
track = self._resolve_track_reference(track_index, track_type)
if device_index < 0 or device_index >= len(track.devices):
raise IndexError("Device index out of range")
device = track.devices[device_index]
param = None
if parameter_index is not None:
if parameter_index < 0 or parameter_index >= len(device.parameters):
raise IndexError("Parameter index out of range")
param = device.parameters[parameter_index]
elif parameter_name:
name_lower = parameter_name.lower()
for candidate in device.parameters:
if candidate.name.lower() == name_lower:
param = candidate
break
if param is None:
raise ValueError("Parameter not found")
if isinstance(value, string_types):
try:
value = float(value)
except Exception:
try:
is_quantized = bool(param.is_quantized)
except Exception:
is_quantized = False
if is_quantized:
try:
items = list(param.value_items)
except Exception:
items = []
if value in items:
value = float(items.index(value))
else:
raise ValueError("Parameter value is not valid")
else:
raise
if isinstance(value, (int, float)):
if value < param.min:
value = param.min
if value > param.max:
value = param.max
param.value = value
return {
"name": param.name,
"value": param.value
}
except Exception as e:
self.log_message("Error setting device parameter: " + str(e))
raise
def _set_device_on(self, track_index, device_index, enabled):
"""Enable or disable a device"""
try:
if track_index < 0 or track_index >= len(self._song.tracks):
raise IndexError("Track index out of range")
track = self._song.tracks[track_index]
if device_index < 0 or device_index >= len(track.devices):
raise IndexError("Device index out of range")
device = track.devices[device_index]
if hasattr(device, "is_enabled"):
device.is_enabled = bool(enabled)
return {"enabled": device.is_enabled}
if hasattr(device, "is_active"):
device.is_active = bool(enabled)
return {"enabled": device.is_active}
for param in device.parameters:
if param.name.lower() in ["device on", "on", "power"]:
param.value = 1.0 if enabled else 0.0
return {"enabled": bool(param.value)}
raise RuntimeError("Device on/off is not supported")
except Exception as e:
self.log_message("Error setting device on: " + str(e))
raise
def _get_browser_categories(self, category_type):
"""Get browser categories (shallow tree)."""
try:
return self.get_browser_tree(category_type, 0)
except Exception as e:
self.log_message("Error getting browser categories: " + str(e))
raise
def _get_browser_items(self, path, item_type):
"""Get browser items at path with optional filtering."""
try:
result = self.get_browser_items_at_path(path)
items = result.get("items", [])
if item_type == "loadable":
items = [item for item in items if item.get("is_loadable")]
elif item_type == "folders":
items = [item for item in items if item.get("is_folder")]
result["items"] = items
return result
except Exception as e:
self.log_message("Error getting browser items: " + str(e))
raise
def _get_browser_item(self, uri, path):
"""Get a browser item by URI or path"""
try:
# Access the application's browser instance instead of creating a new one
app = self.application()
if not app:
raise RuntimeError("Could not access Live application")
result = {
"uri": uri,
"path": path,
"found": False
}
# Try to find by URI first if provided
if uri:
item = self._find_browser_item_by_uri(app.browser, uri)
if item:
result["found"] = True
result["item"] = {
"name": item.name,
"is_folder": item.is_folder,
"is_device": item.is_device,
"is_loadable": item.is_loadable,
"uri": item.uri
}
return result
# If URI not provided or not found, try by path
if path:
# Parse the path and navigate to the specified item
path_parts = path.split("/")
# Determine the root based on the first part
current_item = None
if path_parts[0].lower() == "instruments":
current_item = app.browser.instruments
elif path_parts[0].lower() == "sounds":
current_item = app.browser.sounds
elif path_parts[0].lower() == "drums":
current_item = app.browser.drums
elif path_parts[0].lower() == "audio_effects":
current_item = app.browser.audio_effects
elif path_parts[0].lower() == "midi_effects":
current_item = app.browser.midi_effects
else:
# Default to instruments if not specified
current_item = app.browser.instruments
# Don't skip the first part in this case
path_parts = ["instruments"] + path_parts
# Navigate through the path
for i in range(1, len(path_parts)):
part = path_parts[i]
if not part: # Skip empty parts
continue
found = False
for child in current_item.children:
if child.name.lower() == part.lower():
current_item = child
found = True
break
if not found:
result["error"] = "Path part '{0}' not found".format(part)
return result
# Found the item
result["found"] = True
result["item"] = {
"name": current_item.name,
"is_folder": current_item.is_folder,
"is_device": current_item.is_device,
"is_loadable": current_item.is_loadable,
"uri": current_item.uri
}
return result
except Exception as e:
self.log_message("Error getting browser item: " + str(e))
self.log_message(traceback.format_exc())
raise
def _load_browser_item(self, track_index, item_uri, track_type="track"):
"""Load a browser item onto a track by its URI"""
try:
track = self._resolve_track_reference(track_index, track_type)
# Access the application's browser instance instead of creating a new one
app = self.application()
# Find the browser item by URI
item = self._find_browser_item_by_uri(app.browser, item_uri)
if not item:
raise ValueError("Browser item with URI '{0}' not found".format(item_uri))
# Select the track
self._song.view.selected_track = track
# Load the item
app.browser.load_item(item)
result = {
"loaded": True,
"item_name": item.name,
"track_name": track.name,
"uri": item_uri
}
return result
except Exception as e:
self.log_message("Error loading browser item: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
def _load_instrument_or_effect(self, track_index, uri):
"""Alias for loading a browser item by URI"""
return self._load_browser_item(track_index, uri)
def _load_device(self, track_index, device_name, track_type="track"):
"""Load a device by name onto a track-like target."""
try:
if not device_name:
raise ValueError("Device name is required")
target_track = self._resolve_track_reference(track_index, track_type)
categories = []
if getattr(target_track, "has_midi_input", False):
categories.extend(["instruments", "drums", "sounds", "audio_effects", "midi_effects"])
else:
categories.extend(["audio_effects", "midi_effects", "instruments", "sounds"])
categories.append("all")
for category in categories:
results = self._search_browser_items_internal(device_name, category, 8, 6, True)
if not results:
continue
exact_matches = [
item for item in results
if str(item.get("name", "")).lower() == str(device_name).lower()
]
candidates = exact_matches or results
device_candidates = [item for item in candidates if item.get("is_device")] or candidates
for item in device_candidates:
uri = item.get("uri")
if not uri:
continue
return self._load_browser_item(track_index, uri, track_type)
raise ValueError("No loadable device found for '{0}'".format(device_name))
except Exception as e:
self.log_message("Error loading device: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
def _get_browser_roots(self, category_type):
"""Get browser root items based on category type."""
app = self.application()
if not app or not hasattr(app, "browser"):
raise RuntimeError("Could not access Live browser")
browser = app.browser
roots = []
if category_type in ["all", "instruments"] and hasattr(browser, "instruments"):
roots.append(("Instruments", browser.instruments))
if category_type in ["all", "sounds"] and hasattr(browser, "sounds"):
roots.append(("Sounds", browser.sounds))
if category_type in ["all", "drums"] and hasattr(browser, "drums"):
roots.append(("Drums", browser.drums))
if category_type in ["all", "audio_effects"] and hasattr(browser, "audio_effects"):
roots.append(("Audio Effects", browser.audio_effects))
if category_type in ["all", "midi_effects"] and hasattr(browser, "midi_effects"):
roots.append(("MIDI Effects", browser.midi_effects))
if category_type == "all":
for attr in dir(browser):
if attr.startswith("_"):
continue
if attr in ["instruments", "sounds", "drums", "audio_effects", "midi_effects"]:
continue
try:
item = getattr(browser, attr)
except Exception:
continue
if hasattr(item, "children") or hasattr(item, "name"):
roots.append((attr.replace("_", " ").title(), item))
return roots
def _search_browser_items_internal(self, query, category_type, max_results, max_depth, loadable_only):
"""Search browser items by name."""
results = []
query_lower = query.lower()
def visit(item, path_parts, depth):
if len(results) >= max_results:
return
name = getattr(item, "name", None)
next_path_parts = path_parts
if name and (not path_parts or path_parts[-1] != name):
next_path_parts = path_parts + [name]
if name:
if query_lower in name.lower():
is_loadable = hasattr(item, "is_loadable") and item.is_loadable
if not loadable_only or is_loadable:
results.append({
"name": name,
"path": "/".join(next_path_parts),
"is_folder": hasattr(item, "children") and bool(item.children),
"is_device": hasattr(item, "is_device") and item.is_device,
"is_loadable": is_loadable,
"uri": item.uri if hasattr(item, "uri") else None
})
if depth >= max_depth:
return
if hasattr(item, "children") and item.children:
for child in item.children:
visit(child, next_path_parts, depth + 1)
if len(results) >= max_results:
return
roots = self._get_browser_roots(category_type)
for root_name, root in roots:
visit(root, [root_name], 0)
if len(results) >= max_results:
break
return results
def _search_browser_items(self, query, category_type, max_results, max_depth, loadable_only):
"""Search for browser items by name and return matches."""
try:
results = self._search_browser_items_internal(
query,
category_type,
max_results,
max_depth,
loadable_only
)
return {
"query": query,
"category_type": category_type,
"max_results": max_results,
"items": results
}
except Exception as e:
self.log_message("Error searching browser items: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
def _load_browser_item_by_name(self, track_index, query, category_type, max_depth):
"""Search and load the first matching loadable browser item by name."""
try:
results = self._search_browser_items_internal(
query,
category_type,
1,
max_depth,
True
)
if not results:
raise ValueError("No loadable item found for query '{0}'".format(query))
item = results[0]
if not item.get("uri"):
raise ValueError("Item does not have a URI")
return self._load_browser_item(track_index, item.get("uri"))
except Exception as e:
self.log_message("Error loading browser item by name: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
def _load_browser_item_at_path(self, track_index, path, item_name):
"""Load a browser item from a path, optionally matching by name."""
try:
path_result = self.get_browser_items_at_path(path)
items = path_result.get("items", [])
selected = None
if item_name:
name_lower = item_name.lower()
for item in items:
if item.get("name", "").lower() == name_lower and item.get("is_loadable"):
selected = item
break
else:
for item in items:
if item.get("is_loadable"):
selected = item
break
if not selected or not selected.get("uri"):
raise ValueError("No loadable item found at path")
return self._load_browser_item(track_index, selected.get("uri"))
except Exception as e:
self.log_message("Error loading browser item at path: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
def _find_browser_item_by_uri(self, browser_or_item, uri, max_depth=10, current_depth=0):
"""Find a browser item by its URI"""
try:
# Check if this is the item we're looking for
if hasattr(browser_or_item, 'uri') and browser_or_item.uri == uri:
return browser_or_item
# Stop recursion if we've reached max depth
if current_depth >= max_depth:
return None
# Check if this is a browser with root categories
if hasattr(browser_or_item, 'instruments'):
try:
roots = self._get_browser_roots("all")
except Exception:
roots = []
for _, category in roots:
item = self._find_browser_item_by_uri(category, uri, max_depth, current_depth + 1)
if item:
return item
return None
# Check if this item has children
if hasattr(browser_or_item, 'children') and browser_or_item.children:
for child in browser_or_item.children:
item = self._find_browser_item_by_uri(child, uri, max_depth, current_depth + 1)
if item:
return item
return None
except Exception as e:
self.log_message("Error finding browser item by URI: {0}".format(str(e)))
return None
# Helper methods
def _get_device_type(self, device):
"""Get the type of a device"""
try:
# Simple heuristic - in a real implementation you'd look at the device class
if device.can_have_drum_pads:
return "drum_machine"
elif device.can_have_chains:
return "rack"
elif "instrument" in device.class_display_name.lower():
return "instrument"
elif "audio_effect" in device.class_name.lower():
return "audio_effect"
elif "midi_effect" in device.class_name.lower():
return "midi_effect"
else:
return "unknown"
except:
return "unknown"
def get_browser_tree(self, category_type="all", max_depth=2):
"""
Get a simplified tree of browser categories.
Args:
category_type: Type of categories to get ('all', 'instruments', 'sounds', etc.)
max_depth: Maximum depth to traverse
Returns:
Dictionary with the browser tree structure
"""
try:
# Access the application's browser instance instead of creating a new one
app = self.application()
if not app:
raise RuntimeError("Could not access Live application")
# Check if browser is available
if not hasattr(app, 'browser') or app.browser is None:
raise RuntimeError("Browser is not available in the Live application")
# Log available browser attributes to help diagnose issues
browser_attrs = [attr for attr in dir(app.browser) if not attr.startswith('_')]
self.log_message("Available browser attributes: {0}".format(browser_attrs))
result = {
"type": category_type,
"categories": [],
"available_categories": browser_attrs,
"total_folders": 0
}
folder_count = [0]
# Helper function to process a browser item and its children
def process_item(item, depth=0, path_parts=None):
if not item:
return None
if path_parts is None:
path_parts = []
name = item.name if hasattr(item, 'name') else "Unknown"
node = {
"name": name,
"path": "/".join(path_parts + [name]),
"is_folder": hasattr(item, 'children') and bool(item.children),
"is_device": hasattr(item, 'is_device') and item.is_device,
"is_loadable": hasattr(item, 'is_loadable') and item.is_loadable,
"uri": item.uri if hasattr(item, 'uri') else None,
"children": []
}
if hasattr(item, 'children') and item.children:
if depth >= max_depth:
node["has_more"] = True
return node
for child in item.children:
child_node = process_item(child, depth + 1, path_parts + [name])
if child_node:
node["children"].append(child_node)
folder_count[0] += 1
return node
# Process based on category type and available attributes
if (category_type == "all" or category_type == "instruments") and hasattr(app.browser, 'instruments'):
try:
instruments = process_item(app.browser.instruments, 0, [])
if instruments:
instruments["name"] = "Instruments" # Ensure consistent naming
instruments["path"] = "Instruments"
result["categories"].append(instruments)
except Exception as e:
self.log_message("Error processing instruments: {0}".format(str(e)))
if (category_type == "all" or category_type == "sounds") and hasattr(app.browser, 'sounds'):
try:
sounds = process_item(app.browser.sounds, 0, [])
if sounds:
sounds["name"] = "Sounds" # Ensure consistent naming
sounds["path"] = "Sounds"
result["categories"].append(sounds)
except Exception as e:
self.log_message("Error processing sounds: {0}".format(str(e)))
if (category_type == "all" or category_type == "drums") and hasattr(app.browser, 'drums'):
try:
drums = process_item(app.browser.drums, 0, [])
if drums:
drums["name"] = "Drums" # Ensure consistent naming
drums["path"] = "Drums"
result["categories"].append(drums)
except Exception as e:
self.log_message("Error processing drums: {0}".format(str(e)))
if (category_type == "all" or category_type == "audio_effects") and hasattr(app.browser, 'audio_effects'):
try:
audio_effects = process_item(app.browser.audio_effects, 0, [])
if audio_effects:
audio_effects["name"] = "Audio Effects" # Ensure consistent naming
audio_effects["path"] = "Audio Effects"
result["categories"].append(audio_effects)
except Exception as e:
self.log_message("Error processing audio_effects: {0}".format(str(e)))
if (category_type == "all" or category_type == "midi_effects") and hasattr(app.browser, 'midi_effects'):
try:
midi_effects = process_item(app.browser.midi_effects, 0, [])
if midi_effects:
midi_effects["name"] = "MIDI Effects"
midi_effects["path"] = "MIDI Effects"
result["categories"].append(midi_effects)
except Exception as e:
self.log_message("Error processing midi_effects: {0}".format(str(e)))
# Try to process other potentially available categories
for attr in browser_attrs:
if attr not in ['instruments', 'sounds', 'drums', 'audio_effects', 'midi_effects'] and \
(category_type == "all" or category_type == attr):
try:
item = getattr(app.browser, attr)
if hasattr(item, 'children') or hasattr(item, 'name'):
category = process_item(item, 0, [])
if category:
category["name"] = attr.capitalize()
category["path"] = attr.capitalize()
result["categories"].append(category)
except Exception as e:
self.log_message("Error processing {0}: {1}".format(attr, str(e)))
result["total_folders"] = folder_count[0]
self.log_message("Browser tree generated for {0} with {1} root categories".format(
category_type, len(result['categories'])))
return result
except Exception as e:
self.log_message("Error getting browser tree: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
def get_browser_items_at_path(self, path):
"""
Get browser items at a specific path.
Args:
path: Path in the format "category/folder/subfolder"
where category is one of: instruments, sounds, drums, audio_effects, midi_effects
or any other available browser category
Returns:
Dictionary with items at the specified path
"""
try:
# Access the application's browser instance instead of creating a new one
app = self.application()
if not app:
raise RuntimeError("Could not access Live application")
# Check if browser is available
if not hasattr(app, 'browser') or app.browser is None:
raise RuntimeError("Browser is not available in the Live application")
# Log available browser attributes to help diagnose issues
browser_attrs = [attr for attr in dir(app.browser) if not attr.startswith('_')]
self.log_message("Available browser attributes: {0}".format(browser_attrs))
# Parse the path
path_parts = path.split("/")
if not path_parts:
raise ValueError("Invalid path")
# Determine the root category
root_category = path_parts[0].lower()
current_item = None
# Check standard categories first
if root_category == "instruments" and hasattr(app.browser, 'instruments'):
current_item = app.browser.instruments
elif root_category == "sounds" and hasattr(app.browser, 'sounds'):
current_item = app.browser.sounds
elif root_category == "drums" and hasattr(app.browser, 'drums'):
current_item = app.browser.drums
elif root_category == "audio_effects" and hasattr(app.browser, 'audio_effects'):
current_item = app.browser.audio_effects
elif root_category == "midi_effects" and hasattr(app.browser, 'midi_effects'):
current_item = app.browser.midi_effects
else:
# Try to find the category in other browser attributes
found = False
for attr in browser_attrs:
if attr.lower() == root_category:
try:
current_item = getattr(app.browser, attr)
found = True
break
except Exception as e:
self.log_message("Error accessing browser attribute {0}: {1}".format(attr, str(e)))
if not found:
# If we still haven't found the category, return available categories
return {
"path": path,
"error": "Unknown or unavailable category: {0}".format(root_category),
"available_categories": browser_attrs,
"items": []
}
# Navigate through the path
for i in range(1, len(path_parts)):
part = path_parts[i]
if not part: # Skip empty parts
continue
if not hasattr(current_item, 'children'):
return {
"path": path,
"error": "Item at '{0}' has no children".format('/'.join(path_parts[:i])),
"items": []
}
found = False
for child in current_item.children:
if hasattr(child, 'name') and child.name.lower() == part.lower():
current_item = child
found = True
break
if not found:
return {
"path": path,
"error": "Path part '{0}' not found".format(part),
"items": []
}
# Get items at the current path
items = []
if hasattr(current_item, 'children'):
for child in current_item.children:
item_info = {
"name": child.name if hasattr(child, 'name') else "Unknown",
"is_folder": hasattr(child, 'children') and bool(child.children),
"is_device": hasattr(child, 'is_device') and child.is_device,
"is_loadable": hasattr(child, 'is_loadable') and child.is_loadable,
"uri": child.uri if hasattr(child, 'uri') else None
}
items.append(item_info)
result = {
"path": path,
"name": current_item.name if hasattr(current_item, 'name') else "Unknown",
"uri": current_item.uri if hasattr(current_item, 'uri') else None,
"is_folder": hasattr(current_item, 'children') and bool(current_item.children),
"is_device": hasattr(current_item, 'is_device') and current_item.is_device,
"is_loadable": hasattr(current_item, 'is_loadable') and current_item.is_loadable,
"items": items
}
self.log_message("Retrieved {0} items at path: {1}".format(len(items), path))
return result
except Exception as e:
self.log_message("Error getting browser items at path: {0}".format(str(e)))
self.log_message(traceback.format_exc())
raise
# =========================================================================
# GENERATION COMMANDS
# =========================================================================
def _generate_track(self, params):
"""Generate a track from configuration - safe for Live's main thread"""
try:
self.show_message("MCP: Generating track...")
# 1. Clear existing tracks (if requested)
clear_existing = params.get('clear_existing', False)
if clear_existing:
self._clear_all_tracks()
# 2. Set BPM
bpm = params.get('bpm', 120)
if bpm > 0:
self._song.tempo = float(bpm)
# 3. Create tracks one by one with yields between them
tracks_config = params.get('tracks', [])
created_tracks = []
for idx, track_cfg in enumerate(tracks_config):
track_type = track_cfg.get('type', 'midi')
name = track_cfg.get('name', 'Track ' + str(idx))
# Create track
if track_type == 'midi':
self._song.create_midi_track(idx)
elif track_type == 'audio':
self._song.create_audio_track(idx)
track = self._song.tracks[idx]
track.name = name
# Set color if specified
if 'color' in track_cfg:
track.color = track_cfg['color']
created_tracks.append({"index": idx, "name": name, "type": track_type})
# 4. Create clips and add notes (if specified)
for idx, track_cfg in enumerate(tracks_config):
if 'clip' in track_cfg:
track = self._song.tracks[idx]
clip_cfg = track_cfg['clip']
slot_idx = clip_cfg.get('slot', 0)
length = clip_cfg.get('length', 4.0)
# Ensure enough scenes exist
while len(self._song.scenes) <= slot_idx:
self._song.create_scene(-1)
clip_slot = track.clip_slots[slot_idx]
if not clip_slot.has_clip:
clip_slot.create_clip(length)
# Add notes if specified
if 'notes' in clip_cfg and clip_slot.has_clip:
clip = clip_slot.clip
notes = clip_cfg['notes']
live_notes = self._coerce_live_notes(notes)
if live_notes:
clip.set_notes(live_notes)
clip.name = clip.name + " (" + str(len(notes)) + " notes)"
self.log_message("Added " + str(len(notes)) + " notes to clip")
else:
clip.name = clip.name + " (empty)"
self.log_message("No valid notes to add")
self.show_message("MCP: Track generation complete!")
self.log_message("Generated {0} tracks".format(len(created_tracks)))
return {
"tracks_created": len(created_tracks),
"track_names": [t["name"] for t in created_tracks],
"bpm": bpm
}
except Exception as e:
self.log_message("Error generating track: " + str(e))
self.log_message(traceback.format_exc())
raise
def _generate_track_async(self, params, response_queue):
"""Generate a track incrementally to avoid blocking Live's main thread."""
self.show_message("MCP: Generating track...")
state = {
"params": params,
"response_queue": response_queue,
"clear_existing": params.get("clear_existing", False),
"bpm": float(params.get("bpm", 120) or 120),
"tracks_config": list(params.get("tracks", [])),
"created_tracks": [],
"phase": "clear_existing" if params.get("clear_existing", False) else "tempo",
"track_index": 0,
"clip_index": 0,
}
def fail(exc):
self.log_message("Error generating track: " + str(exc))
self.log_message(traceback.format_exc())
response_queue.put({"status": "error", "message": str(exc)})
def finish():
result = {
"tracks_created": len(state["created_tracks"]),
"track_names": [t["name"] for t in state["created_tracks"]],
"bpm": state["bpm"],
}
self.show_message("MCP: Track generation complete!")
self.log_message("Generated {0} tracks".format(len(state["created_tracks"])))
response_queue.put({"status": "success", "result": result})
def queue_next():
self._enqueue_main_thread_task(step)
def step():
try:
phase = state["phase"]
if phase == "clear_existing":
if len(self._song.tracks) > 0:
self._song.delete_track(len(self._song.tracks) - 1)
queue_next()
return
state["phase"] = "tempo"
queue_next()
return
if phase == "tempo":
if state["bpm"] > 0:
self._song.tempo = state["bpm"]
state["phase"] = "create_tracks"
queue_next()
return
if phase == "create_tracks":
if state["track_index"] < len(state["tracks_config"]):
idx = state["track_index"]
track_cfg = state["tracks_config"][idx]
track_type = track_cfg.get("type", "midi")
name = track_cfg.get("name", "Track " + str(idx))
if track_type == "midi":
self._song.create_midi_track(idx)
elif track_type == "audio":
self._song.create_audio_track(idx)
else:
raise ValueError("Unsupported track type: {0}".format(track_type))
track = self._song.tracks[idx]
track.name = name
if "color" in track_cfg:
track.color = track_cfg["color"]
state["created_tracks"].append({"index": idx, "name": name, "type": track_type})
state["track_index"] += 1
queue_next()
return
state["phase"] = "create_clips"
queue_next()
return
if phase == "create_clips":
if state["clip_index"] < len(state["tracks_config"]):
idx = state["clip_index"]
track_cfg = state["tracks_config"][idx]
state["clip_index"] += 1
if "clip" not in track_cfg:
queue_next()
return
track = self._song.tracks[idx]
clip_cfg = track_cfg["clip"]
slot_idx = clip_cfg.get("slot", 0)
length = clip_cfg.get("length", 4.0)
while len(self._song.scenes) <= slot_idx:
self._song.create_scene(-1)
clip_slot = track.clip_slots[slot_idx]
if not clip_slot.has_clip:
clip_slot.create_clip(length)
if "notes" in clip_cfg and clip_slot.has_clip:
clip = clip_slot.clip
notes = clip_cfg["notes"]
live_notes = self._coerce_live_notes(notes)
if live_notes:
clip.set_notes(live_notes)
clip.name = clip.name + " (" + str(len(notes)) + " notes)"
self.log_message("Added " + str(len(notes)) + " notes to clip")
else:
clip.name = clip.name + " (empty)"
self.log_message("No valid notes to add")
queue_next()
return
finish()
return
raise RuntimeError("Unknown generation phase: {0}".format(phase))
except Exception as exc:
fail(exc)
queue_next()
def _clear_all_tracks(self):
"""Clear all existing tracks"""
try:
count = 0
while len(self._song.tracks) > 0:
self._song.delete_track(len(self._song.tracks) - 1)
count += 1
self.log_message("Cleared {0} tracks".format(count))
return {"tracks_deleted": count}
except Exception as e:
self.log_message("Error clearing tracks: " + str(e))
raise
Reference in New Issue View Git Blame Copy Permalink