diff --git a/src/io.py b/src/io.py index 1407732..af59d16 100644 --- a/src/io.py +++ b/src/io.py @@ -354,6 +354,29 @@ def main(): action="store_true", help="Show analysis statistics after generation", ) + parser.add_argument( + "--osc-enable", + action="store_true", + help="Enable OSC output for real-time playback", + ) + parser.add_argument( + "--osc-ip", + type=str, + default="192.168.4.200", + help="OSC destination IP (default: 192.168.4.200)", + ) + parser.add_argument( + "--osc-port", + type=int, + default=54001, + help="OSC destination port (default: 54001)", + ) + parser.add_argument( + "--fundamental", + type=float, + default=100, + help="Fundamental frequency in Hz for OSC output (default: 100)", + ) args = parser.parse_args() # Select dims @@ -509,6 +532,19 @@ def main(): print() print(format_analysis(metrics)) + # OSC playback if enabled + if args.osc_enable: + from .osc_sender import OSCSender + + chords_file = os.path.join(args.output_dir, "output_chords.json") + sender = OSCSender( + ip=args.osc_ip, port=args.osc_port, fundamental=args.fundamental + ) + sender.load_chords(chords_file) + print(f"\nOSC enabled - sending to {args.osc_ip}:{args.osc_port}") + print(f"Fundamental: {args.fundamental} Hz") + sender.play() + if __name__ == "__main__": main() diff --git a/src/osc_sender.py b/src/osc_sender.py new file mode 100644 index 0000000..721f155 --- /dev/null +++ b/src/osc_sender.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python +""" +OSC Sender - send chord frequencies via OSC for real-time playback. +""" + +import json +import sys +from pathlib import Path + +try: + from pythonosc import osc_message_builder + from pythonosc.udp_client import SimpleUDPClient +except ImportError: + print("Error: python-osc not installed. Run: pip install python-osc") + sys.exit(1) + + +class OSCSender: + """Send chord frequencies via OSC.""" + + def __init__(self, ip="192.168.4.200", port=54001, fundamental=100): + self.ip = ip + self.port = port + self.fundamental = fundamental + self.client = SimpleUDPClient(ip, port) + self.chords = [] + self.current_index = 0 + + def load_chords(self, chords_file): + """Load chords from output_chords.json.""" + with open(chords_file) as f: + self.chords = json.load(f) + print(f"Loaded {len(self.chords)} chords") + + def send_chord(self, index): + """Send frequencies for a specific chord index.""" + if index < 0 or index >= len(self.chords): + return + + chord = self.chords[index] + + for voice_idx, pitch in enumerate(chord): + from fractions import Fraction + + frac = Fraction(pitch["fraction"]) + freq = self.fundamental * float(frac) + addr = f"/freq" + msg = osc_message_builder.OscMessageBuilder(addr=addr) + msg.add_arg(voice_idx + 1) # Voice number (1-indexed) + msg.add_arg(freq) # Frequency + self.client.send(msg.build()) + + def display_chord(self, index): + """Display current chord info.""" + if index < 0 or index >= len(self.chords): + return + + chord = self.chords[index] + + print(f"\r{'=' * 50}") + print(f"Chord {index + 1}/{len(self.chords)}") + print(f"{'=' * 50}") + + print("Frequencies (Hz):") + from fractions import Fraction + + for voice_idx, pitch in enumerate(chord): + frac = Fraction(pitch["fraction"]) + freq = self.fundamental * float(frac) + print(f" Voice {voice_idx + 1}: {freq:.1f}") + + print("\nHS Arrays:") + for voice_idx, pitch in enumerate(chord): + print(f" Voice {voice_idx + 1}: {pitch['hs_array']}") + + print(f"\nFundamental: {self.fundamental} Hz") + print(f"Destination: {self.ip}:{self.port}") + print(f"{'=' * 50}") + print("← Previous | Next → | Ctrl+C to quit") + + def play(self): + """Interactive playback with keyboard control.""" + if not self.chords: + print("No chords loaded!") + return + + try: + import threading + import tty + import termios + import os + + # Send and display first chord + self.send_chord(self.current_index) + self.display_chord(self.current_index) + + def get_key(): + """Get a single keypress.""" + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + try: + tty.setraw(sys.stdin.fileno()) + ch = sys.stdin.read(1) + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + return ch + + print("\nUse arrow keys to navigate. Press Ctrl+C to quit.") + + while True: + key = get_key() + + # Left arrow + if key == "\x1b": # Escape sequence start + next1 = get_key() + if next1 == "[": + next2 = get_key() + if next2 == "D": # Left arrow + if self.current_index > 0: + self.current_index -= 1 + self.send_chord(self.current_index) + self.display_chord(self.current_index) + elif next2 == "C": # Right arrow + if self.current_index < len(self.chords) - 1: + self.current_index += 1 + self.send_chord(self.current_index) + self.display_chord(self.current_index) + + # Alternative: use < and > keys + elif key == "," or key == "<": + if self.current_index > 0: + self.current_index -= 1 + self.send_chord(self.current_index) + self.display_chord(self.current_index) + elif key == "." or key == ">": + if self.current_index < len(self.chords) - 1: + self.current_index += 1 + self.send_chord(self.current_index) + self.display_chord(self.current_index) + + except KeyboardInterrupt: + print("\n\nPlayback stopped.") + + def send_all(self): + """Send all chords in sequence (for testing).""" + for i in range(len(self.chords)): + self.send_chord(i) + print(f"Sent chord {i + 1}/{len(self.chords)}") + print("All chords sent!") + + +def load_chords_from_output(output_dir="output"): + """Load chords from the standard output directory.""" + chords_file = Path(output_dir) / "output_chords.json" + if not chords_file.exists(): + raise FileNotFoundError(f"Chords file not found: {chords_file}") + + with open(chords_file) as f: + return json.load(f) + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="OSC Sender for Compact Sets") + parser.add_argument("--ip", default="192.168.4.200", help="Destination IP") + parser.add_argument("--port", type=int, default=54001, help="Destination port") + parser.add_argument( + "--fundamental", type=float, default=100, help="Fundamental frequency in Hz" + ) + parser.add_argument("--output-dir", default="output", help="Output directory") + parser.add_argument( + "--send-all", action="store_true", help="Send all chords in sequence" + ) + + args = parser.parse_args() + + # Load chords + chords_file = Path(args.output_dir) / "output_chords.json" + if not chords_file.exists(): + print(f"Error: {chords_file} not found!") + print("Run compact_sets.py first to generate chords.") + sys.exit(1) + + # Create sender + sender = OSCSender(ip=args.ip, port=args.port, fundamental=args.fundamental) + sender.load_chords(chords_file) + + if args.send_all: + sender.send_all() + else: + sender.play()