325 lines
9.4 KiB
Python
325 lines
9.4 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
|
|
|
|
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 = 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,
|
|
"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)
|
|
start_freq = data.get("startFreq", 20)
|
|
siren_ip = data.get("ip", "192.168.4.200") # Default to actual siren IP
|
|
|
|
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
|
|
pitch = chord[node_index]
|
|
fraction = Fraction(pitch.get("fraction", "1"))
|
|
frequency = fundamental * float(fraction)
|
|
|
|
# Ramp single voice - let sender get start frequency from current position
|
|
siren_sender.ramp_to_chord(frequency, duration_ms, start_freq=None)
|
|
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)
|
|
|
|
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
|
|
|
|
|
|
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=True)
|