import datetime import os from typing import Dict, Tuple from uuid import UUID import altair as alt import argilla as rg from argilla.feedback import FeedbackDataset from argilla.client.feedback.dataset.remote.dataset import RemoteFeedbackDataset import gradio as gr import pandas as pd # ترجمة الأساطير والعناوين ANNOTATED = "التعليقات المُضافة" NUMBER_ANNOTATED = "إجمالي التعليقات المُضافة" PENDING = "قيد الانتظار" NUMBER_ANNOTATORS = "عدد المُعلقين" NAME = "اسم المستخدم" NUMBER_ANNOTATIONS = "عدد التعليقات" CATEGORY = "الفئة" def obtain_source_target_datasets() -> ( Tuple[ FeedbackDataset | RemoteFeedbackDataset, FeedbackDataset | RemoteFeedbackDataset ] ): """ This function returns the source and target datasets to be used in the application. Returns: A tuple with the source and target datasets. The source dataset is filtered by the response status 'pending'. """ # Obtain the public dataset and see how many pending records are there source_dataset = rg.FeedbackDataset.from_argilla( "DIBT Translation for Arabic", workspace=os.getenv("SOURCE_WORKSPACE") ) filtered_source_dataset = source_dataset.filter_by(response_status=["pending"]) # Obtain a list of users from the private workspace # target_dataset = rg.FeedbackDataset.from_argilla( # os.getenv("RESULTS_DATASET"), workspace=os.getenv("RESULTS_WORKSPACE") # ) target_dataset = source_dataset.filter_by(response_status=["submitted"]) return filtered_source_dataset, target_dataset 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 donut_chart_total() -> alt.Chart: """ This function returns a donut chart with the progress of the total annotations. Counts each record that has been annotated at least once. Returns: An altair chart with the donut chart. """ # Load your data annotated_records = len(target_dataset) pending_records = int(os.getenv("TARGET_RECORDS")) - annotated_records # Prepare data for the donut chart source = pd.DataFrame( { "values": [annotated_records, pending_records], "category": [ANNOTATED, PENDING], "colors": ["#4CAF50", "#757575"], # Green for Completed, Grey 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("category:N", 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 kpi_chart_remaining() -> alt.Chart: """ This function returns a KPI chart with the remaining amount of records to be annotated. Returns: An altair chart with the KPI chart. """ pending_records = int(os.getenv("TARGET_RECORDS")) - len(target_dataset) # Assuming you have a DataFrame with user data, create a sample DataFrame data = pd.DataFrame({"Category": [PENDING], "Value": [pending_records]}) # Create Altair chart chart = ( alt.Chart(data) .mark_text(fontSize=100, align="center", baseline="middle", color="#e68b39") .encode(text="Value:N") .properties(title=PENDING, width=250, height=200) ) return chart def kpi_chart_submitted() -> alt.Chart: """ This function returns a KPI chart with the total amount of records that have been annotated. Returns: An altair chart with the KPI chart. """ total = len(target_dataset) # Assuming you have a DataFrame with user data, create a sample DataFrame data = pd.DataFrame({"Category": [NUMBER_ANNOTATED], "Value": [total]}) # Create Altair chart chart = ( alt.Chart(data) .mark_text(fontSize=100, align="center", baseline="middle", color="steelblue") .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. """ # Obtain the total amount of annotators total_annotators = len(user_ids_annotations) # 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="steelblue") .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_ids_annotations: 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. """ dataframe = pd.DataFrame( user_ids_annotations.items(), columns=[NAME, NUMBER_ANNOTATIONS] ) dataframe[NAME] = dataframe[NAME].apply(render_hub_user_link) dataframe = dataframe.sort_values(by=NUMBER_ANNOTATIONS, ascending=False) return dataframe.head(N) def fetch_data() -> None: """ This function fetches the data from the source and target datasets and updates the global variables. """ print(f"Starting to fetch data: {datetime.datetime.now()}") global source_dataset, target_dataset, user_ids_annotations, annotated, remaining, percentage_completed, top_dataframe source_dataset, target_dataset = obtain_source_target_datasets() user_ids_annotations = get_user_annotations_dictionary(target_dataset) annotated = len(target_dataset) remaining = int(os.getenv("TARGET_RECORDS")) - annotated percentage_completed = round( (annotated / int(os.getenv("TARGET_RECORDS"))) * 100, 1 ) # Print the current date and time print(f"Data fetched: {datetime.datetime.now()}") 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(user_ids_annotations, N=N) def main() -> None: # Connect to the space with rg.init() rg.init( api_url=os.getenv("ARGILLA_API_URL"), api_key=os.getenv("ARGILLA_API_KEY"), ) fetch_data() # CSS to handle RTL direction for Arabic content and remove the orange border css = """ .generating { border: none; } .markdown-arabic { direction: rtl; text-align: right; unicode-bidi: embed; } """ with gr.Blocks(css=css) as demo: gr.Markdown( """
![image/png](https://cdn-uploads.huggingface.co/production/uploads/64d5698102e58cc1fdd0b585/MnWb3lFLVu6ufcmupu3_o.png) # 🌍 العربية - مشروع تقييم المطالبات متعدد اللغات أطلقت Hugging Face و Argilla مشروع [مشروع تقييم المطالبات متعدد اللغات](https://github.com/huggingface/data-is-better-together/tree/main/prompt_translation). إنه معيار متعدد اللغات مفتوح لتقييم نماذج اللغة، وبالطبع، للغة العربية أيضًا. ## 💡الهدف هو ترجمة 500 مطالبة وكالعادة: البيانات عالية الجودة مطلوبة! لقد اختار المجتمع أفضل 500 مطالبة التي ستشكل المعيار. باللغة الإنجليزية، بالطبع. **لذلك نحتاج إلى مساعدتك**: إذا قمنا جميعًا بترجمة الـ 500 مطالبة، يمكننا إضافة العربية إلى قائمة المتصدرين. ## 📌كيفية المشاركة المشاركة سهلة. اذهب إلى [فضاء التعليق](https://huggingface.co/spaces/2A2I/prompt-translation-for-Arabic)، قم بتسجيل الدخول أو إنشاء حساب على Hugging Face، ويمكنك البدء في العمل. شكرًا لك مقدمًا 🤗! آه، وسنقدم لك دفعة صغيرة: GPT4 قد أعد بالفعل اقتراحًا للترجمة لك.
""" ) gr.Markdown( """
## 🚀 التقدم الحالي وهذا ما حققناه حتى الآن!
""" ) with gr.Row(): kpi_submitted_plot = gr.Plot(label="Plot") demo.load(kpi_chart_submitted, inputs=[], outputs=[kpi_submitted_plot]) kpi_remaining_plot = gr.Plot(label="Plot") demo.load(kpi_chart_remaining, inputs=[], outputs=[kpi_remaining_plot]) donut_total_plot = gr.Plot(label="Plot") demo.load(donut_chart_total, inputs=[], outputs=[donut_total_plot]) gr.Markdown( """
## 🌠 قاعة المشاهير : هنا يمكنك رؤية المستخدمين الذين لديهم أكبر عدد من المساهمات
""" ) with gr.Row(): kpi_hall_plot = gr.Plot(label="Plot") demo.load(kpi_chart_total_annotators, inputs=[], outputs=[kpi_hall_plot]) top_df_plot = gr.Dataframe( headers=[NAME, NUMBER_ANNOTATIONS], datatype=[ "markdown", "number", ], row_count=50, col_count=(2, "fixed"), interactive=False, ) demo.load(get_top, None, [top_df_plot]) # Launch the Gradio interface demo.launch() if __name__ == "__main__": main()