import argparse

import io
import os
import torch
import evaluate
import soundfile

from tqdm import tqdm
from normalizer import data_utils
import numpy as np

import fluid_speech
import time

# The same decoding for every dataset: batched beam search (modified ALSD for TDT) with 4 beams; every other beam
# setting stays as saved in the model config.
BEAM_STRATEGY = "malsd_batch"
BEAM_SIZE = 4


wer_metric = evaluate.load("wer")


def main(args):

    data_cache_root = args.data_cache_root if args.data_cache_root is not None else os.getcwd()
    DATA_CACHE_DIR = os.path.join(data_cache_root, "audio_cache")
    DATASET_NAME = args.dataset
    SPLIT_NAME = args.split

    CACHE_DIR = os.path.join(DATA_CACHE_DIR, DATASET_NAME, SPLIT_NAME)
    if not os.path.exists(CACHE_DIR):
        os.makedirs(CACHE_DIR)

    if args.device >= 0:
        device = torch.device(f"cuda:{args.device}")
        compute_dtype=torch.bfloat16
    else:
        device = torch.device("cpu")
        compute_dtype=torch.float32

    # Hub repo id or local .nemo; adds the model's extra encoder block, then loads every tensor strictly.
    # --clean_mode selects the edited (non-verbatim) transcription style; submit_jobs.sh sets it per dataset.
    asr_model = fluid_speech.load(args.model_id, map_location=device, clean_mode=args.clean_mode)
    print(f"Transcription mode: {'clean (edited)' if args.clean_mode else 'verbatim'}")
    
    asr_model.to(compute_dtype)
    asr_model.eval()
    print(f"Model size: {sum(p.numel() for p in asr_model.parameters()) / 1e9:.2f}B parameters")

    is_chunked = data_utils.is_chunked_dataset(args.dataset_path)

    dataset = data_utils.load_data(args)

    if args.max_eval_samples is not None and args.max_eval_samples > 0:
        print(f"Subsampling dataset to first {args.max_eval_samples} samples !")
        dataset = dataset.take(args.max_eval_samples)

    # Prepare data FIRST - this casts audio to proper format with "array" and "sampling_rate" keys
    dataset = data_utils.prepare_data(dataset)

    def download_audio_files(batch):

        # download audio files and write the paths, transcriptions and durations to a manifest file
        audio_paths = []
        original_audio_paths = []
        durations = []
        file_names = batch.get("file_name", [None] * len(batch["audio"]))

        # Use 'id' column if available, otherwise generate sequential IDs
        if "id" in batch:
            ids = batch["id"]
        else:
            # Generate IDs based on index
            start_idx = len([f for f in os.listdir(CACHE_DIR) if f.endswith('.wav')]) if os.path.exists(CACHE_DIR) else 0
            ids = [f"sample_{start_idx + i}" for i in range(len(batch["audio"]))]

        for id, file_name, audio_sample in zip(ids, file_names, batch["audio"]):

            # first step added here to make ID and wav filenames unique
            # several datasets like earnings22 have a hierarchical structure
            # for eg. earnings22/test/4432298/281.wav, earnings22/test/4450488/281.wav
            # lhotse uses the filename (281.wav) here as unique ID to create and name cuts
            # ref: https://github.com/lhotse-speech/lhotse/blob/master/lhotse/dataset/collation.py#L186
            original_id = id  # preserve before sanitization for use as audio_filepath
            id = id.replace('/', '_').removesuffix('.wav')

            audio_path = os.path.join(CACHE_DIR, f"{id}.wav")
            audio_array = np.float32(audio_sample["array"])
            sample_rate = audio_sample["sampling_rate"]

            if not os.path.exists(audio_path):
                os.makedirs(os.path.dirname(audio_path), exist_ok=True)
                soundfile.write(audio_path, audio_array, sample_rate)

            audio_paths.append(audio_path)
            # Prefer the original file_name from the dataset; fall back to the
            # sample id (before path-sanitization) so audio_filepath in the
            # JSONL is always a meaningful identifier rather than "sample_N".
            if file_name is not None:
                original_audio_paths.append(os.path.basename(str(file_name)))
            else:
                original_audio_paths.append(original_id)
            durations.append(len(audio_array) / sample_rate)

        
        batch["references"] = batch["original_text"]  # raw text; normalization applied at scoring time
        batch["audio_filepaths"] = audio_paths
        batch["original_audio_filepaths"] = original_audio_paths
        batch["durations"] = durations

        return batch

    asr_model.cfg.decoding.strategy = BEAM_STRATEGY
    asr_model.cfg.decoding.beam.beam_size = BEAM_SIZE
    asr_model.change_decoding_strategy(asr_model.cfg.decoding)
    
    # prepraing the offline dataset
    dataset = dataset.map(download_audio_files, batch_size=args.batch_size, batched=True, remove_columns=["audio"])

    # Write manifest from daraset batch using json and keys audio_filepath, duration, text

    all_data = {
        "audio_filepaths": [],
        "original_audio_filepaths": [],
        "durations": [],
        "references": [],
    }
    if is_chunked:
        all_data.update({key: [] for key in data_utils.CHUNK_METADATA_KEYS})

    data_itr = iter(dataset)
    for data in tqdm(data_itr, desc="Downloading Samples"):
        for key in all_data:
            all_data[key].append(data[key])
    
    # Sort audio_filepaths and references based on durations values
    sorted_indices = sorted(range(len(all_data["durations"])), key=lambda k: all_data["durations"][k], reverse=True)
    for key, values in all_data.items():
        all_data[key] = [values[i] for i in sorted_indices]
    
    
    total_time = 0
    for _ in range(2): # warmup once and calculate rtf
        if _ == 0:
            audio_files = all_data["audio_filepaths"][:args.batch_size * 4] # warmup with 4 batches
        else:
            audio_files = all_data["audio_filepaths"]
        start_time = time.time()
        with torch.inference_mode(), torch.no_grad(): 

            if 'canary' in args.model_id and 'v2' not in args.model_id:
                pnc = 'nopnc'
            else:
                pnc = 'pnc'
                
            if 'canary' in args.model_id:
                transcriptions = asr_model.transcribe(audio_files, batch_size=args.batch_size, verbose=False, pnc=pnc, num_workers=1)
            else:
                transcriptions = asr_model.transcribe(audio_files, batch_size=args.batch_size, verbose=False, num_workers=1)
        end_time = time.time()
        if _ == 1:
            total_time += end_time - start_time
    total_time = total_time

    if isinstance(transcriptions, tuple) and len(transcriptions) == 2:
        transcriptions = transcriptions[0]
    # beam search returns each clip's hypotheses best first
    predictions = [(pred[0] if isinstance(pred, list) else pred).text for pred in transcriptions]  # raw; normalization applied at scoring time

    avg_time = total_time / len(all_data["audio_filepaths"])

    # Write manifest results (WER and RTFX)
    manifest_path = data_utils.write_manifest(
        all_data["references"],
        predictions,
        args.model_id,
        args.dataset_path,
        args.dataset,
        args.split,
        audio_length=all_data["durations"],
        transcription_time=[avg_time] * len(all_data["audio_filepaths"]),
        audio_filepaths=all_data["original_audio_filepaths"],
        extra_fields={key: all_data[key] for key in data_utils.CHUNK_METADATA_KEYS}
        if is_chunked
        else None,
    )

    print("Results saved at path:", os.path.abspath(manifest_path))

    if is_chunked:
        sessions = data_utils.merge_chunked_manifest(
            data_utils.read_manifest(manifest_path)
        )
        references = [session["text"] for session in sessions]
        predictions = [session["pred_text"] for session in sessions]
    else:
        references = all_data["references"]

    norm_references = [data_utils.normalizer(r) for r in references]
    norm_predictions = [data_utils.normalizer(p) for p in predictions]
    wer = wer_metric.compute(references=norm_references, predictions=norm_predictions)
    wer = round(100 * wer, 2)

    # transcription_time = sum(all_results["transcription_time"])
    audio_length = sum(all_data["durations"])
    rtfx = audio_length / total_time
    rtfx = round(rtfx, 2)

    print("RTFX:", rtfx)
    print("WER:", wer, "%")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--model_id", type=str, required=True, help="Model identifier. Should be loadable with NVIDIA NeMo.",
    )
    parser.add_argument(
        '--dataset_path', type=str, default='hf-audio/open-asr-leaderboard', help='Dataset path. By default, it is `hf-audio/open-asr-leaderboard`'
    )
    parser.add_argument(
        '--data_cache_root', type=str, default=None, help='Root directory for audio cache. By default, it is the current working directory.'
    )
    parser.add_argument(
        "--dataset",
        type=str,
        required=True,
        help="Dataset name. *E.g.* `'librispeech_asr` for the LibriSpeech ASR dataset, or `'common_voice'` for Common Voice. The full list of dataset names "
        "can be found at `https://huggingface.co/datasets/hf-audio/open-asr-leaderboard`",
    )
    parser.add_argument(
        "--split",
        type=str,
        default="test",
        help="Split of the dataset. *E.g.* `'validation`' for the dev split, or `'test'` for the test split.",
    )
    parser.add_argument(
        "--device",
        type=int,
        default=-1,
        help="The device to run the pipeline on. -1 for CPU (default), 0 for the first GPU and so on.",
    )
    parser.add_argument(
        "--batch_size", type=int, default=32, help="Number of samples to go through each streamed batch.",
    )
    parser.add_argument(
        "--max_eval_samples",
        type=int,
        default=None,
        help="Number of samples to be evaluated. Put a lower number e.g. 64 for testing this script.",
    )
    parser.add_argument(
        "--streaming",
        action="store_true",
        help="Stream the dataset lazily over the network instead of downloading it in full before the evaluation. Off by default for reproducible benchmark timings.",
    )
    parser.add_argument(
        "--clean_mode",
        action="store_true",
        help="Transcription mode: clean (no repeated words or filler words). Off = verbatim.",
    )
    args = parser.parse_args()

    main(args)