app.py

from __future__ import annotations

import logging
import os
import re
from functools import lru_cache
from typing import Any

import gradio as gr
import librosa
import numpy as np
import pandas as pd
import soundfile as sf
import torch
from datasets import Dataset, DatasetDict, load_dataset
from gradio.themes.base import Base
from gradio_client import Client, handle_file

LOGGER = logging.getLogger("audio-dataset-manager")
logging.basicConfig(
    level=os.getenv("LOG_LEVEL", "INFO"),
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)

TARGET_SAMPLE_RATE = 16_000
HF_TOKEN_ENV = "DataBase"
LOGIN_TOKEN_ENV = "DataBase"


class DeepFilterNetUnavailable(RuntimeError):
    """Raised when DeepFilterNet is not installed or cannot be initialized."""


@lru_cache(maxsize=1)
def get_enhancer():
    """
    Load DeepFilterNet lazily.
    Lazy loading prevents the whole Space from crashing during startup and
    avoids loading the model until enhancement is actually requested.
    """
    try:
        from df.enhance import enhance, init_df
    except ModuleNotFoundError as exc:
        raise DeepFilterNetUnavailable(
            "DeepFilterNet is not installed. Add `deepfilternet==0.5.6` "
            "to requirements.txt, then rebuild the Space."
        ) from exc

    try:
        model, state, _ = init_df()
        model.eval()
        return enhance, model, state
    except Exception as exc:
        LOGGER.exception("DeepFilterNet initialization failed")
        raise DeepFilterNetUnavailable(
            f"DeepFilterNet could not be initialized: {exc}"
        ) from exc


def _as_mono_float32(audio: Any) -> np.ndarray:
    """Convert supported audio input to finite mono float32 samples."""
    array = np.asarray(audio, dtype=np.float32)

    if array.ndim == 2:
        # Gradio may return (samples, channels).
        array = array.mean(axis=1)
    elif array.ndim != 1:
        array = array.reshape(-1)

    array = np.nan_to_num(array, nan=0.0, posinf=0.0, neginf=0.0)
    peak = float(np.max(np.abs(array))) if array.size else 0.0
    if peak > 1.0:
        array = array / peak
    return np.ascontiguousarray(array, dtype=np.float32)


def normalize_audio(audio: Any, eps: float = 1e-8) -> np.ndarray:
    array = _as_mono_float32(audio)
    if array.size == 0:
        return array

    peak = float(np.max(np.abs(array)))
    return array if peak < eps else array / peak


def read_dataset(link: str) -> pd.DataFrame:
    if not link or not link.strip():
        raise gr.Error("أدخل رابط أو اسم Dataset صحيح.")

    token = os.getenv(HF_TOKEN_ENV) or None
    try:
        dataset = load_dataset(link.strip(), token=token)
    except Exception as exc:
        LOGGER.exception("Dataset loading failed: %s", link)
        raise gr.Error(f"تعذر تحميل Dataset: {exc}") from exc

    split = "train" if "train" in dataset else next(iter(dataset.keys()))
    frame = dataset[split].to_pandas()

    required = {
        "text": "",
        "audio": None,
        "samplerate": TARGET_SAMPLE_RATE,
        "secs": 0.0,
        "speaker_id": -1,
        "_speaker_id": -1,
        "flag": 0,
    }
    for column, default in required.items():
        if column not in frame.columns:
            frame[column] = default

    return frame


# Backward-compatible name used by the original callbacks.
Read_DataSet = read_dataset


def remove_nn(wav: Any, sample_rate: int = TARGET_SAMPLE_RATE):
    """
    Enhance a waveform with DeepFilterNet and return a Gradio audio tuple.
    """
    if wav is None:
        raise gr.Error("اختر ملفًا صوتيًا أولًا.")

    audio = normalize_audio(wav)
    if audio.size == 0:
        raise gr.Error("الملف الصوتي فارغ.")

    enhance_fn, model, state = get_enhancer()
    model_sr = int(state.sr())

    if sample_rate != model_sr:
        audio = librosa.resample(
            audio,
            orig_sr=int(sample_rate),
            target_sr=model_sr,
            res_type="soxr_hq",
        )

    tensor = torch.from_numpy(audio).unsqueeze(0)

    with torch.inference_mode():
        enhanced = enhance_fn(model, state, tensor)

    enhanced_np = enhanced.squeeze(0).detach().cpu().numpy()

    if model_sr != TARGET_SAMPLE_RATE:
        enhanced_np = librosa.resample(
            enhanced_np,
            orig_sr=model_sr,
            target_sr=TARGET_SAMPLE_RATE,
            res_type="soxr_hq",
        )

    return TARGET_SAMPLE_RATE, normalize_audio(enhanced_np)



class DataViewerApp:
    def __init__(self,df):
        #df=read_dataset(link)
        self.df=df
       # self.df1=df
        self.data =self.df[['text','speaker_id','secs','flag']]
        self.dataa =self.df[['text','speaker_id','secs','flag']]
        self.sdata =self.df['audio'].to_list()   # Separate audio data storage
        self.current_page = 0
        self.current_selected = -1
        self.speaker_id= -1
        class Seafoam(Base):
            pass
        self.seafoam = Seafoam()

        #self.data =df[['text','speaker_id']]
        #self.sdata = df['audio'].to_list()  # Separate audio data storage
        #self.current_page = 0
        #self.current_selected = -1
    def set1(self,df):
        self.data =df[['text','speaker_id','secs','flag']]
        self.sdata =df['audio'].to_list()
        return self.get_page_data(self.current_page)
    def settt(self,df):
        self.df=pd.DataFrame()
        self.data =pd.DataFrame()
        self.sdata =[]
        self.df=df
        self.data =df[['text','speaker_id','secs','flag']]
        self.dataa =df[['text','speaker_id','secs','flag']]
        self.sdata =df['audio'].to_list()
        self.current_page = 0
        self.current_selected =1
        self.speaker_id= -1
        return self.data
    def clear(self,text):
        text=re.sub(r'[a-zA-Z]', '', text)
        return text
    def clearenglish(self):
         for i in range(len(self.df)):
             x=self.clear(self.df['text'][i])
             x1=self.df['text'][i]
             if x!=x1:
                self.df.drop(i, inplace=True)

         self.df.reset_index(drop=True, inplace=True)
         return self.settt(self.df)
    def splitt(self,link,num):
        df=download_youtube_video(link,num)
        v=self.settt(df)
        return self.get_page_data(self.current_page),len(v)
    def getdataset(self,link):
        self.link_dataset=link
        df=read_dataset(link)
        v=self.settt(df)
        return self.get_page_data(self.current_page),len(v),self.link_dataset
    def remove_hamza_from_alif_and_symbols(self,text):
            text = re.sub(r"[أإآ]", "ا", text)
            text = re.sub(r"ٱ", "ا", text)
            text = re.sub(r"[_\-\+\,\(\)]", " ", text)
            text = re.sub(r"\d", " ", text)
            return text
    def save_row(self, text,data_oudio):
        if text!="" :
                row = self.data.iloc[self.current_selected]
                row['text'] = text
                row['flag']=1
                self.data.iloc[self.current_selected] = row
                sr,audio=data_oudio
                if sr!=16000:
                    audio=audio.astype(np.float32)
                    audio=normalize_audio(audio)
                    audio=librosa.resample(audio,orig_sr=sr,target_sr=16000)
                     
                     
                         
                      
        
                self.sdata[self.current_selected] = audio
                self.df.loc[self.current_selected, 'text'] = text
                self.df.at[self.current_selected, 'audio'] = audio
                self.df.loc[self.current_selected, 'flag'] = 1
        return self.get_page_data(self.current_page),None,""
    def GetDataset_2(self,filename,ds=1.5):
        audios_data = []
        audios_samplerate = []
        num_specker=[]
        texts=[]
        secs=[]

        audiodata,samplerate = librosa.load(filename, sr=16000) # Removed extra indent here
        audios_data.append(audiodata*ds)
        audios_samplerate.append(samplerate)
        texts.append(filename.replace('.wav',''))
        secs.append(round(len(audiodata)/samplerate,2))
        df = pd.DataFrame()
        df['secs'] = secs
        df['audio'] = audios_data
        df['samplerate'] = audios_samplerate
        df['text'] =os.path.splitext(os.path.basename(filename))[0]
        df['speaker_id'] =self.speaker_id
        df['_speaker_id'] =self.speaker_id
        df['flag']=1
        df = df[['text','audio','samplerate','secs','speaker_id','_speaker_id','flag']]
        self.df = pd.concat([self.df, df], axis=0, ignore_index=True)
        self.data =self.df[['text','speaker_id','secs','flag']]
        self.sdata =self.df['audio'].to_list()

        return self.get_page_data(self.current_page)
    def trim_audio(self, text,data_oudio):
        if text!="" :
                audios_data = []
                audios_samplerate = []
                sr,audio=data_oudio
                audio=audio.astype(np.float32)
                audio=normalize_audio(audio)
                audio=librosa.resample(audio,orig_sr=sr,target_sr=16000)
                audios_data.append(audio)
                secs=round(len(audio)/TARGET_SAMPLE_RATE, 2)
                audios_samplerate.append(16000)
                df = pd.DataFrame()
                df['secs'] = secs
                df['audio'] =[ audio]
                df['samplerate'] = 16000
                df['text'] =text
                df['speaker_id'] =self.speaker_id
                df['_speaker_id'] =self.speaker_id
                df['flag']=1
                df = df[['text','audio','samplerate','secs','speaker_id','_speaker_id','flag']]
                self.df = pd.concat([self.df, df], axis=0, ignore_index=True)
                self.data =self.df[['text','speaker_id','secs','flag']]
                self.sdata =self.df['audio'].to_list()
        return self.get_page_data(self.current_page),None,"" 
    def order_data(self):
        self.df[['text','speaker_id','secs','flag']]=self.data
        self.df=self.df.sort_values(by=['flag'], ascending=False)
        vv=self.settt(self.df)
        return vv
    def connect_drive(self):
        from google.colab import drive
        drive.mount('/content/drive')
    def get_page_data(self, page_number):
        start_index = page_number * 10
        end_index = start_index + 10
        return self.data.iloc[start_index:end_index]
    def update_page(self, new_page):
        self.current_page = new_page
        return (
            self.get_page_data(self.current_page),
            self.current_page > 0,
            self.current_page < len(self.data) // 10 - 1,
            self.current_page
        )
    def clear_txt(self):
        self.data['text'] =self.data['text'].apply(self.remove_hamza_from_alif_and_symbols)
        return self.get_page_data(self.current_page)
    def  get_text_from_audio(self,audio):
         if len(audio)!=0:
             sf.write("temp.wav", normalize_audio(audio), TARGET_SAMPLE_RATE, format='WAV')
    
             client = Client("MohamedRashad/Arabic-Whisper-CodeSwitching-Edition")
             result = client.predict(
             inputs=handle_file('temp.wav'),
             api_name="/predict_1"
        )
             return result
         else:
            return ""
    
    def on_column_dropdown_change_operater(self,selected_column,selected_column1):
        if selected_column1==">":
            return self.data[self.data['secs'] > selected_column ]
        elif selected_column1=="<":
            return self.data[self.data['secs'] < selected_column]
        elif selected_column1=="=":
            return self.data[self.data['secs'] == selected_column]
        else:
            return self.data
            # Perform actions based on the selected column

    def on_column_dropdown_change(self,selected_column):
        data=self.df
        if selected_column=="all":

            return self.set1(data),len(data)
        elif selected_column=="0":
             data=data[data['flag'] ==0]
             return self.set1(data),len(data)
        else :
            data=data[data['flag'] ==1]
            return self.set1(data),len(data)

    def on_select(self,evt:gr.SelectData):
        index_now = evt.index[0]
        self.current_selected = (self.current_page * 10) + index_now
        row = self.data.iloc[self.current_selected]
        row_audio = self.sdata[self.current_selected]
        self.speaker_id=row['speaker_id']
        return (TARGET_SAMPLE_RATE, row_audio), row['text']
    def finsh_data(self):
        self.df['audio'] = self.sdata
        self.df[['text','speaker_id','secs','flag']]=self.data

        return self.df
    def All_enhance(self):
        for i in range(0,len(self.sdata)):
              _,y=remove_nn(self.sdata[i])
              self.sdata[i]=y
        return self.data

        return self.get_page_data(self.current_page)
    def get_output_audio(self):
        return self.sdata[self.current_selected] if self.current_selected >= 0 else None
    def Convert_DataFreme_To_DataSet(self,namedata):
           df=self.df

           df['audio'] = df['audio'].apply(lambda x: np.array(x, dtype=np.float32))
           if "__index_level_0__" in df.columns:
                df =df.drop(columns=["__index_level_0__"])
           train_df =df

         

           ds = {
                "train": Dataset.from_pandas(train_df)
                
                 }

           dataset = DatasetDict(ds)
           dataset.push_to_hub(namedata, token=os.environ.get(HF_TOKEN_ENV), private=True)
           return namedata

    def delete_row(self):
        if len(self.data) != 0 and self.current_selected != -1:
                self.data.drop(self.current_selected, inplace=True)
                self.data.reset_index(drop=True, inplace=True)
                self.df.drop(self.current_selected, inplace=True)
                self.df.reset_index(drop=True, inplace=True)
                self.sdata.pop(self.current_selected)
                self.current_selected = -1
                # self.audio_player.update(None)  # Clear audio player
                # self.txt_audio.update("")  # Clear text input

        return self.get_page_data(self.current_page),None,""
    def login(self, token):
        # Your actual login logic here (e.g., database check)
        if token and token == os.environ.get(LOGIN_TOKEN_ENV):
            return gr.update(visible=False),gr.update(visible=True),True
        else:
            return gr.update(visible=True), gr.update(visible=False),None
    def load_demo(self,sesion):
        if sesion:
            return  gr.update(visible=False),gr.update(visible=True)
        
        return gr.update(visible=True), gr.update(visible=False)
    def start_tab1(self):
        with gr.Blocks(theme=self.seafoam, css="""
         table.svelte-82jkx.svelte-82jkx{
              font-size: x-small;
        }
    .checkbox-group label {
        background-color: #f0f0f5; /* لون خلفية فاتح */
        padding: 10px;
        border-radius: 5px; /* زوايا دائرية */
    }
        const textbox = document.querySelector('.txt_audio'); // تحديد المكون النصي
        textbox.style.direction = 'ltr';
    .checkbox-group input:checked + label {
        background-color: #e0f0ff; /* لون خلفية عند التحديد */
        font-weight: bold;
    }
""") as demo:
            sesion_state = gr.State()
            
            with gr.Column(scale=1, min_width=200,visible=True) as login_panal:  # Login panel
                gr.Markdown("## auth acess page")
                token_login = gr.Textbox(label="token")
              
                login_button = gr.Button("Login")
            with gr.Column(scale=1, visible=False) as main_panel:
                with gr.Row(equal_height=False):
                      with gr.Tabs():
                          with gr.TabItem("Processing Data  "):
                              self.data_Processing()
            login_button.click(self.login, inputs=[token_login], outputs=[login_panal,main_panel,sesion_state])
            demo.load(self.load_demo, [sesion_state], [login_panal,main_panel])


        return demo
    def  create_Tabs(self): # fix: method was missing
        #with gr.Blocks() as interface:
             with gr.Tabs():
                  with gr.TabItem("Excel"):
                       with gr.Row():
                           txt_filepath_excel=gr.Text("NameFile")
                           txt_text_excel=gr.Text("Text" )
                           but_send_excel=gr.Button("Send",size="sm")

                  with gr.TabItem("CVC"):
                          with gr.Row():
                           txt_filepath_cvc=gr.Text("File")
                           txt_text_cvc=gr.Text("Text" )
                           but_send_cvc=gr.Button("Send",size="sm")
                  with gr.TabItem("DateSet"):
                           self.txt_filepath_dir=gr.Text(placeholder="link dir",interactive=True)
                           #self.txt_text=gr.Text("Text" )
                           self.but_send_dir=gr.Button("Send",size="sm")
                  with gr.TabItem("Dir"):
                        txt_filepath_dateSet=gr.Text("link DateSet")
                           #self.txt_text=gr.Text("Text" )
                        but_send_dateSet=gr.Button("Send",size="sm")
                  with gr.TabItem("Cut  Video"):
                       self.txt_filepath_dateSet=gr.Text("رابط الفيديو",interactive=True)
                       self.num = gr.Number(label=" ادخل رقم طبيعي")

                       self.but_send_dateSet_cut=gr.Button("Send",size="sm")

    def Convert_DataFrame_to_Bitch(self):
        with gr.Row():
                           self.txt_output_dir=gr.Text("output Name dir",interactive=True)
                           self.txt_train_batch_size=gr.Text("train_batch_size",interactive=True)
                           self.txt_eval_batch_size=gr.Text("eval_batch_size",interactive=True )
                           self.but_convert_bitch=gr.Button("Convert Bitch",size="sm")
        with gr.Row():
                           self.label_Bitch=gr.Label("Dir Output Bitch :")


    def data_Processing(self):

         #with gr.Column(scale=2,min_width=40):

             #with gr.Row():
                    #with gr.Accordion("Open Data", open=False):
                         #with gr.Row():
                            #  self.txt_filepath_dateSet=gr.Text("link DateSet",interactive=True)
                              #self.txt_text=gr.Text("Text" )
                              #self.but_send_dateSet=gr.Button("Send",size="sm")


         with gr.Accordion("Install Data", open=False):
                with gr.Row():
                              self.create_Tabs()
                with gr.Row():
                             columns = []
                             columns1 = []

                             columns =["all","0","1"]
                             columns.append("all")
                             self.labell=gr.Label("count:")
                             self.column_dropdown = gr.Dropdown(choices=columns, label="speaker_id")
                             with gr.Row():

                                  columns1=unique_speaker_ids =self.df['secs'].unique().tolist()
                                  columns1.append("all")
                                  self.column_dropdown1 = gr.Dropdown(choices=columns1 , label="secs")

                                  self.column_dropdown11 = gr.Dropdown(choices=["all","<",">","="], label="operater")


         with gr.Row():


                   with gr.Column(scale=5):
                        gr.Markdown("## Data Viewer")
                        #d=self.get_page_data(self.current_page)
                        # Correct the indentation here:
                        self.data_table = gr.DataFrame(  # Notice 'self.' here
                            value=self.get_page_data(self.current_page),
                            headers=["Text","speaker_id"])

                   # interactive=True

                        #self.data_table1 = gr.DataFrame(headers=[ "Text","Id_spiker"])
                        with gr.Row(equal_height=False):
                            self.prev_button = gr.Button("<",scale=1, size="sm",min_width=30)
                         
                            self.page_number = gr.Number(value=self.current_page + 1, label="Page",scale=1,min_width=100)
                            self.next_button = gr.Button(">",scale=1, size="sm",min_width=30)

                        with gr.Row(equal_height=False):

                             #inputs=gr.CheckboxGroup(["John", "Mary", "Peter", "Susan"])
                             self.but_cleartxt=gr.Button("clear Text",variant="primary",size="sm",min_width=30)
                             self.btn_all_enhance=gr.Button("All enhance",size="sm",variant="primary",min_width=30)
                             self.btn_ClearEnglish=gr.Button("ClearEnglish",size="sm",variant="primary",min_width=30)










                   with gr.Column(scale=4):
                            gr.Markdown("## Row Data")
                            self.txt_audio = gr.Textbox(label="Text", interactive=True,rtl=True)
                            with gr.Row(equal_height=False):
                                self.audio_player = gr.Audio(label="Audio")
                            with gr.Row(equal_height=False):
                                  self.btn_del = gr.Button("Delete  ", size="sm",variant="primary",min_width=50)
                                  self.btn_save = gr.Button("Save", size="sm",variant="primary",min_width=50)
                                  self.totext=gr.Button("to text",size="sm" ,variant="primary",min_width=50)
                                  
                           # with gr.Row(equal_height=False):
                                 
                                 
                            with gr.Row(equal_height=False):
                                  self.btn_newsave=gr.Button("New Save Cut",size="sm",variant="primary",min_width=50)
                                  self.btn_enhance = gr.Button("enhance ", size="sm",variant="primary",min_width=50)
                                  self.order= gr.Button("order ", size="sm",variant="primary",min_width=50)


         with gr.Row(equal_height=False,variant="heading-1"):
              with gr.Accordion("Save Bitch", open=False):

                                       self.txt_dataset=gr.Text("save dataset",interactive=True)
                                       self.btn_convertDataset=gr.Button("Dir Output Bitch :",variant="primary")
                                       self.label_dataset=gr.Label("count:")
         self.order.click(self.order_data,[],[self.data_table])
         self.btn_ClearEnglish.click(self.clearenglish,[],[self.data_table])
         self.but_send_dir.click(self.getdataset, [self.txt_filepath_dir],[self.data_table,self.labell,self.txt_dataset])
         #self.but_send_dateSet_cut.click(self.splitt, [self.txt_filepath_dateSet,self.num],[self.data_table,self.labell])
         #self.txt_audio.Style(container=False, css=".txt_audio { direction: rtl; }")
         #self.but_send_dateSet.click(self.Read_DataSet, [self.txt_filepath_dateSet],[self.data_table ])
         self.data_table.select(self.on_select, None, [self.audio_player, self.txt_audio])
         self.prev_button.click(lambda page: self.update_page(page - 1), [self.page_number], [self.data_table, self.prev_button, self.next_button, self.page_number])
                   #self.btn_save.click(self.save_row, [self.txt_audio,self.audio_player], [self.data_table])
         self.next_button.click(lambda page: self.update_page(page + 1), [self.page_number], [self.data_table, self.prev_button, self.next_button, self.page_number])
         self.column_dropdown.change(self.on_column_dropdown_change,[self.column_dropdown], [self.data_table,self.labell])
         self.column_dropdown11.change(self.on_column_dropdown_change_operater,[self.column_dropdown1,self.column_dropdown11], [self.data_table])
         self.btn_convertDataset.click(self.Convert_DataFreme_To_DataSet,[self.txt_dataset],[self.label_dataset])
         self.totext.click(lambda:self.get_text_from_audio(self.get_output_audio()), [], self.txt_audio)
         self.btn_newsave.click(self.trim_audio,[self.txt_audio,self.audio_player],[self.data_table,self.audio_player,self.txt_audio])
         self.btn_save.click(self.save_row, [self.txt_audio,self.audio_player], [self.data_table,self.audio_player,self.txt_audio])
         #self.btn_save.click(self.save_row, [self.txt_audio,self.audio_player], [self.data_table])
         self.btn_all_enhance.click(self.All_enhance,[],[self.data_table])
         #self.btn_enhance.click(remove_nn, [self.audio_player], [self.audio_player])
         self.but_cleartxt.click(self.clear_txt,[],[self.data_table])
         self.btn_del.click(self.delete_row,[], [self.data_table,self.audio_player,self.txt_audio])
         self.btn_enhance.click(lambda: remove_nn(self.get_output_audio()), [], self.audio_player)
         #self.column_dropdown.change(lambda selected_column:self.settt(self.on_column_dropdown_change(selected_column)), [self.column_dropdown], [self.data_table])
         #self.column_dropdown.change(lambda selected_column:self.settt(x.on_column_dropdown_change(selected_column)), [x.column_dropdown], [self.data_table])
         #self.btn_denoise.click(self.remove_nn, [self.audio_player], [self.audio_player])



dff=pd.DataFrame(columns=['text', 'audio', 'samplerate', 'secs', 'speaker_id', '_speaker_id','flag'])
app=DataViewerApp(dff)
s=app.start_tab1()

s.queue(default_concurrency_limit=2).launch(debug=True,show_error=True, ssr_mode=False)