ESP32 Forum

Code: Select all

import datetime
import wave
from flask import Flask, request, Response
import sys

app = Flask(__name__)

def _get_chunk_size():
    data = request.stream.read(2)
    print(type(data))
    
    while data[-2:] != b"\r\n":
        data += request.stream.read(1)
    return int(data[:-2], 16)

def _get_chunk_data( chunk_size):
    data = request.stream.read(chunk_size)
    request.stream.read(2)
    return data

def _write_wav( data, rates, bits, ch):
    t = datetime.datetime.utcnow()
    time = t.strftime('%Y%m%dT%H%M%SZ')
    filename = str.format('{}_{}_{}_{}.wav', time, rates, bits, ch)

    wavfile = wave.open(filename, 'wb')
    wavfile.setparams((ch, int(bits/8), rates, 0, 'NONE', 'NONE'))
    wavfile.writeframesraw(bytearray(data))
    wavfile.close()
    return filename

@app.route('/')
def hello_world():
    return 'Hello, World!'

@app.route("/upload", methods=["POST"])
def upload():
    total_bytes = 0
    sample_rates = 0
    bits = 0
    channel = 0
    if request.headers.get('Transfer-Encoding').lower() == 'chunked':
        data = []
        sample_rates = request.headers.get('x-audio-sample-rates', '').lower()
        bits = request.headers.get('x-audio-bits', '').lower()
        channel = request.headers.get('x-audio-channel', '').lower()
        sample_rates = request.headers.get('x-audio-sample-rates', '').lower()
        print("Audio information, sample rates: {}, bits: {}, channel(s): {}".format(sample_rates, bits, channel))
        while True:
            chunk_size = _get_chunk_size()
            total_bytes += chunk_size
            print("Total bytes received: {}".format(total_bytes))
            sys.stdout.write("\033[F")
            if (chunk_size == 0):
                break
            else:
                chunk_data = _get_chunk_data(chunk_size)
                data += chunk_data

        filename = _write_wav(data, int(sample_rates), int(bits), int(channel))
        def generate():
            yield 'File {} was written, size {}'.format(filename, total_bytes).encode('utf-8')


        response = Response(generate(), mimetype='text/html')
        response.status_code = 200
        response.headers['Content-type'] = 'text/html;charset=utf-8'
        response.headers['Content-Length'] = str(total_bytes)
        
        return response
        
        

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000, debug=True)