avh-align_core/core.py

from typing import List
from rest.request import RequestData
from decorators import track_altair_license, altair_ready
from typing import Union
import settings
from core import config
import time

import logging
logger = logging.getLogger("ingp")

READY = False
from avh_align import *

def load(folder_name='model'):
    global READY
    try:
        logger.debug(f"APP_LANG: {settings.APP_LANG}")
        logger.info("### loading models ")
        load_models(folder_name)
        logger.info('### models loaded! System ready')

    except Exception as e:
        logger.error(f"Error loading models: {e}")
        READY = False


def unload(folder_name=None):
    global READY
    READY = False


@altair_ready()
def ready():
    return READY


@track_altair_license(use_logger=logger)
def process(data: Union[List[RequestData], RequestData]):
    if type(data) is list:
        return [process_audio_video(elm.content) for elm in data]
    return process_audio_video(data.content)


@track_altair_license(use_logger=logger)
def process_file(files: List[bytes], lang: List[str]):
    out = []
    for file in files:
        try:
            out.append(process_audio_video(file.decode()))
        except UnicodeDecodeError:
            out.append({'error': 'File type not supported'})
    return out


def reverse(t: str):
    time.sleep(config.CORE_SLEEP)
    return {'reversed': t[::-1]}