import datetime import os from typing import Dict, List, Tuple from uuid import UUID import altair as alt from apscheduler.schedulers.background import BackgroundScheduler import argilla as rg from argilla.feedback import FeedbackDataset from huggingface_hub import restart_space from argilla.client.feedback.dataset.remote.dataset import RemoteFeedbackDataset import gradio as gr import pandas as pd # Translation of legends and titels ANNOTATED = "Annotated" NUMBER_ANNOTATED = "Total Annotations" NUMBER_ANNOTATORS = "Total Annotators" PENDING = "Pending Annotations" NAME = "Username" CATEGORY = "Category" SUPPORTED_LANGUAGES = [ "Russian", "Dutch", "Vietnamese", "Arabic", "Filipino", "German", "Swahili", "Malagasy", "Czech", # "Tamil", # "Telugu", "Hungarian" ] def restart() -> None: """ This function restarts the space where the dashboard is hosted. """ # Update Space name with your Space information gr.Info("Restarting space at " + str(datetime.datetime.now())) restart_space( "DIBT/PromptTranslationMultilingualDashboard", token=os.getenv("HF_TOKEN"), # factory_reboot=True, ) def get_user_annotations_dictionary( dataset: FeedbackDataset | RemoteFeedbackDataset, ) -> Dict[str, int]: """ This function returns a dictionary with the username as the key and the number of annotations as the value. Args: dataset: The dataset to be analyzed. Returns: A dictionary with the username as the key and the number of annotations as the value. """ output = {} for record in dataset: for response in record.responses: if str(response.user_id) not in output.keys(): output[str(response.user_id)] = 1 else: output[str(response.user_id)] += 1 # Changing the name of the keys, from the id to the username for key in list(output.keys()): output[rg.User.from_id(UUID(key)).username] = output.pop(key) return output def fetch_data() -> Tuple[Dict[str, int], Dict[str, dict]]: """ This function fetches the data from all the datasets and stores the annotation information in two dictionaries. To do so, looks for all the environment variables that follow this pattern: - SPANISH_API_URL - SPANISH_API_KEY - SPANISH_DATASET - SPANISH_WORKSPACE If the language name matches with one of the languages present in our SUPPORTED_LANGUAGES list, it will fetch the data with the total amount of annotations and the total annotators. Returns: Tuple[Dict[str, int], Dict[str, dict]]: A tuple with two dictionaries. The first one contains the total amount of annotations for each language. The second one contains the total annotators for each language. """ print(f"Starting to fetch data: {datetime.datetime.now()}") # Obtain all the environment variables environment_variables_languages = {} for language in SUPPORTED_LANGUAGES: print("Fetching data for: ", language) if not os.getenv(f"{language.upper()}_API_URL"): print(f"Missing environment variables for {language}") continue environment_variables_languages[language] = { "api_url": os.getenv(f"{language.upper()}_API_URL"), "api_key": os.getenv(f"{language.upper()}_API_KEY"), "dataset_name": os.getenv(f"{language.upper()}_DATASET"), "workspace_name": os.getenv(f"{language.upper()}_WORKSPACE"), } global annotations, annotators annotations = {} annotators = {} # Connect to each space and obtain the total amount of annotations and annotators for language, environment_variables in environment_variables_languages.items(): rg.init( api_url=environment_variables["api_url"], api_key=environment_variables["api_key"], ) # Obtain the dataset and see how many pending records are there dataset = rg.FeedbackDataset.from_argilla( environment_variables["dataset_name"], workspace=environment_variables["workspace_name"], ) # filtered_source_dataset = source_dataset.filter_by(response_status=["pending"]) target_dataset = dataset.filter_by(response_status=["submitted"]) annotations[language.lower()] = len(target_dataset) annotators[language.lower()] = { "annotators": get_user_annotations_dictionary(target_dataset) } # Print the current date and time print(f"Data fetched: {datetime.datetime.now()}") return annotations, annotators def kpi_chart_total_annotations() -> alt.Chart: """ This function returns a KPI chart with the total amount of annotations. Returns: An altair chart with the KPI chart. """ total_annotations = 0 for language in annotations.keys(): total_annotations += annotations[language] # Assuming you have a DataFrame with user data, create a sample DataFrame data = pd.DataFrame({"Category": [NUMBER_ANNOTATED], "Value": [total_annotations]}) # Create Altair chart chart = ( alt.Chart(data) .mark_text(fontSize=100, align="center", baseline="middle", color="#e68b39") .encode(text="Value:N") .properties(title=NUMBER_ANNOTATED, width=250, height=200) ) return chart def kpi_chart_total_annotators() -> alt.Chart: """ This function returns a KPI chart with the total amount of annotators. Returns: An altair chart with the KPI chart. """ total_annotators = 0 for _, value in annotators.items(): total_annotators += len(list(value["annotators"].keys())) # Assuming you have a DataFrame with user data, create a sample DataFrame data = pd.DataFrame({"Category": [NUMBER_ANNOTATORS], "Value": [total_annotators]}) # Create Altair chart chart = ( alt.Chart(data) .mark_text(fontSize=100, align="center", baseline="middle", color="#e68b39") .encode(text="Value:N") .properties(title=NUMBER_ANNOTATORS, width=250, height=200) ) return chart def render_hub_user_link(hub_id: str) -> str: """ This function returns a link to the user's profile on Hugging Face. Args: hub_id: The user's id on Hugging Face. Returns: A string with the link to the user's profile on Hugging Face. """ link = f"https://huggingface.co/{hub_id}" return f'{hub_id}' def obtain_top_users(user_annotators_list: Dict[str, int], N: int = 50) -> pd.DataFrame: """ This function returns the top N users with the most annotations. Args: user_ids_annotations: A dictionary with the user ids as the key and the number of annotations as the value. Returns: A pandas dataframe with the top N users with the most annotations. """ user_id_annotations = {} for _, user_annotators in user_annotators_list.items(): for user_id, number_annotations in user_annotators["annotators"].items(): if user_id not in user_id_annotations: user_id_annotations[user_id] = number_annotations else: user_id_annotations[user_id] += number_annotations dataframe = pd.DataFrame( user_id_annotations.items(), columns=[NAME, NUMBER_ANNOTATED] ) dataframe[NAME] = dataframe[NAME].apply(render_hub_user_link) dataframe = dataframe.sort_values(by=NUMBER_ANNOTATED, ascending=False) return dataframe.head(N) def get_top(N=50) -> pd.DataFrame: """ This function returns the top N users with the most annotations. Args: N: The number of users to be returned. 50 by default Returns: A pandas dataframe with the top N users with the most annotations. """ return obtain_top_users(annotators, N=N) def donut_chart_total() -> alt.Chart: """ This function returns a donut chart with the progress of the total annotations in each language. Returns: An altair chart with the donut chart. """ # Load your data annotated_records = [annotation for annotation in annotations.values()] languages = [language.capitalize() for language in annotations.keys()] # Prepare data for the donut chart source = pd.DataFrame( { "values": annotated_records, "category": languages, # "colors": ["#4682b4", "#e68c39"], # Blue for Completed, Orange for Remaining } ) base = alt.Chart(source).encode( theta=alt.Theta("values:Q", stack=True), radius=alt.Radius( "values", scale=alt.Scale(type="sqrt", zero=True, rangeMin=20) ), color=alt.Color( field="category", type="nominal", legend=alt.Legend(title=CATEGORY), ), ) c1 = base.mark_arc(innerRadius=20, stroke="#fff") c2 = base.mark_text(radiusOffset=20).encode(text="values:Q") chart = c1 + c2 return chart def bar_chart_total() -> alt.Chart: """A bar chart with the progress of the total annotations in each language. Returns: An altair chart with the bar chart. """ # Load your data annotated_records = [annotation for annotation in annotations.values()] languages = [language.capitalize() for language in annotations.keys()] # Prepare data for the bar chart source = pd.DataFrame( { "values": annotated_records, "category": languages, } ) base = alt.Chart(source, width=300, height=200).encode( x=alt.X("values:Q", title="Translations"), y=alt.Y("category:N", title="Language"), text="values:Q", color=alt.Color("category:N", legend=None), ) rule = alt.Chart(source).mark_rule(color="red").encode(x=alt.datum(500)) return base.mark_bar() + base.mark_text(align="left", dx=2) + rule def main() -> None: fetch_data() # To avoid the orange border for the Gradio elements that are in constant loading css = """ .generating { border: none; } """ with gr.Blocks(css=css, delete_cache=(300, 300)) as demo: gr.Markdown( """ # 🌍 Translation Efforts Dashboard - Multilingual Prompt Evaluation Project You can check out the progress done in each language for the Multilingual Prompt Evaluation Project in this dashboard. If you want to add a new language to this dashboard, please open an issue and we will contact you to obtain the necessary API KEYs and URLs include your language in this dashboard. ## How to participate Participating is easy. Go to one of the Annotation Spaces of the language of your choice, log in or create a Hugging Face account, and you can start working. - [Spanish](https://somosnlp-dibt-prompt-translation-for-es.hf.space) - [Russian](https://dibt-russian-prompt-translation-for-russian.hf.space) - [Dutch](https://dibt-dutch-prompt-translation-for-dutch.hf.space) - [Vietnamese](https://ai-vietnam-prompt-translation-for-vie.hf.space) - [Arabic](https://2a2i-prompt-translation-for-arabic.hf.space) - [Filipino](https://dibt-filipino-prompt-translation-for-filipino.hf.space) - [German](https://huggingface.co/spaces/DIBT-German/prompt-translation-for-German) - [Swahili](https://dibt-swahili-prompt-translation-for-swahili.hf.space) - [Malagasy](https://dibt-malagasy-prompt-translation-for-malagasy.hf.space) - [Tamil](https://data-indica-prompt-translation-for-tamil.hf.space) - [Telugu](https://data-indica-prompt-translation-for-telugu.hf.space) - [Czech](https://dibt-czech-prompt-translation-for-czech.hf.space) - [Hungarian](https://dibt-hungarian-prompt-translation-for-hungarian.hf.space) """ ) gr.Markdown( f""" ## 🚀 Annotations among Languages Here you can see the progress of the annotations among the different languages. """ ) with gr.Row(): kpi_chart_annotations = gr.Plot(label="Plot") demo.load( kpi_chart_total_annotations, inputs=[], outputs=[kpi_chart_annotations], ) bar_languages = gr.Plot(label="Plot") demo.load( bar_chart_total, inputs=[], outputs=[bar_languages], ) gr.Markdown( """ ## 👾 Hall of Fame Check out the users with more contributions among the different translation efforts. """ ) with gr.Row(): kpi_chart_annotators = gr.Plot(label="Plot") demo.load( kpi_chart_total_annotators, inputs=[], outputs=[kpi_chart_annotators], ) top_df_plot = gr.Dataframe( headers=[NAME, NUMBER_ANNOTATED], datatype=[ "markdown", "number", ], row_count=50, col_count=(2, "fixed"), interactive=False, ) demo.load(get_top, None, [top_df_plot]) # Manage background refresh scheduler = BackgroundScheduler() _ = scheduler.add_job(restart, "interval", minutes=30) scheduler.start() # Launch the Gradio interface demo.launch() if __name__ == "__main__": main()