""" Main module for MIDI-to-Hue application. Ties together the config, MIDI controller, Hue controller, mapper, animations, and speech-to-text. """ import sys import time import threading import signal from config import ConfigManager from hue_controller import HueController from midi_controller import MidiController, DeviceMappingManager from mapper import MidiToHueMapper from animations import AnimationManager, MidiLedAnimation from speech_to_text import SpeechToText import mido from mcp_server import initialize_hue_controller as mcp_init_hue, main as mcp_main from llm_processor import LLMProcessor import os def main(): """Main application entry point.""" # Load configuration config_manager = ConfigManager() # Create animation manager animation_manager = AnimationManager() # Initialize LLM processor if enabled llm_processor = None if config_manager.get("enable_llm", False): print("Initializing LLM processor...") script_path = os.path.abspath(__file__) llm_processor = LLMProcessor(script_path) # if test flag set from args, process text print("trace: sys.argv", sys.argv) if any("--test-llm" in arg for arg in sys.argv): while not llm_processor.initialized: time.sleep(1) llm_processor.process_text("hello LLM", lambda result: print(f"\nLLM Response:\n{result}\n")) # Initialize speech-to-text if enabled stt = None if config_manager.is_stt_enabled(): print("Initializing speech-to-text...") stt = SpeechToText(config_manager.get_stt_config()) if not stt.initialize(): print("Warning: Failed to initialize speech-to-text.") # Make LLM processor available to STT callback if stt and llm_processor: stt.llm_processor = llm_processor # Initialize Hue controller try: bridge_ip = config_manager.get_bridge_ip() update_interval = config_manager.get_update_interval_sec() hue_controller = HueController(bridge_ip, update_interval) hue_controller.list_lights() # Initialize the MCP server with the same Hue controller # This runs in a separate thread if config_manager.get("enable_mcp", True): # Enable by default mcp_thread = threading.Thread( target=run_mcp_server, args=(bridge_ip, update_interval), daemon=True # Make thread exit when main thread exits ) mcp_thread.start() print("MCP server thread started") except Exception as e: print(f"Failed to initialize Hue controller: {e}") return 1 # Initialize MIDI controller midi_controller = MidiController(config_manager.get_midi_device_index()) midi_controller.list_devices() # Setup device mappings device_mappings = { "traktor_kontrol_s2": { "note_on/1/3": 'left_deck_1', "note_on/1/4": 'left_deck_2', "note_on/1/5": 'left_deck_3', "note_on/1/6": 'left_deck_4', "note_on/1/7": 'left_deck_5', "note_on/1/8": 'left_deck_6', "note_on/1/9": 'left_deck_7', "note_on/1/10": 'left_deck_8', "note_on/3/3": 'right_deck_1', "note_on/3/4": 'right_deck_2', "note_on/3/5": 'right_deck_3', "note_on/3/6": 'right_deck_4', "note_on/3/7": 'right_deck_5', "note_on/3/8": 'right_deck_6', "note_on/3/9": 'right_deck_7', "note_on/3/10": 'right_deck_8', "control_change/5/1": 'left_volume_slider', "control_change/6/1": 'right_volume_slider', } } device_mapper = DeviceMappingManager(device_mappings) device_mapper.set_active_device("traktor_kontrol_s2") # Bind the device mapper to the MIDI controller midi_controller.get_input_name = device_mapper.get_input_name # Set up speech-to-text MIDI handling if enabled if stt: midi_trigger = config_manager.get_stt_midi_trigger() stt_trigger_type = midi_trigger.get("type", "note_on") stt_channel = midi_trigger.get("channel", 1) stt_note = midi_trigger.get("note", 1) print(f"Speech-to-text trigger: {stt_trigger_type}/{stt_channel}/{stt_note}") # Define STT result callback def stt_result_callback(text): print(f"\nSpeech recognition result: {text}\n") # Process the text through LLM if configured if hasattr(stt, 'llm_processor') and stt.llm_processor: print("Processing through LLM...") stt.llm_processor.process_text(text, lambda result: print(f"\nLLM Response:\n{result}\n")) # Set result callback stt.set_callback(stt_result_callback) # Register MIDI handler for speech-to-text def handle_stt_midi(msg): print(f"Received: {msg}") # Check if message matches our trigger if hasattr(msg, 'type') and msg.type == stt_trigger_type and hasattr(msg, 'channel') and msg.channel == stt_channel: if stt_trigger_type == "note_on" and hasattr(msg, 'note') and msg.note == stt_note: # For note_on messages, check velocity to determine press/release if hasattr(msg, 'velocity') and msg.velocity > 0: midi_controller.send_message(mido.Message('note_on', note=msg.note, velocity=127, channel=msg.channel)) print("\nStarting speech recognition (button pressed)...") stt.start_recording() else: midi_controller.send_message(mido.Message('note_on', note=msg.note, velocity=0, channel=msg.channel)) print("\nStopping speech recognition (button released)...") stt.stop_recording() elif stt_trigger_type == "control_change" and hasattr(msg, 'control') and msg.control == stt_note: # For control_change messages, use value threshold if hasattr(msg, 'value') and msg.value > 64: print("\nStarting speech recognition...") stt.start_recording() else: print("\nStopping speech recognition...") stt.stop_recording() # Register the handler midi_controller.register_handler(handle_stt_midi) # Create MIDI-to-Hue mapper with configuration mapper = MidiToHueMapper( hue_controller, midi_controller, config_manager.get_mappings() ) # Setup right deck animation with alternating chase directions print("Setting up right deck animation sequence...") right_deck_animation = animation_manager.create_alternating_chase_animation( name="right_deck_alternating", midi_controller=midi_controller, deck_side="right", interval=0.12, # Animation speed in seconds between steps cycles_per_direction=1 # Number of full cycles in each direction before changing ) # Open MIDI port if not midi_controller.open(): print("Failed to open MIDI port.") return 1 # Start the deck animation print("Starting right deck animation...") animation_manager.start_animation("right_deck_chase") # Handle graceful shutdown def signal_handler(sig, frame): print("\nStopping animations and exiting...") animation_manager.stop_all() midi_controller.close() # Clean up STT resources if stt: stt.cleanup() sys.exit(0) # Register signal handler for Ctrl+C signal.signal(signal.SIGINT, signal_handler) try: # Process MIDI messages in main thread midi_controller.process_messages() except KeyboardInterrupt: print("\nExiting program...") finally: animation_manager.stop_all() midi_controller.close() # Clean up STT resources if stt: stt.cleanup() # Clean up LLM processor resources if llm_processor: llm_processor.cleanup() return 0 def run_mcp_server(bridge_ip, update_interval): """Run the MCP server in a separate thread.""" try: print("Starting MCP server...") mcp_main(bridge_ip, update_interval) except Exception as e: print(f"Error in MCP server: {e}") if __name__ == "__main__": # Check if we should run only the MCP server if len(sys.argv) > 1 and sys.argv[1] == "--mcp-only": # Load config and run only MCP server config_manager = ConfigManager() run_mcp_server(config_manager.get_bridge_ip(), config_manager.get_update_interval_sec()) else: # Run full application sys.exit(main())