Create app.py

app.py

@@ -0,0 +1,194 @@
+from flask import Flask, request, jsonify, render_template
+import requests
+import threading
+import time
+import json
+import os
+
+app = Flask(__name__)
+
+# Global variables to track servo state
+servo_state = {
+    'a_pressed': False,
+    'current_angles': [0, 0, 0],
+    'rotation_active': False,
+    'esp_ip': None
+}
+
+# Thread lock for thread safety
+state_lock = threading.Lock()
+
+def send_servo_command(servo_num, angle):
+    """Send command to specific servo"""
+    try:
+        with state_lock:
+            esp_ip = servo_state['esp_ip']
+        
+        if not esp_ip:
+            return f"Error: No ESP32 IP configured"
+            
+        url = f"http://{esp_ip}/{servo_num}?angle={angle}"
+        response = requests.get(url, timeout=1)
+        return f"Servo {servo_num} → {angle}°"
+    except Exception as e:
+        return f"Error: {e}"
+
+def send_stop_command():
+    """Send emergency stop command"""
+    try:
+        with state_lock:
+            esp_ip = servo_state['esp_ip']
+        
+        if not esp_ip:
+            return "Error: No ESP32 IP configured"
+            
+        url = f"http://{esp_ip}/stop"
+        response = requests.get(url, timeout=1)
+        return "EMERGENCY STOP"
+    except Exception as e:
+        return f"Stop Error: {e}"
+
+def continuous_rotation():
+    """Background thread for continuous rotation"""
+    ROTATION_SPEED = 5
+    ROTATION_DELAY = 0.1  # seconds between updates
+    
+    while True:
+        with state_lock:
+            rotation_active = servo_state['rotation_active']
+            esp_ip = servo_state['esp_ip']
+        
+        if rotation_active and esp_ip:
+            try:
+                # Rotate all servos
+                for i, servo_num in enumerate([1, 2, 3]):
+                    with state_lock:
+                        servo_state['current_angles'][i] += ROTATION_SPEED
+                        if servo_state['current_angles'][i] > 180:
+                            servo_state['current_angles'][i] = 0
+                        
+                        angle = servo_state['current_angles'][i]
+                    
+                    # Send command to servo
+                    send_servo_command(servo_num, angle)
+                
+                time.sleep(ROTATION_DELAY)
+            except Exception as e:
+                print(f"Rotation error: {e}")
+                time.sleep(0.5)
+        else:
+            time.sleep(0.1)
+
+@app.route('/')
+def index():
+    """Serve the main control page"""
+    return render_template('index.html')
+
+@app.route('/set_ip', methods=['POST'])
+def set_ip():
+    """Set the ESP32 IP address"""
+    data = request.get_json()
+    esp_ip = data.get('ip', '').strip()
+    
+    if esp_ip:
+        with state_lock:
+            servo_state['esp_ip'] = esp_ip
+        return jsonify({'status': 'success', 'message': f'ESP32 IP set to: {esp_ip}'})
+    else:
+        return jsonify({'status': 'error', 'message': 'No IP provided'})
+
+@app.route('/start_rotation', methods=['POST'])
+def start_rotation():
+    """Start continuous rotation (equivalent to holding 'A')"""
+    with state_lock:
+        servo_state['rotation_active'] = True
+        servo_state['a_pressed'] = True
+    
+    return jsonify({
+        'status': 'success', 
+        'message': 'Continuous rotation started',
+        'angles': servo_state['current_angles']
+    })
+
+@app.route('/stop_rotation', methods=['POST'])
+def stop_rotation():
+    """Stop rotation and send emergency stop (equivalent to releasing 'A')"""
+    with state_lock:
+        servo_state['rotation_active'] = False
+        servo_state['a_pressed'] = False
+    
+    stop_msg = send_stop_command()
+    
+    return jsonify({
+        'status': 'success', 
+        'message': 'Rotation stopped',
+        'stop_message': stop_msg,
+        'angles': servo_state['current_angles']
+    })
+
+@app.route('/reset_servos', methods=['POST'])
+def reset_servos():
+    """Reset all servos to 0 degrees (equivalent to pressing 'R')"""
+    messages = []
+    with state_lock:
+        for i in range(3):
+            servo_state['current_angles'][i] = 0
+            msg = send_servo_command(i+1, 0)
+            messages.append(msg)
+    
+    return jsonify({
+        'status': 'success', 
+        'message': 'Servos reset to 0°',
+        'details': messages,
+        'angles': servo_state['current_angles']
+    })
+
+@app.route('/set_angle', methods=['POST'])
+def set_angle():
+    """Set specific angle for a servo"""
+    data = request.get_json()
+    servo_num = data.get('servo')
+    angle = data.get('angle')
+    
+    try:
+        servo_num = int(servo_num)
+        angle = int(angle)
+        
+        if servo_num < 1 or servo_num > 3:
+            return jsonify({'status': 'error', 'message': 'Servo number must be 1, 2, or 3'})
+        
+        if angle < 0 or angle > 180:
+            return jsonify({'status': 'error', 'message': 'Angle must be between 0 and 180'})
+        
+        with state_lock:
+            servo_state['current_angles'][servo_num-1] = angle
+        
+        msg = send_servo_command(servo_num, angle)
+        
+        return jsonify({
+            'status': 'success', 
+            'message': f'Servo {servo_num} set to {angle}°',
+            'details': msg,
+            'angles': servo_state['current_angles']
+        })
+    except ValueError:
+        return jsonify({'status': 'error', 'message': 'Invalid servo number or angle'})
+
+@app.route('/get_status', methods=['GET'])
+def get_status():
+    """Get current servo status"""
+    with state_lock:
+        return jsonify({
+            'a_pressed': servo_state['a_pressed'],
+            'rotation_active': servo_state['rotation_active'],
+            'current_angles': servo_state['current_angles'],
+            'esp_ip': servo_state['esp_ip']
+        })
+
+if __name__ == '__main__':
+    # Start the continuous rotation thread
+    rotation_thread = threading.Thread(target=continuous_rotation, daemon=True)
+    rotation_thread.start()
+    
+    # Run Flask app
+    app.run(host='0.0.0.0', port=5000, debug=False)