reterminal-dm4/archive/chromium-setup-legacy/flask_buzzer_app.py

#!/usr/bin/env python3
"""
Flask Buzzer Control Application for reTerminal DM4
Provides REST API for controlling the built-in buzzer
"""

from flask import Flask, jsonify, request
import subprocess
import time
import threading

app = Flask(__name__)
BUZZER_PATH = '/sys/class/leds/usr-buzzer/brightness'

class BuzzerController:
    """Buzzer controller class with pattern support"""
    
    def __init__(self):
        self.is_playing = False
        self.play_thread = None
    
    def _write_buzzer(self, value):
        """Internal method to write to buzzer"""
        try:
            subprocess.run(['sudo', 'tee', BUZZER_PATH], 
                          input=str(value), text=True, 
                          stdout=subprocess.DEVNULL, 
                          stderr=subprocess.DEVNULL,
                          check=True)
            return True
        except subprocess.CalledProcessError:
            return False
    
    def on(self):
        """Turn buzzer ON"""
        return self._write_buzzer(1)
    
    def off(self):
        """Turn buzzer OFF"""
        return self._write_buzzer(0)
    
    def beep(self, duration=0.2):
        """Play a single beep"""
        self.on()
        time.sleep(duration)
        self.off()
    
    def play_pattern(self, pattern):
        """
        Play a pattern
        pattern: list of tuples [(duration_on, duration_off), ...]
        Example: [(0.1, 0.1), (0.1, 0.1), (0.1, 0.3)] = two short beeps, pause, one beep
        """
        if self.is_playing:
            return False
        
        def _play():
            self.is_playing = True
            for on_time, off_time in pattern:
                self.on()
                time.sleep(on_time)
                self.off()
                time.sleep(off_time)
            self.is_playing = False
        
        self.play_thread = threading.Thread(target=_play, daemon=True)
        self.play_thread.start()
        return True
    
    def starwars_theme(self, duration=5):
        """Play Star Wars theme pattern"""
        if self.is_playing:
            return False
        
        def _play():
            self.is_playing = True
            start_time = time.time()
            
            # Opening sequence
            self.beep(0.05)
            time.sleep(0.05)
            self.beep(0.05)
            time.sleep(0.05)
            self.beep(0.05)
            time.sleep(0.1)
            self.beep(0.1)
            time.sleep(0.1)
            self.beep(0.15)
            time.sleep(0.15)
            
            # Continue pattern until duration reached
            while time.time() - start_time < duration:
                self.beep(0.08)
                time.sleep(0.05)
                self.beep(0.08)
                time.sleep(0.05)
                self.beep(0.08)
                time.sleep(0.1)
                self.beep(0.12)
                time.sleep(0.1)
            
            self.off()
            self.is_playing = False
        
        self.play_thread = threading.Thread(target=_play, daemon=True)
        self.play_thread.start()
        return True

# Create global buzzer controller
buzzer = BuzzerController()

@app.route('/')
def index():
    """API documentation endpoint"""
    return jsonify({
        'status': 'Buzzer Control API',
        'version': '1.0',
        'endpoints': {
            '/buzzer/on': {
                'method': 'POST, GET',
                'description': 'Turn buzzer ON'
            },
            '/buzzer/off': {
                'method': 'POST, GET',
                'description': 'Turn buzzer OFF'
            },
            '/buzzer/beep': {
                'method': 'GET',
                'description': 'Play a beep',
                'parameters': {'duration': 'float (0-5 seconds)'}
            },
            '/buzzer/pattern': {
                'method': 'POST',
                'description': 'Play custom pattern',
                'body': {'pattern': '[[on_time, off_time], ...]'}
            },
            '/buzzer/starwars': {
                'method': 'GET',
                'description': 'Play Star Wars theme',
                'parameters': {'duration': 'float (0-10 seconds)'}
            },
            '/buzzer/status': {
                'method': 'GET',
                'description': 'Get buzzer status'
            }
        }
    })

@app.route('/buzzer/on', methods=['POST', 'GET'])
def turn_on():
    """Turn buzzer ON"""
    if buzzer.on():
        return jsonify({'status': 'success', 'message': 'Buzzer turned ON'})
    return jsonify({'status': 'error', 'message': 'Failed to turn buzzer ON'}), 500

@app.route('/buzzer/off', methods=['POST', 'GET'])
def turn_off():
    """Turn buzzer OFF"""
    if buzzer.off():
        return jsonify({'status': 'success', 'message': 'Buzzer turned OFF'})
    return jsonify({'status': 'error', 'message': 'Failed to turn buzzer OFF'}), 500

@app.route('/buzzer/beep', methods=['POST', 'GET'])
def beep():
    """Play a beep for specified duration"""
    duration = float(request.args.get('duration', 0.2))
    if duration < 0 or duration > 5:
        return jsonify({
            'status': 'error', 
            'message': 'Duration must be between 0 and 5 seconds'
        }), 400
    
    buzzer.beep(duration)
    return jsonify({
        'status': 'success', 
        'message': f'Beeped for {duration} seconds'
    })

@app.route('/buzzer/pattern', methods=['POST'])
def play_pattern():
    """Play a custom pattern"""
    data = request.get_json()
    if not data or 'pattern' not in data:
        return jsonify({
            'status': 'error', 
            'message': 'Pattern required in JSON body'
        }), 400
    
    pattern = data['pattern']
    if not isinstance(pattern, list):
        return jsonify({
            'status': 'error', 
            'message': 'Pattern must be a list'
        }), 400
    
    # Validate pattern format
    try:
        validated_pattern = []
        for item in pattern:
            if isinstance(item, list) and len(item) == 2:
                validated_pattern.append((float(item[0]), float(item[1])))
            else:
                return jsonify({
                    'status': 'error', 
                    'message': 'Each pattern item must be [on_time, off_time]'
                }), 400
    except (ValueError, TypeError) as e:
        return jsonify({
            'status': 'error', 
            'message': f'Invalid pattern format: {str(e)}'
        }), 400
    
    if buzzer.play_pattern(validated_pattern):
        return jsonify({
            'status': 'success', 
            'message': 'Pattern playing'
        })
    return jsonify({
        'status': 'error', 
        'message': 'Buzzer is already playing'
    }), 409

@app.route('/buzzer/starwars', methods=['GET'])
def starwars():
    """Play Star Wars theme"""
    duration = float(request.args.get('duration', 5))
    if duration < 0 or duration > 10:
        return jsonify({
            'status': 'error', 
            'message': 'Duration must be between 0 and 10 seconds'
        }), 400
    
    if buzzer.starwars_theme(duration):
        return jsonify({
            'status': 'success', 
            'message': f'Playing Star Wars theme for {duration} seconds'
        })
    return jsonify({
        'status': 'error', 
        'message': 'Buzzer is already playing'
    }), 409

@app.route('/buzzer/status', methods=['GET'])
def status():
    """Get buzzer status"""
    try:
        result = subprocess.run(['cat', BUZZER_PATH], 
                              capture_output=True, text=True, check=True)
        state = result.stdout.strip()
        return jsonify({
            'status': 'success',
            'buzzer': 'ON' if state == '1' else 'OFF',
            'state': state,
            'is_playing': buzzer.is_playing
        })
    except Exception as e:
        return jsonify({
            'status': 'error', 
            'message': str(e)
        }), 500

if __name__ == '__main__':
    print("Starting Buzzer Control API...")
    print("API available at http://0.0.0.0:5000")
    print("Documentation at http://0.0.0.0:5000/")
    app.run(host='0.0.0.0', port=5000, debug=True)