import os
import sox
import uuid
import logging
import json
from multiprocessing import Pool, Manager, Queue

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AD17")

src_dir = "/home/sk2/trans/workspace/audio/asr/data_gen/arabic_dialect_dataset/train_segments"
dialects = ["ALG", "IRA", "KSA", "LEB", "MAU", "OMA", "QAT", "SYR", "YEM", "EGY", "JOR", "KUW", "LIB", "MOR", "PAL",
            "SUD", "UAE"]
tgt_wav_dir = "data/arabic/ad17/data/wav"
tgt_manifest = "data/arabic/ad17/data/manifest.json"


def clean_manifests(sq: Queue, fl_list: list):
    durations = 0.0
    for wave_file_path in fl_list:
        if wave_file_path.endswith(".wav") and os.path.isfile(wave_file_path):
            rec_id = str(uuid.uuid4())
            logger.info("rec_id: {}".format(rec_id))
            tfm = sox.Transformer()
            tfm.set_output_format(file_type='wav', rate=16000, channels=1, bits=16, encoding='signed-integer')
            out_wav_file = os.path.join(tgt_wav_dir, "{}.wav".format(rec_id))
            status, out, err = tfm.build_file(input_filepath=wave_file_path,
                                              output_filepath=out_wav_file, return_output=True)
            logger.debug("Status: {}, Out: {}".format(status, out))
            if status == 0:
                duration = sox.file_info.duration(out_wav_file)
                if duration and duration > 0.0:
                    metadata = {
                        "audio_filepath": out_wav_file,
                        "duration": duration,
                    }
                    sq.put(metadata)
                    durations = durations + duration
                else:
                    logger.info(
                        'Discarding file: {} since duration: {}'.format(out_wav_file, duration))

                    logger.debug(
                        'Duration: {}, rec_id: {}'.format(duration, rec_id))

            else:
                logger.warning(
                    "Build failed for: {}, status: {}, err: {}".format(out_wav_file, status, err))

        logger.info('Total durations processed: {}, wav: {}\n\n\n'.format(durations, wave_file_path))

    return True, durations, os.getpid()


def write_manifests(sq: Queue):
    tgt_json = open(tgt_manifest, encoding="utf-8", mode="w")
    count = 0
    while True:
        meta_data = sq.get()
        if "last_item" in meta_data and meta_data["last_item"]:
            break

        json.dump(meta_data, tgt_json)
        tgt_json.write("\n")
        count = count + 1

    tgt_json.flush()
    tgt_json.close()
    return True, os.getpid(), count


def proc_parallel():
    os.makedirs(tgt_wav_dir, exist_ok=True)
    batch_size = 128
    with Manager() as man:
        q = man.Queue()
        with Pool(72) as pool:
            ret = pool.apply_async(write_manifests, (q,))
            proc_rets = []
            mini_batch = []
            for dialect in dialects:
                src_wav_dir = os.path.join(src_dir, dialect)
                dir_stream = os.scandir(src_wav_dir)
                for fl_en in dir_stream:
                    try:
                        if len(mini_batch) <= batch_size:
                            mini_batch.append(os.path.join(src_wav_dir, str(fl_en.name)))
                        else:
                            proc_ret = pool.apply_async(clean_manifests, (q, mini_batch.copy()))
                            proc_rets.append(proc_ret)
                            mini_batch.clear()
                            mini_batch.append(str(fl_en.name))
                    except Exception as e:
                        logger.exception(e)

                dir_stream.close()

            if len(mini_batch) > 0:
                proc_ret = pool.apply_async(clean_manifests, (q, mini_batch.copy()))
                proc_rets.append(proc_ret)
                mini_batch.clear()

            p_count = 0
            t_dur = 0.0
            for p_ret in proc_rets:
                p_stat, dur, pid = p_ret.get()
                if p_stat:
                    t_dur = t_dur + dur
                    p_count = p_count + 1
                    logger.info("Completed proc: {} {}/{}".format(pid, p_count, len(proc_rets) + 1))

            q.put({"last_item": True})

            p_stat, pid, q_count = ret.get()
            if p_stat:
                p_count = p_count + 1
                logger.info("Completed proc: {} {}/{}".format(pid, p_count, len(proc_rets) + 1))

            logger.info("Total processed: {}\n\n".format(q_count))
            logger.info("Total durations processed: {}\n\n".format(t_dur))


if __name__ == '__main__':
    proc_parallel()

import os
import sox
import uuid
import logging
import json
from multiprocessing import Pool, Manager, Queue

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AD17")

src_dir = "/home/sk2/trans/workspace/audio/asr/data_gen/arabic_dialect_dataset/train_segments"
dialects = ["ALG", "IRA", "KSA", "LEB", "MAU", "OMA", "QAT", "SYR", "YEM", "EGY", "JOR", "KUW", "LIB", "MOR", "PAL",
            "SUD", "UAE"]
tgt_wav_dir = "data/arabic/ad17/data/wav"
tgt_manifest = "data/arabic/ad17/data/manifest.json"


def clean_manifests(sq: Queue, fl_list: list):
    durations = 0.0
    for wave_file_path in fl_list:
        if wave_file_path.endswith(".wav") and os.path.isfile(wave_file_path):
            rec_id = str(uuid.uuid4())
            logger.info("rec_id: {}".format(rec_id))
            tfm = sox.Transformer()
            tfm.set_output_format(file_type='wav', rate=16000, channels=1, bits=16, encoding='signed-integer')
            out_wav_file = os.path.join(tgt_wav_dir, "{}.wav".format(rec_id))
            status, out, err = tfm.build_file(input_filepath=wave_file_path,
                                              output_filepath=out_wav_file, return_output=True)
            logger.debug("Status: {}, Out: {}".format(status, out))
            if status == 0:
                duration = sox.file_info.duration(out_wav_file)
                if duration and duration > 0.0:
                    metadata = {
                        "audio_filepath": out_wav_file,
                        "duration": duration,
                    }
                    sq.put(metadata)
                    durations = durations + duration
                else:
                    logger.info(
                        'Discarding file: {} since duration: {}'.format(out_wav_file, duration))

                    logger.debug(
                        'Duration: {}, rec_id: {}'.format(duration, rec_id))

            else:
                logger.warning(
                    "Build failed for: {}, status: {}, err: {}".format(out_wav_file, status, err))

        logger.info('Total durations processed: {}, wav: {}\n\n\n'.format(durations, wave_file_path))

    return True, durations, os.getpid()


def write_manifests(sq: Queue):
    tgt_json = open(tgt_manifest, encoding="utf-8", mode="w")
    count = 0
    while True:
        meta_data = sq.get()
        if "last_item" in meta_data and meta_data["last_item"]:
            break

        json.dump(meta_data, tgt_json)
        tgt_json.write("\n")
        count = count + 1

    tgt_json.flush()
    tgt_json.close()
    return True, os.getpid(), count


def proc_parallel():
    os.makedirs(tgt_wav_dir, exist_ok=True)
    batch_size = 128
    with Manager() as man:
        q = man.Queue()
        with Pool(72) as pool:
            ret = pool.apply_async(write_manifests, (q,))
            proc_rets = []
            mini_batch = []
            for dialect in dialects:
                src_wav_dir = os.path.join(src_dir, dialect)
                dir_stream = os.scandir(src_wav_dir)
                for fl_en in dir_stream:
                    try:
                        if len(mini_batch) <= batch_size:
                            mini_batch.append(os.path.join(src_wav_dir, str(fl_en.name)))
                        else:
                            proc_ret = pool.apply_async(clean_manifests, (q, mini_batch.copy()))
                            proc_rets.append(proc_ret)
                            mini_batch.clear()
                            mini_batch.append(str(fl_en.name))
                    except Exception as e:
                        logger.exception(e)

                dir_stream.close()

            if len(mini_batch) > 0:
                proc_ret = pool.apply_async(clean_manifests, (q, mini_batch.copy()))
                proc_rets.append(proc_ret)
                mini_batch.clear()

            p_count = 0
            t_dur = 0.0
            for p_ret in proc_rets:
                p_stat, dur, pid = p_ret.get()
                if p_stat:
                    t_dur = t_dur + dur
                    p_count = p_count + 1
                    logger.info("Completed proc: {} {}/{}".format(pid, p_count, len(proc_rets) + 1))

            q.put({"last_item": True})

            p_stat, pid, q_count = ret.get()
            if p_stat:
                p_count = p_count + 1
                logger.info("Completed proc: {} {}/{}".format(pid, p_count, len(proc_rets) + 1))

            logger.info("Total processed: {}\n\n".format(q_count))
            logger.info("Total durations processed: {}\n\n".format(t_dur))


if __name__ == '__main__':
    proc_parallel()
