fourier-ondemand/ondemand/service_streamlit.py

# -*- coding: utf8 -*-
from __future__ import print_function, absolute_import
from tornado.ioloop import IOLoop
from client import Client, ConnectionError
from boxconfig import parse_config
from dejavu.recognize import FilePerSecondRecognizer
from endpoint import setup_endpoint
from calibration import Calibrations
from dejavu import Dejavu, CouldntDecodeError
from multiprocessing import Process
import logging as log
import mutagen.mp3
import math
import sys
import os
import time
from datetime import datetime
import streamlit as st

st.set_page_config(layout="wide")

if sys.version_info >= (3, 0):
    from queue import Queue, Empty
else:
    from Queue import Queue, Empty

log.basicConfig(format='[%(asctime)s] [%(module)s] %(message)s', level=log.INFO)

AUDIOS_PATH = '/tmp'
AHEAD_TIME_AUDIO_TOLERANCE = 2  # second
MAX_SEGMENT_THREADS = 4
THRESHOLD = 10
SEGMENTS_TOLERANCE_RATE = 0.6
FALL_TOLERANCE_SEGMENTS = 1

# THRESHOLD
THRESHOLD_FIXED = 1
THRESHOLD_AVERAGE = 2

# Modos de procesamiento de queue
#  - QUEQUE_SINGLE: procesa solo un segmento a la vez
#  - QUEUE_THREAD:  inicia un hilo para cada segmento
# Por default se usará el threaded.
# TODO: hacerlo configurable por medio de argumentos
#       de ejecución.
QUEUE_SINGLE = 1
QUEUE_THREAD = 2

# Se pueden usar diferentes API'se
# la de threading y la de multiprocessing.
MultiAPI = Process

config = parse_config()
queue = Queue()
client = Client(config['device_id'],
                config['apiSecret'])
cloud_base_url = 'https://storage.googleapis.com/{}' \
    .format(config['bucket'])
base_path = config.get("basepath", "/var/fourier")
device_id = config['device_id']
device_path = os.path.join(base_path, device_id)
recognizer = FilePerSecondRecognizer

# settings
queue_mode = QUEUE_SINGLE
threshold_mode = THRESHOLD_FIXED

db_path = config.get('localDatabase', os.path.join(device_path, 'files.db'))
#db = sqlite3.connect(db_path)
cloud_cache = {}

def process_segment(anuncios, grabaciones, audios=None, calibration=None):
    segment_size = int(calibration['ss'])
    audio_length = 0

    dejavu = Dejavu({"database_type": "mem"})
    try:
        for i in range(0, len(anuncios)):
            path = "anuncios/{}".format(anuncios[i].name,)
            dejavu.fingerprint_file(path)
    except Exception as ex:
        log.error('[process_segment] cannot fingerprint: {}'.format(ex))

    audios_counter = 0
    results = []
    v = []

    st.subheader("Resultados de la comparación")
    for i in range(0, len(grabaciones)):
        path = "grabaciones/{}".format(grabaciones[i].name,)
        values = []
        try:
            seconds = 0
            for match in dejavu.recognize(recognizer, path, segment_size):
                name = ""
                if "name" in match:
                    name = match["name"]

                results.append({
                    "path": path,
                    "name": name,
                    "confidence": match["confidence"],
                    "offset": match["offset"],
                    "offset_seconds": seconds
                })
                values.append(str(match['confidence']))
                seconds += segment_size

            v.append(','.join(values))
            log.info('{0} {1}'.format(
                grabaciones[i].name,
                ','.join(values),
            ))
            st.text('{0} {1}'.format(
                grabaciones[i].name,
                ','.join(values),
            ))

        except CouldntDecodeError as ex:
            log.error('[process_segment] {}'.format(ex))

    try:
        encontrados = find_repetitions(results, segments_needed=int(calibration['sn']), calibration=calibration)
        st.subheader("Encontrados")
        st.write(encontrados)
    except ConnectionError as ex:
        log.error('[process_segment] {}'.format(str(ex)))
    except UserWarning as warn:
        log.warning(str(warn))


def find_repetitions(results, segments_needed=2, calibration=None):
    found_counter = 0
    found_down_counter = 0
    found_index = None
    expect_space = False
    expect_recover = False
    last_value_in_threshold_index = -1
    fall_tolerance = calibration['tf']
    found = []
    last_found = None

    if threshold_mode == THRESHOLD_FIXED:
        threshold = int(calibration['th'])
    elif threshold_mode == THRESHOLD_AVERAGE:
        values = [x['confidence'] for x in results]
        threshold = math.ceil(float(sum(values)) / float(len(values)))

    if segments_needed < 1:
        segments_needed = 1

    for index, result in enumerate(results):
        if not expect_space:
            if result['confidence'] >= threshold:
                found_counter += 1
                last_value_in_threshold_index = index
                if found_index is None:
                    found_index = index
                if expect_recover:
                    found_counter += found_down_counter
                    expect_recover = False

            elif fall_tolerance:
                if not expect_recover:
                    if last_value_in_threshold_index != -1:
                        expect_recover = True
                        found_down_counter += 1
                    else:
                        pass
                else:
                    found_counter = 0
                    found_down_counter = 0
                    found_index = None
                    expect_recover = False

            else:
                found_counter = 0
                found_down_counter = 0
                found_index = None
                expect_recover = False

        else:
            if result['confidence'] <= threshold:
                expect_space = False

        if found_counter >= segments_needed and last_found != found_index:
            found_row = results[found_index]
            found.append(found_row)
            last_found = found_index
            found_counter = 0
            expect_space = True

    return found


def limpiar_archivos():
    anuncios = os.listdir('anuncios/')
    for audio in anuncios:
        if audio.find('.mp3') > -1:
            os.remove('anuncios/{}'.format(audio,))

    grabaciones = os.listdir('grabaciones/')
    for audio in grabaciones:
        if audio.find('.mp3') > -1:
            os.remove('grabaciones/{}'.format(audio))


def main():

    st.subheader('Subir archivos para comparar')

    u1, u2 = st.beta_columns([3, 3])
    anuncios = u1.file_uploader("Anuncios", accept_multiple_files=True, type="mp3")
    for i in range(0, len(anuncios)):
        with open("anuncios/{}".format(anuncios[i].name,), "wb") as audio:
            audio.write(anuncios[i].getvalue())

    grabaciones = u2.file_uploader("Grabaciones", accept_multiple_files=True, type="mp3")
    grabaciones.sort(key=lambda x: x.name)
    for i in range(0, len(grabaciones)):
        with open("grabaciones/{}".format(grabaciones[i].name,), "wb") as audio:
            audio.write(grabaciones[i].getvalue())

    if st.button("Borrar archivos anteriores"):
        limpiar_archivos()

    st.subheader('Parámetros de calibración')

    col1, col2, col3, col4 = st.beta_columns([1,1,1,1])
    umbral = col1.text_input("Umbral", 12)
    segmentos_necesarios = col2.text_input("Sementos necesarios", 4)
    caida = col3.text_input("Tolerancia a caida", 1)
    segmento = col4.text_input("Tamaño del Segmento", 5)

    calibracion = {
        "th": umbral,
        "tf": caida,
        "sn": segmentos_necesarios,
        "ss": segmento
    }

    if st.button("Comparar"):
        process_segment(anuncios, grabaciones, calibration=calibracion)

main()