Michael Winter
b8f50a4563
- Y-drag creates ghost nodes at quantized ratios (2/1, 3/2, 5/4, etc.) - Ghost nodes store their own frequency (fundamental * new fraction) - Preview ghost shows frequency in real-time while dragging - Final ghost created on release (if dragged >100 cents from original) - Server API accepts frequency directly for ghost nodes - Fix frequency display: fundamental * fraction (not cents-adjusted) - Fix negative snap ratios to use correct reciprocals - Add colored circle on click (both ramp and non-ramp modes) - Ghost nodes: no border initially, opacity 0.7, border shows on click - Chord label clicks exclude ghost nodes - Remove octaveOffset, use cents or direct frequency instead
548 lines
17 KiB
Python
548 lines
17 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Flask server for Path Navigator with OSC support.
|
|
"""
|
|
|
|
from flask import Flask, jsonify, request, send_from_directory
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Add parent directory to path for imports
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from src.pitch import Pitch
|
|
from src.osc_sender import OSCSender
|
|
from fractions import Fraction
|
|
|
|
app = Flask(__name__)
|
|
|
|
# Path to output directory
|
|
DATA_DIR = Path(__file__).parent.parent / "output"
|
|
DEFAULT_FILE = "output_chords.json" # default file
|
|
|
|
# State
|
|
current_index = 0
|
|
chords = []
|
|
|
|
# Cache persistent OSCSenders per IP for the siren
|
|
_sender_cache = {} # ip -> OSCSender
|
|
|
|
|
|
def get_siren_sender(ip):
|
|
"""Get or create a persistent OSCSender for the given siren IP."""
|
|
if ip not in _sender_cache:
|
|
_sender_cache[ip] = OSCSender(ip=ip, port=54001, fundamental=fundamental)
|
|
return _sender_cache[ip]
|
|
|
|
|
|
# OSC settings
|
|
fundamental = 110.0
|
|
osc_sender = OSCSender(ip="127.0.0.1", port=57120, fundamental=fundamental)
|
|
|
|
|
|
def load_chords(filepath=None):
|
|
global chords, current_index
|
|
if filepath is None:
|
|
filepath = DATA_DIR / DEFAULT_FILE
|
|
if filepath.exists():
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
chords = data.get("chords", [])
|
|
current_index = 0
|
|
else:
|
|
chords = []
|
|
|
|
|
|
load_chords()
|
|
|
|
|
|
def parse_fraction(frac_str):
|
|
"""Parse a fraction string to float."""
|
|
if isinstance(frac_str, (int, float)):
|
|
return float(frac_str)
|
|
if "/" in frac_str:
|
|
num, den = frac_str.split("/")
|
|
return int(num) / int(den)
|
|
return float(frac_str)
|
|
|
|
|
|
def calculate_cents(fraction_str):
|
|
"""Calculate cents from fraction string."""
|
|
import math
|
|
|
|
fr = parse_fraction(fraction_str)
|
|
if fr <= 0:
|
|
return 0
|
|
return 1200 * math.log2(fr)
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return send_from_directory(".", "path_navigator.html")
|
|
|
|
|
|
@app.route("/<path:filename>")
|
|
def serve_static(filename):
|
|
return send_from_directory(".", filename)
|
|
|
|
|
|
@app.route("/api/chords")
|
|
def get_chords():
|
|
return jsonify({"chords": chords, "total": len(chords)})
|
|
|
|
|
|
@app.route("/api/set-fundamental", methods=["POST"])
|
|
def set_fundamental():
|
|
"""Set the fundamental frequency for OSC playback."""
|
|
global fundamental, osc_sender
|
|
data = request.json
|
|
fundamental = data.get("fundamental", 110.0)
|
|
osc_sender = OSCSender(ip="127.0.0.1", port=57120, fundamental=fundamental)
|
|
return jsonify({"fundamental": fundamental})
|
|
|
|
|
|
@app.route("/api/play-freq", methods=["POST"])
|
|
def play_freq():
|
|
"""Play a single frequency for a node."""
|
|
data = request.json
|
|
chord_index = data.get("chordIndex")
|
|
node_index = data.get("nodeIndex")
|
|
octave = data.get("octave", 0)
|
|
|
|
if chord_index < 0 or chord_index >= len(chords):
|
|
return jsonify({"error": "Invalid chord index"}), 400
|
|
|
|
chord = chords[chord_index]
|
|
if node_index < 0 or node_index >= len(chord):
|
|
return jsonify({"error": "Invalid node index"}), 400
|
|
|
|
pitch = chord[node_index]
|
|
fraction = Fraction(pitch.get("fraction", "1"))
|
|
frequency = fundamental * float(fraction) * (2**octave)
|
|
voice = node_index + 1 # 1-indexed
|
|
|
|
osc_sender.send_single(frequency, voice)
|
|
|
|
return jsonify(
|
|
{
|
|
"frequency": frequency,
|
|
"voice": voice,
|
|
"fundamental": fundamental,
|
|
"fraction": pitch.get("fraction", "1"),
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/api/play-siren", methods=["POST"])
|
|
def play_siren():
|
|
"""Play a single frequency for a node on the siren."""
|
|
data = request.json
|
|
|
|
chord_index = data.get("chordIndex")
|
|
node_index = data.get("nodeIndex")
|
|
siren_ip = data.get("ip", "192.168.4.200") # Default to actual siren IP
|
|
siren_number = data.get("sirenNumber") # optional 1-4
|
|
frequency_input = data.get("frequency") # direct frequency for ghost nodes
|
|
|
|
# If frequency provided directly (ghost node), use it
|
|
if frequency_input is not None:
|
|
frequency = float(frequency_input)
|
|
voice = siren_number if siren_number else 1
|
|
else:
|
|
# Original node: calculate from chordIndex/nodeIndex
|
|
if chord_index is None or node_index is None:
|
|
return jsonify({"error": "Missing chordIndex/nodeIndex"}), 400
|
|
|
|
if chord_index < 0 or chord_index >= len(chords):
|
|
return jsonify({"error": "Invalid chord index"}), 400
|
|
|
|
chord = chords[chord_index]
|
|
if node_index < 0 or node_index >= len(chord):
|
|
return jsonify({"error": "Invalid node index"}), 400
|
|
|
|
pitch = chord[node_index]
|
|
fraction = Fraction(pitch.get("fraction", "1"))
|
|
frequency = fundamental * float(fraction)
|
|
voice = siren_number if siren_number else node_index + 1 # 1-indexed
|
|
|
|
# Send to siren using cached sender
|
|
siren_sender = get_siren_sender(siren_ip)
|
|
siren_sender.set_chords(chords) # Set chords to ensure proper voice count
|
|
siren_sender.send_single(frequency, voice)
|
|
|
|
return jsonify(
|
|
{
|
|
"frequency": frequency,
|
|
"voice": voice,
|
|
"siren": siren_number,
|
|
"destination": "siren",
|
|
"ip": siren_ip,
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/api/kill-siren", methods=["POST"])
|
|
def kill_siren():
|
|
"""Send kill message to siren (soft or hard)."""
|
|
data = request.json
|
|
soft = data.get("soft", True) # default to soft kill
|
|
siren_ip = data.get("ip", "192.168.4.200") # Default to actual siren IP
|
|
|
|
kill_freq = 20.0 if soft else 0.0
|
|
kill_type = "soft" if soft else "hard"
|
|
|
|
# Cancel any running ramp (for this IP)
|
|
siren_sender = get_siren_sender(siren_ip)
|
|
for event in siren_sender._ramp_events.values():
|
|
event.set()
|
|
|
|
# Send kill to all voices using cached sender
|
|
siren_sender.set_chords(chords) # Set chords to get correct num_voices
|
|
siren_sender.send_kill(soft=soft)
|
|
|
|
return jsonify(
|
|
{
|
|
"kill_freq": kill_freq,
|
|
"kill_type": kill_type,
|
|
"voices": list(range(1, siren_sender.num_voices + 1)),
|
|
"ip": siren_ip,
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/api/ramp-to-chord", methods=["POST"])
|
|
def ramp_to_chord():
|
|
"""Ramp to a chord or single voice on the siren."""
|
|
data = request.json
|
|
|
|
chord_index = data.get("chordIndex")
|
|
node_index = data.get("nodeIndex") # Optional - if present, ramp single voice
|
|
duration_ms = data.get("duration", 3000)
|
|
exponent = data.get("exponent", 1.0)
|
|
siren_ip = data.get("ip", "192.168.4.200") # Default to actual siren IP
|
|
siren_number = data.get("sirenNumber") # optional 1-4
|
|
frequency_input = data.get("frequency") # direct frequency for ghost nodes
|
|
|
|
if chord_index is None:
|
|
return jsonify({"error": "Missing chordIndex"}), 400
|
|
|
|
if chord_index < 0 or chord_index >= len(chords):
|
|
return jsonify({"error": "Invalid chord index"}), 400
|
|
|
|
chord = chords[chord_index]
|
|
|
|
# Use cached sender
|
|
siren_sender = get_siren_sender(siren_ip)
|
|
siren_sender.set_chords(chords)
|
|
|
|
if node_index is not None:
|
|
# Ramp single voice
|
|
if node_index < 0 or node_index >= len(chord):
|
|
return jsonify({"error": "Invalid node index"}), 400
|
|
|
|
# If frequency provided directly (ghost node), use it
|
|
if frequency_input is not None:
|
|
frequency = float(frequency_input)
|
|
voice = siren_number if siren_number else 1
|
|
else:
|
|
pitch = chord[node_index]
|
|
fraction = Fraction(pitch.get("fraction", "1"))
|
|
frequency = fundamental * float(fraction)
|
|
voice = siren_number if siren_number else (node_index + 1) # 1-indexed
|
|
|
|
# Ramp single voice - let sender get start frequency from current position
|
|
siren_sender.ramp_to_pitch(
|
|
frequency, voice, duration_ms, start_freq=None, exponent=exponent
|
|
)
|
|
else:
|
|
# Ramp entire chord - let sender get start frequencies from current positions
|
|
frequencies = [
|
|
fundamental * float(Fraction(p.get("fraction", "1"))) for p in chord
|
|
]
|
|
|
|
siren_sender.ramp_to_chord(
|
|
frequencies, duration_ms, start_freq=None, exponent=exponent
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"status": "ramping",
|
|
"chordIndex": chord_index,
|
|
"voiceIndex": node_index,
|
|
"duration": duration_ms,
|
|
}
|
|
)
|
|
|
|
|
|
@app.route("/api/load-file", methods=["POST"])
|
|
def load_file():
|
|
global current_index, chords
|
|
data = request.json
|
|
filepath = data.get("filepath")
|
|
|
|
if not filepath:
|
|
return jsonify({"error": "No filepath provided"}), 400
|
|
|
|
try:
|
|
with open(filepath) as f:
|
|
chords_data = json.load(f)
|
|
chords = chords_data.get("chords", [])
|
|
current_index = 0
|
|
return jsonify({"loaded": len(chords), "filepath": filepath})
|
|
except FileNotFoundError:
|
|
return jsonify({"error": f"File not found: {filepath}"}), 404
|
|
except json.JSONDecodeError:
|
|
return jsonify({"error": f"Invalid JSON in: {filepath}"}), 400
|
|
|
|
|
|
@app.route("/api/parse-fraction", methods=["POST"])
|
|
def parse_fraction_api():
|
|
"""Parse a fraction string to float."""
|
|
data = request.json
|
|
fraction = data.get("fraction", "1")
|
|
try:
|
|
value = parse_fraction(fraction)
|
|
return jsonify({"fraction": fraction, "value": value})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.route("/api/calculate-cents", methods=["POST"])
|
|
def calculate_cents_api():
|
|
"""Calculate cents from fraction string."""
|
|
data = request.json
|
|
fraction = data.get("fraction", "1")
|
|
try:
|
|
cents = calculate_cents(fraction)
|
|
return jsonify({"fraction": fraction, "cents": cents})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.route("/api/batch-calculate-cents", methods=["POST"])
|
|
def batch_calculate_cents_api():
|
|
"""Calculate cents for multiple fractions at once."""
|
|
data = request.json
|
|
fractions = data.get("fractions", [])
|
|
try:
|
|
results = []
|
|
for fraction in fractions:
|
|
cents = calculate_cents(fraction)
|
|
results.append({"fraction": fraction, "cents": cents})
|
|
return jsonify({"results": results})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.route("/generate")
|
|
def generate_page():
|
|
"""Render the generator page."""
|
|
return send_from_directory(app.root_path, "generate.html")
|
|
|
|
|
|
@app.route("/generator_settings.json")
|
|
def default_settings():
|
|
"""Serve default generator settings from output directory."""
|
|
settings_path = DATA_DIR / "generator_settings.json"
|
|
if settings_path.exists():
|
|
return send_from_directory(DATA_DIR, "generator_settings.json")
|
|
else:
|
|
return jsonify({"error": "No default settings file"}), 404
|
|
|
|
|
|
@app.route("/api/transcribe", methods=["POST"])
|
|
def run_transcribe():
|
|
"""Run the transcriber to create LilyPond output."""
|
|
data = request.json
|
|
transcribe_name = data.get("name", "compact_sets_transcription")
|
|
fundamental = data.get("fundamental", 55)
|
|
|
|
try:
|
|
import subprocess
|
|
import os
|
|
import sys
|
|
import json
|
|
|
|
input_file = DATA_DIR / "output_chords.json"
|
|
|
|
if not input_file.exists():
|
|
return jsonify({"error": "No chords to transcribe. Generate first."}), 400
|
|
|
|
# Load chords to get count
|
|
with open(input_file) as f:
|
|
chords_data = json.load(f)
|
|
chord_count = len(chords_data.get("chords", chords_data))
|
|
|
|
# Create lilypond directory if needed
|
|
lilypond_dir = DATA_DIR.parent / "lilypond"
|
|
os.makedirs(lilypond_dir, exist_ok=True)
|
|
|
|
# Run the transcriber
|
|
args = [
|
|
sys.executable,
|
|
"-m",
|
|
"src.io",
|
|
"--transcribe",
|
|
str(input_file),
|
|
transcribe_name,
|
|
"--fundamental",
|
|
str(fundamental),
|
|
]
|
|
|
|
result = subprocess.run(
|
|
args,
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
return jsonify({"error": result.stderr}), 400
|
|
|
|
output_file = lilypond_dir / f"{transcribe_name}.pdf"
|
|
|
|
return jsonify(
|
|
{
|
|
"status": "success",
|
|
"chordCount": chord_count,
|
|
"outputFile": str(output_file),
|
|
"output": result.stdout,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.route("/api/generate", methods=["POST"])
|
|
def run_generator():
|
|
"""Run the path generator with provided options."""
|
|
data = request.json
|
|
|
|
try:
|
|
import subprocess
|
|
import os
|
|
import sys
|
|
|
|
# Build args list
|
|
args = [
|
|
sys.executable,
|
|
"-m",
|
|
"src.io",
|
|
"--dims",
|
|
str(data.get("dims", 7)),
|
|
"--chord-size",
|
|
str(data.get("chordSize", 3)),
|
|
"--max-path",
|
|
str(data.get("maxPath", 50)),
|
|
"--symdiff-min",
|
|
str(data.get("symdiffMin", 2)),
|
|
"--symdiff-max",
|
|
str(data.get("symdiffMax", 2)),
|
|
"--melodic-min",
|
|
str(data.get("melodicMin", 0)),
|
|
"--melodic-max",
|
|
str(data.get("melodicMax", 500)),
|
|
"--target-register",
|
|
str(data.get("targetRegister", 0)),
|
|
"--target-register-power",
|
|
str(data.get("targetRegisterPower", 1.0)),
|
|
"--target-register-oscillations",
|
|
str(data.get("targetRegisterOscillations", 0)),
|
|
"--target-register-amplitude",
|
|
str(data.get("targetRegisterAmplitude", 0.25)),
|
|
"--weight-melodic",
|
|
str(data.get("weightMelodic", 1)),
|
|
"--weight-contrary-motion",
|
|
str(data.get("weightContraryMotion", 0)),
|
|
"--weight-dca-hamiltonian",
|
|
str(data.get("weightDcaHamiltonian", 1)),
|
|
"--weight-dca-voice-movement",
|
|
str(data.get("weightDcaVoiceMovement", 1)),
|
|
"--weight-rgr-voice-movement",
|
|
str(data.get("weightRgrVoiceMovement", 0)),
|
|
"--rgr-voice-movement-threshold",
|
|
str(data.get("rgrVoiceMovementThreshold", 5)),
|
|
"--weight-harmonic-compactness",
|
|
str(data.get("weightHarmonicCompactness", 0)),
|
|
"--weight-target-register",
|
|
str(data.get("weightTargetRegister", 1)),
|
|
"--output-dir",
|
|
data.get("outputDir", "output"),
|
|
"--fundamental",
|
|
str(data.get("fundamental", 55)),
|
|
"--cache-dir",
|
|
data.get("cacheDir", "cache"),
|
|
]
|
|
|
|
seed = data.get("seed")
|
|
# If no seed provided, generate a random one
|
|
if seed is None:
|
|
import random
|
|
|
|
seed = random.randint(1, 999999)
|
|
args.extend(["--seed", str(seed)])
|
|
if data.get("allowVoiceCrossing", False):
|
|
args.append("--allow-voice-crossing")
|
|
if data.get("disableDirectTuning", False):
|
|
args.append("--disable-direct-tuning")
|
|
if data.get("uniformSymdiff", False):
|
|
args.append("--uniform-symdiff")
|
|
if data.get("rebuildCache", False):
|
|
args.append("--rebuild-cache")
|
|
if data.get("noCache", False):
|
|
args.append("--no-cache")
|
|
|
|
# Create output directory
|
|
os.makedirs(data.get("outputDir", "output"), exist_ok=True)
|
|
|
|
# Run the generator
|
|
result = subprocess.run(
|
|
args,
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
return jsonify({"error": result.stderr}), 400
|
|
|
|
# Try to count generated chords from output
|
|
total_chords = 0
|
|
for line in result.stdout.split("\n"):
|
|
if "Path length:" in line:
|
|
try:
|
|
total_chords = int(line.split(":")[-1].strip())
|
|
except:
|
|
pass
|
|
|
|
return jsonify(
|
|
{
|
|
"totalChords": total_chords,
|
|
"status": "success",
|
|
"seed": seed,
|
|
"output": result.stdout,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting Path Navigator server...")
|
|
filepath = DATA_DIR / DEFAULT_FILE
|
|
print(f"Loading chords from: {filepath}")
|
|
load_chords()
|
|
print(f"Loaded {len(chords)} chords")
|
|
app.run(host="0.0.0.0", port=8080, debug=False)
|