from flask import Flask, render_template, request, redirect, stream_with_context
from utils.transcribe import handle_transcription
from utils.handle_upload import check_inputs, get_inputs, handle_file, format_duration
from utils.archive import read_archive, create_metadata, delete_transcription, open_file_directory, TIMESTAMP_FORMAT
import traceback
import webview
from screeninfo import get_monitors
import yaml
from datetime import datetime
from wakepy import keep
import webbrowser
from version import VERSION

app = Flask(__name__)
app.jinja_env.filters['format_duration'] = format_duration

@app.get("/")
def render_home():
    return render_template("index.html", version=VERSION)

@app.get("/transcribe")
def render_transcribe():
    return render_template("transcribe.html")

@app.get("/archive")
def render_archive():
    archive_data = read_archive()
    return render_template("archive.html", archive_data=archive_data)

@app.get("/faq")
def render_faq():
    faq_path = "faq.yaml"
    with open(faq_path,"r", encoding='utf-8') as faq_file:
        faqs = yaml.safe_load(faq_file)
    return render_template("faq.html", faqs = faqs)

@app.post("/upload")
def upload():
    timestamp = datetime.now().strftime(TIMESTAMP_FORMAT)
    try:
        file, model, language, speaker_detection, num_speakers = get_inputs(request) 
        inputs_correct = check_inputs(file, model, language, num_speakers)
        if inputs_correct is False:
            return render_template("modal_wrongInput.html")
        filename, file_id, estimated_process_time, device, audio_duration = handle_file(file, timestamp, model)
        create_metadata(file_id, filename, audio_duration, model, language, speaker_detection, num_speakers, device, timestamp)
        return render_template("modal_process.html", id=file_id, task="Processing file", total_duration=estimated_process_time, device=device)
    except Exception as error:
        delete_transcription(timestamp)
        traceback_str = traceback.format_exc()
        error = str(error)
        return render_template("modal_error.html",error=error, traceback=traceback_str)

@app.route("/transcription/<upload_id>")
def stream_data(upload_id):
    return stream_with_context(handle_transcription(upload_id)), {"Content-Type": "text/event-stream"}

@app.route('/open/<file_id>')
def open_transcription(file_id):
    open_file_directory(file_id)
    return ""
    
@app.route("/delete/<file_id>")
def delete_archive_files(file_id):
    delete_transcription(file_id)
    archive_data = read_archive()
    return render_template("archive.html", archive_data=archive_data)

@app.route("/revert_changes/<upload_id>")
def revert_latest_changes(upload_id):
    delete_transcription(upload_id)
    return redirect(request.referrer)

@app.get("/wakeup") #This route gets called every 3s by the client to prevent server from shutting down when no request is incoming for a while
def stay_awake():
    return ""

@app.get("/openbrowser/<website>")
def open_browser(website):
    match website:
        case "github": url = "https://github.com/BANDAS-Center/aTrain"
        case "feedback": url = "https://survey.uni-graz.at/index.php/381219"
    webbrowser.open_new(url)
    return ""

def get_screen_size():
    scaler = 0.8
    height = min([monitor.height for monitor in get_monitors()])*scaler
    width = min([monitor.width for monitor in get_monitors()])*scaler
    return round(width), round(height)

if __name__ == "__main__":
    #app.run(port=5000)
    width, height = get_screen_size()
    webview.create_window("aTrain",app, height=height, width=width)
    with keep.running() as k:
        webview.start()
app.py
BANDAS-Center/aTrain
import os
import shutil
from showinfm import show_in_file_manager
import yaml
from datetime import datetime

USER_DIR = os.path.expanduser("~")
DOCUMENTS_DIR = os.path.join(USER_DIR,"Documents")
ATRAIN_DIR = os.path.join(DOCUMENTS_DIR,"aTrain")
TRANSCRIPT_DIR = os.path.join(ATRAIN_DIR,"transcriptions")
METADATA_FILENAME = "metadata.txt"
TIMESTAMP_FORMAT = "%Y-%m-%d %H-%M-%S"

def read_archive():
    os.makedirs(TRANSCRIPT_DIR, exist_ok=True)
    all_file_directories = [ directory.name for directory in os.scandir(TRANSCRIPT_DIR) if directory.is_dir() ]
    all_file_directories.sort(reverse=True)
    all_metadata = []
    for directory_name in all_file_directories:
        metadata_file_path = os.path.join(TRANSCRIPT_DIR,directory_name,METADATA_FILENAME)
        if os.path.exists(metadata_file_path):
            with open(metadata_file_path, "r", encoding="utf-8") as metadata_file:
                metadata = yaml.safe_load(metadata_file)
                metadata["file_id"] = directory_name
            all_metadata.append(metadata)
        else:
            metadata = {
                "file_id" : directory_name,
                "filename" : directory_name[20:] if len(directory_name)>20 else "-",
                "timestamp" : directory_name[:20] if len(directory_name)>=20 else "-"
            }
            all_metadata.append(metadata)
    return all_metadata

def create_metadata(file_id, filename, audio_duration, model, language, speaker_detection, num_speakers, device, timestamp):
    metadata_file_path = os.path.join(TRANSCRIPT_DIR,file_id,METADATA_FILENAME)
    metadata = {
        "file_id" : file_id,
        "filename" : filename,
        "audio_duration" : audio_duration,
        "model" : model,
        "language" : language,
        "speaker_detection" : speaker_detection,
        "num_speakers" : num_speakers,
        "device": device,
        "timestamp": timestamp 
        }
    with open(metadata_file_path,"w", encoding="utf-8") as metadata_file:
        yaml.dump(metadata, metadata_file)

def read_metadata(file_id):
    metadata_file_path = os.path.join(TRANSCRIPT_DIR,file_id,METADATA_FILENAME)
    with open(metadata_file_path,"r", encoding="utf-8") as metadata_file:
        metadata = yaml.safe_load(metadata_file)
    filename = metadata["filename"]
    model = metadata["model"]
    language = metadata["language"]
    speaker_detection = metadata["speaker_detection"]
    num_speakers = metadata["num_speakers"]
    return filename, model, language, speaker_detection, num_speakers

def add_processing_time_to_metadata(file_id):
    metadata_file_path = os.path.join(TRANSCRIPT_DIR,file_id,METADATA_FILENAME)
    with open(metadata_file_path, "r", encoding="utf-8") as metadata_file:
        metadata = yaml.safe_load(metadata_file)
    timestamp = metadata["timestamp"]
    start_time = datetime.strptime(timestamp,TIMESTAMP_FORMAT)
    stop_time = timestamp = datetime.now()
    processing_time = stop_time-start_time
    metadata["processing_time"] = int(processing_time.total_seconds())
    with open(metadata_file_path, "w", encoding="utf-8") as metadata_file:
        yaml.dump(metadata,metadata_file)

def delete_transcription(file_id):
    file_id = "" if file_id == "all" else file_id
    directory_name = os.path.join(TRANSCRIPT_DIR,file_id)
    if os.path.exists(directory_name):    
        shutil.rmtree(directory_name)
    if not os.path.exists(TRANSCRIPT_DIR):
        os.makedirs(TRANSCRIPT_DIR, exist_ok=True)
    
def open_file_directory(file_id):
    file_id = "" if file_id == "all" else file_id
    directory_name = os.path.join(TRANSCRIPT_DIR,file_id)
    if os.path.exists(directory_name):
        show_in_file_manager(directory_name)

if __name__ == "__main__":
    ...
utils/archive.py
BANDAS-Center/aTrain
import os
import ffmpeg
from utils.archive import TRANSCRIPT_DIR
from scipy.io import wavfile
import requests
import shutil

def get_inputs(request):
    file = request.files["file"]
    model = request.form.get('model')
    language = request.form.get('language')
    speaker_detection = True if request.form.get('speaker_detection') else False
    num_speakers = request.form.get("num_speakers")
    return file, model, language, speaker_detection, num_speakers

def check_inputs(file, model, language, num_speakers):
    file_correct = check_file(file)
    model_correct = check_model(model)
    language_correct = check_language(language)
    num_speakers_correct = check_num_speakers(num_speakers)
    return file_correct and model_correct and language_correct and num_speakers_correct

def check_file(file):
    filename = file.filename
    file_extension = os.path.splitext(filename)[1]
    file_extension_lower = str(file_extension).lower()
    correct_file_formats = ['.3dostr', '.4xm', '.aa', '.aac', '.ac3', '.acm', '.act', '.adf', '.adp', '.ads', '.adx', '.aea', '.afc', '.aiff', '.aix', '.alaw', '.alias_pix', '.alp', '.amr', '.amrnb', '.amrwb', '.anm', '.apc', '.ape', '.apm', '.apng', '.aptx', '.aptx_hd', '.aqtitle', '.argo_asf', '.asf', '.asf_o', '.ass', '.ast', '.au', '.av1', '.avi', '.avr', '.avs', '.avs2', '.bethsoftvid', '.bfi', '.bfstm', '.bin', '.bink', '.bit', '.bmp_pipe', '.bmv', '.boa', '.brender_pix', '.brstm', '.c93', '.caf', '.cavsvideo', '.cdg', '.cdxl', '.cine', '.codec2', '.codec2raw', '.concat', '.data', '.daud', '.dcstr', '.dds_pipe', '.derf', '.dfa', '.dhav', '.dirac', '.dnxhd', '.dpx_pipe', '.dsf', '.dsicin', '.dss', '.dts', '.dtshd', '.dv', '.dvbsub', '.dvbtxt', '.dxa', '.ea', '.ea_cdata', '.eac3', '.epaf', '.exr_pipe', '.f32be', '.f32le', '.f64be', '.f64le', '.fbdev', '.ffmetadata', '.film_cpk', '.filmstrip', '.fits', '.flac', '.flic', '.flv', '.frm', '.fsb', '.fwse', '.g722', '.g723_1', '.g726', '.g726le', '.g729', '.gdv', '.genh', '.gif', '.gif_pipe', '.gsm', '.gxf', '.h261', '.h263', '.h264', '.hca', '.hcom', '.hevc', '.hls', '.hnm', '.ico', '.idcin', '.idf', '.iff', '.ifv', '.ilbc', '.image2', '.image2pipe', '.ingenient', '.ipmovie', '.ircam', '.iss', '.iv8', '.ivf', '.ivr', '.j2k_pipe', '.jacosub', '.jpeg_pipe', '.jpegls_pipe', '.jv', '.kux', '.kvag', '.lavfi', '.live_flv', '.lmlm4', '.loas', '.lrc', '.lvf', '.lxf', '.m4v', '.matroska', '.webm', '.mgsts', '.microdvd', '.mjpeg', '.mjpeg_2000', '.mlp', '.mlv', '.mm', '.mmf', '.mov', '.mp4', '.m4a', '.3gp', '.3g2', '.mj2', '.mp3', '.mpc', '.mpc8', '.mpeg', '.mpegts', '.mpegtsraw', '.mpegvideo', '.mpjpeg', '.mpl2', '.mpsub', '.msf', '.msnwctcp', '.mtaf', '.mtv', '.mulaw', '.musx', '.mv', '.mvi', '.mxf', '.mxg', '.nc', '.nistsphere', '.nsp', '.nsv', '.nut', '.nuv', '.ogg', '.oma', '.oss', '.paf', '.pam_pipe', '.pbm_pipe', '.pcx_pipe', '.pgm_pipe', '.pgmyuv_pipe', '.pictor_pipe', '.pjs', '.pmp', '.png_pipe', '.pp_bnk', '.ppm_pipe', '.psd_pipe', '.psxstr', '.pva', '.pvf', '.qcp', '.qdraw_pipe', '.r3d', '.rawvideo', '.realtext', '.redspark', '.rl2', '.rm', '.roq', '.rpl', '.rsd', '.rso', '.rtp', '.rtsp', '.s16be', '.s16le', '.s24be', '.s24le', '.s32be', '.s32le', '.s337m', '.s8', '.sami', '.sap', '.sbc', '.sbg', '.scc', '.sdp', '.sdr2', '.sds', '.sdx', '.ser', '.sgi_pipe', '.shn', '.siff', '.sln', '.smjpeg', '.smk', '.smush', '.sol', '.sox', '.spdif', '.srt', '.stl', '.subviewer', '.subviewer1', '.sunrast_pipe', '.sup', '.svag', '.svg_pipe', '.swf', '.tak', '.tedcaptions', '.thp', '.tiertexseq', '.tiff_pipe', '.tmv', '.truehd', '.tta', '.tty', '.txd', '.ty', '.u16be', '.u16le', '.u24be', '.u24le', '.u32be', '.u32le', '.u8', '.v210', '.v210x', '.vag', '.vc1', '.vc1test', '.vidc', '.video4linux2', '.v4l2', '.vividas', '.vivo', '.vmd', '.vobsub', '.voc', '.vpk', '.vplayer', '.vqf', '.w64', '.wav', '.wc3movie', '.webm_dash_manifest', '.webp_pipe', '.webvtt', '.wsaud', '.wsd', '.wsvqa', '.wtv', '.wv', '.wve', '.xa', '.xbin', '.xmv', '.xpm_pipe', '.xvag', '.xwd_pipe', '.xwma', '.yop', '.yuv4mpegpipe']
    return file_extension_lower in correct_file_formats

def check_model(model):
    correct_models = ["tiny", "base", "small", "medium" , "large-v1" , "large-v2"]
    return model in correct_models

def check_language(language):
    correct_languages = ['auto-detect','af', 'ar', 'hy', 'az', 'be', 'bs', 'bg', 'ca', 'zh', 'hr', 'cs', 'da', 'nl', 'en', 'et', 'fi', 'fr', 'gl', 'de', 'el', 'he', 'hi', 'hu', 'is', 'id', 'it', 'ja', 'kn', 'kk', 'ko', 'lv', 'lt', 'mk', 'ms', 'mr', 'mi', 'ne', 'no', 'fa', 'pl', 'pt', 'ro', 'ru', 'sr', 'sk', 'sl', 'es', 'sw', 'sv', 'tl', 'ta', 'th', 'tr', 'uk', 'ur', 'vi', 'cy']
    return language in correct_languages

def check_num_speakers(num_speakers):
    correct_num_speakers = ["auto-detect","1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
    return num_speakers in correct_num_speakers

def handle_file(file, timestamp, model):
    os.makedirs(TRANSCRIPT_DIR, exist_ok=True)
    filename = file.filename
    file_id = create_file_id(filename, timestamp)
    file_directory = os.path.join(TRANSCRIPT_DIR,file_id)
    os.makedirs(file_directory)
    file_path = os.path.join(file_directory, filename)
    file.save(file_path)
    processed_file = prepare_audio(file_id, file_path, file_directory)
    audio_duration = get_audio_duration(processed_file)
    estimated_process_time, device = estimate_processing_time(audio_duration,model)
    os.remove(file_path)
    return filename, file_id, estimated_process_time, device, audio_duration

def create_file_id(filename, timestamp):
    file_base_name = os.path.normpath(filename)
    short_base_name = file_base_name[0:49] if len(file_base_name) >= 50 else file_base_name
    file_id = timestamp + " " + short_base_name
    return file_id

def prepare_audio (file_id,file_path,file_directory):
    get_ffmpeg() #download ffmpeg if it does not exist
    output_file = file_id + ".wav"
    output_path =  os.path.join(file_directory,output_file)
    stream = ffmpeg.input(file_path)
    stream = ffmpeg.output(stream, output_path)
    ffmpeg.run(stream,quiet=True)
    return output_path

def get_ffmpeg():
    ffmpeg_path = "ffmpeg.exe"
    if not os.path.exists(ffmpeg_path):
        url = 'https://github.com/GyanD/codexffmpeg/releases/download/2023-10-02-git-9e531370b3/ffmpeg-2023-10-02-git-9e531370b3-essentials_build.zip'
        r = requests.get(url, allow_redirects=True)
        ffmpeg_zip = "ffmpeg.zip"
        ffmpeg_dir = "ffmpeg"
        ffmpeg_exe = os.path.join(ffmpeg_dir,"ffmpeg-2023-10-02-git-9e531370b3-essentials_build","bin","ffmpeg.exe")
        with open(ffmpeg_zip, 'wb') as ffmpeg_file:
            ffmpeg_file.write(r.content)
        shutil.unpack_archive(ffmpeg_zip, ffmpeg_dir,"zip")  
        shutil.move(ffmpeg_exe,ffmpeg_path)    
        shutil.rmtree(ffmpeg_dir)
        os.remove(ffmpeg_zip)

def get_audio_duration(file_path):
    sample_rate, data = wavfile.read(file_path)
    len_data = len(data)
    duration = int(len_data / sample_rate)
    return duration

def estimate_processing_time(duration, model):
    from torch import cuda
    device = "GPU" if cuda.is_available() else "CPU"
    match model, device: 
        case "tiny", "CPU": estimate = duration * 0.5
        case "base", "CPU": estimate = duration * 0.75
        case "small", "CPU": estimate = duration * 1
        case "medium", "CPU": estimate = duration * 2
        case "large-v1", "CPU": estimate = duration * 2.75
        case "large-v2", "CPU": estimate = duration * 3.5
        case "tiny", "GPU": estimate = duration * 0.1
        case "base", "GPU": estimate = duration * 0.125
        case "small", "GPU": estimate = duration * 0.15
        case "medium", "GPU": estimate = duration * 0.2
        case "large-v1", "GPU": estimate = duration * 0.25 
        case "large-v2", "GPU": estimate = duration * 0.3
    return estimate, device

def format_duration(time):
    seconds = int(time)
    hours = seconds // 3600  # Divide by 3600 to get the number of hours
    minutes = (seconds % 3600) // 60  # Divide the remaining seconds by 60 to get the number of minutes
    remaining_seconds = seconds % 60  # The remaining seconds
    formatted_time = f"{hours:02d}h {minutes:02d}m {remaining_seconds:02d}s"
    return formatted_time

if __name__ == "__main__":
    ...
utils/handle_upload.py
BANDAS-Center/aTrain
import os
import json 
import zipfile

def create_json_file(file_directory, outfile_name, content):
        output_file_text = os.path.join(file_directory,outfile_name)
        with open(output_file_text,"w", encoding="utf-8") as json_file:
            json.dump(content, json_file,ensure_ascii=False)

def create_txt_files (result, speaker_detection, file_directory, orig_filename):
        create_txt_file(result, file_directory, orig_filename, speaker_detection, maxqda=False, timestamps = False)
        create_txt_file(result, file_directory, orig_filename, speaker_detection, maxqda=False, timestamps = True)
        create_txt_file(result, file_directory, orig_filename, speaker_detection, maxqda=True, timestamps = True)

def create_txt_file (result,file_directory, orig_filename, speaker_detection, timestamps, maxqda):
    segments = result["segments"]
    match maxqda, timestamps:
         case True, _ :  filename = "transcription_maxqda.txt"
         case False, True: filename = "transcription_timestamps.txt"
         case False, False: filename = "transcription.txt"
    file_path = os.path.join(file_directory, filename)
    with open(file_path, "w",encoding="utf-8") as file:
        headline = f"Transcription for {orig_filename}" + ( "" if maxqda and speaker_detection else "\n") + ("" if speaker_detection else "\n" )
        file.write(headline)
        current_speaker = None
        for segment in segments:
            speaker = segment["speaker"] if "speaker" in segment else "Speaker undefined"
            if speaker != current_speaker and speaker_detection:
                file.write(("\n\n" if maxqda else "\n") + speaker + "\n")
                current_speaker = speaker
            text = str(segment["text"]).lstrip()
            if timestamps:
                time = segment["start"]
                formatted_time = format_time(time)
                text = f"{formatted_time} - {text}"
            file.write(text + (" " if maxqda else  "\n"))

def format_time(time):
    seconds = int(time)
    hours = seconds // 3600  # Divide by 3600 to get the number of hours
    minutes = (seconds % 3600) // 60  # Divide the remaining seconds by 60 to get the number of minutes
    remaining_seconds = seconds % 60  # The remaining seconds
    formatted_time = "[{:02d}:{:02d}:{:02d}]".format(hours, minutes, remaining_seconds)
    return formatted_time

def create_zip(directory_path, zip_file_path):
    with zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for filename in os.listdir(directory_path):
            file_path = os.path.join(directory_path, filename)
            zipf.write(file_path, arcname=filename)

if __name__ == "__main__":
    ...
utils/output_files.py
BANDAS-Center/aTrain
import os
from flask import render_template
from utils.use_custom_transcriber import use_transcriber
from utils.output_files import create_txt_files
import traceback
from utils.archive import read_metadata, delete_transcription, add_processing_time_to_metadata, TRANSCRIPT_DIR

def handle_transcription(file_id):
    try:
        filename, model, language, speaker_detection, num_speakers = read_metadata(file_id)
        file_directory = os.path.join(TRANSCRIPT_DIR,file_id)
        prepared_file = os.path.join(file_directory, file_id + ".wav")
        for step in use_transcriber(file_directory, prepared_file, model, language, speaker_detection, num_speakers):
            response = f"data: {step['task']}\n\n"
            yield response
        create_txt_files(step["result"], speaker_detection, file_directory, filename)
        add_processing_time_to_metadata(file_id)
        os.remove(prepared_file)
        html = render_template("modal_download.html", file_id=file_id).replace('\n', '')
        response = f"event: stopstream\ndata: {html}\n\n"
        yield response
    except Exception as e:
        delete_transcription(file_id)
        traceback_str = traceback.format_exc()
        error = str(e)
        html = render_template("modal_error.html", error=error, traceback=traceback_str).replace('\n', '')
        response = f"event: stopstream\ndata: {html}\n\n"
        yield response

if __name__ == "__main__":
    ...
    
utils/transcribe.py
BANDAS-Center/aTrain
import json
import os
import pandas as pd
from huggingface_hub import snapshot_download
from utils.output_files import create_json_file

def use_transcriber (file_directory, audio_file, model, language, speaker_detection, num_speakers):   
    import gc, torch #Import inside the function to speed up the startup time of the destkop app.
    from faster_whisper import WhisperModel
    from pyannote.audio import Pipeline
    from whisperx import load_audio, assign_word_speakers

    language = None if language == "auto-detect" else language
    min_speakers = max_speakers = None if num_speakers == "auto-detect" else int(num_speakers)
    device = "cuda" if torch.cuda.is_available() else "cpu"
    compute_type = "float16" if device == "cuda" else "int8"
    
    yield {"task":"Loading whisper model"}
    model_path = get_model(model)
    transcription_model = WhisperModel(model_path,device,compute_type=compute_type)

    yield {"task":"Transcribing file with whisper"}
    audio_array = load_audio(audio_file)
    transcription_segments, _ = transcription_model.transcribe(audio=audio_array,vad_filter=True, word_timestamps=True,language=language,no_speech_threshold=0.6)
    transcript = {"segments":[named_tuple_to_dict(segment) for segment in transcription_segments]}
    
    del transcription_model; gc.collect(); torch.cuda.empty_cache()
    
    if not speaker_detection:
        create_json_file(file_directory,outfile_name="transcription.json",content=transcript)
        yield {"task":"Finishing up", "result" : transcript}
    
    if speaker_detection:
        yield {"task":"Loading speaker detection model"}
        get_model("diarize")
        pipeline_config = os.path.join("models","config.yaml")
        diarize_model = Pipeline.from_pretrained(pipeline_config, cache_dir="models").to(device)
        
        yield {"task":"Detecting speakers"}
        diarization_segments = diarize_model(audio_file,min_speakers=min_speakers, max_speakers=max_speakers)
        speaker_results = transform_speakers_results(diarization_segments)
        del diarize_model; gc.collect(); torch.cuda.empty_cache()
        transcript_with_speaker = assign_word_speakers(speaker_results,transcript)
        create_json_file(file_directory,outfile_name="transcription.json",content=transcript_with_speaker)
        
        yield {"task":"Finishing up", "result":transcript_with_speaker}

def get_model(model):
    models_config_path = os.path.join("models","models.json")
    with open(models_config_path, "r") as models_config_file:
        models_config = json.load(models_config_file)
    model_info = models_config[model]
    model_path = os.path.join(*model_info["path"])
    if not os.path.exists(model_path):
        snapshot_download(repo_id=model_info["repo_id"],revision=model_info["revision"],cache_dir="models")
    return model_path

def transform_speakers_results(diarization_segments):    
    diarize_df = pd.DataFrame(diarization_segments.itertracks(yield_label=True))
    diarize_df['start'] = diarize_df[0].apply(lambda x: x.start)
    diarize_df['end'] = diarize_df[0].apply(lambda x: x.end)
    diarize_df.rename(columns={2: "speaker"}, inplace=True)
    return diarize_df

def named_tuple_to_dict(obj):
    if isinstance(obj, dict):
        return {key: named_tuple_to_dict(value) for key, value in obj.items()}
    elif isinstance(obj, list):
        return [named_tuple_to_dict(value) for value in obj]
    elif isnamedtupleinstance(obj):
        return {key: named_tuple_to_dict(value) for key, value in obj._asdict().items()}
    elif isinstance(obj, tuple):
        return tuple(named_tuple_to_dict(value) for value in obj)
    else:
        return obj

def isnamedtupleinstance(x):
    _type = type(x)
    bases = _type.__bases__
    if len(bases) != 1 or bases[0] != tuple:
        return False
    fields = getattr(_type, '_fields', None)
    if not isinstance(fields, tuple):
        return False
    return all(type(i)==str for i in fields)

if __name__ == "__main__":
    ...
utils/use_custom_transcriber.py
BANDAS-Center/aTrain