app.py

import gradio as gr
from os import environ
from transformers import pipeline, AutoModelForSequenceClassification
from libs.eurovoc.EurovocTh import EurovocThExt

api_key = environ.get("api_key")

def app(input, filter_strategy, relevence_threshold, k, length_management_strategy, token_overlap, results_merge_stategy, drop_last, model_name):

    th = EurovocThExt('?','altro')

    classifier_pipeline = pipeline(model=model_name, tokenizer=model_name)
    length_limit = classifier_pipeline.tokenizer.model_max_length
    
    kwargs = {'padding':True,'truncation':True,'max_length':length_limit}

    kwargs['top_k'] = classifier_pipeline.model.num_labels

    if length_management_strategy=='solo inizio':
        res = classifier_pipeline(input, **kwargs)
        res = parse_results(res) # convert to expected format for gradio interface    
        
    elif length_management_strategy=='intero documento':
        res = execute_pipeline_su_chunks(classifier_pipeline, input, length_limit,drop_last = drop_last, token_overlap=token_overlap, merge_strategy =results_merge_stategy ,pipeline_kwargs=kwargs)
        

    if filter_strategy == 'soglia di confidenza':
        res = {k:v for k,v in res.items() if v >= relevence_threshold}
    elif filter_strategy == 'top k':
        rev_res = sorted([(v,k) for k,v in res.items()], reverse=True)
        res = {k:v for (v,k) in rev_res[:k] }

    # label fullname
    res = {f'{k} - {th.get_label(k)}':v for k,v in res.items()}
    
    
    return res


def execute_pipeline_su_chunks(pipeline, input_string, max_length, drop_last = False, token_overlap=0, merge_strategy ='avg' ,pipeline_kwargs={}):
    tokens = pipeline.tokenizer.tokenize(input_string)
    results = {}
    
    overlap_len = int(max_length * token_overlap)
    shift_len = max_length - overlap_len
    
    start = 0
    end = max_length

    iterations = 0

    while start < len(tokens):
        chunk_tokens = tokens[start:end]

        start += shift_len
        end = start + max_length

        is_last = not (start < len(tokens))
        
        if not (is_last and drop_last):
            chunk_string = pipeline.tokenizer.convert_tokens_to_string(chunk_tokens)
            chunck_res = pipeline(chunk_string, **pipeline_kwargs)
            chunck_res = parse_results(chunck_res) # convert to expected format for gradio interface   

            if results != {}:
                new_results = {}
                for k, v in results.items():
                    if merge_strategy == 'avg':
                        new_results[k] = v + chunck_res[k]
                    elif merge_strategy == 'max':
                        new_results[k] = max(v, chunck_res[k])
                results = new_results
            else:
                results = chunck_res

            iterations += 1
    
    if merge_strategy == 'avg':
        results = {k:v/iterations for k,v in results.items()}

    return results


def parse_results(res):
    return  {el['label']:el['score'] for el in res} # convert to expected format for gradio interface

def load_example(evt: gr.SelectData):

    txt = None
    choice = evt.value
    print(choice)
    if choice == 'esempio 1':
        with open('Example1.txt') as f:
            txt = f.read()
    elif choice == 'esempio 2':
        with open('Example2.txt') as f:
            txt = f.read()
    else:
        txt = 'no choice'

    return txt

def load_example_from_list(l):

    choice = l[0]

    txt = ''

    try:
        fn = f'e_{choice}.txt'
        with open(fn) as f:
            txt = f.read()
    except Exception as e:
       print(e)
       txt = ''

    return txt

samples = [
    ['1', "Accordo tra l'Unione europea e il Principato del Liechtenstein ai fini dell'applicazione di talune disposizioni della decisione 2008/615/GAI del Consiglio sul potenziamento della cooperazione transfrontaliera, soprattutto nella lotta al terrorismo e alla criminalità transfrontaliera, della decisione 2008/616/GAI del Consiglio relativa all'attuazione della decisione 2008/615/GAI sul potenziamento della cooperazione transfrontaliera, soprattutto nella lotta al terrorismo e alla criminalità transfrontaliera, compreso l'allegato, e della decisione quadro 2009/905/GAI del Consiglio sull'accreditamento dei fornitori di servizi forensi che effettuano attività di laboratorio", ['28 QUESTIONI SOCIALI, 36 SCIENZE, 04 VITA POLITICA, 72 GEOGRAFIA, 10 UNIONE EUROPEA, 48 TRASPORTO, 32 ISTRUZIONE E COMUNICAZIONE, 08 RELAZIONI INTERNAZIONALI']],
    ['2', "Causa C-377/21: Sentenza della Corte (Settima Sezione) del 7 luglio 2022 (domanda di pronuncia pregiudiziale proposta dalla Cour du travail de Mons — Belgio) — Ville de Mons, Zone de secours Hainaut — Centre / RM (Rinvio pregiudiziale – Politica sociale – Direttiva 97/81/CE – Accordo quadro sul lavoro a tempo parziale – Clausola 4 – Principio di non discriminazione – Principio del pro rata temporis – Presa in considerazione, ai fini del calcolo della retribuzione di un vigile del fuoco professionista assunto a tempo pieno, dell’anzianità da quest’ultimo acquisita in qualità di vigile del fuoco volontario, secondo il principio del pro rata temporis)", ['12 DIRITTO, 52 AMBIENTE, 28 QUESTIONI SOCIALI, 44 OCCUPAZIONE E LAVORO']],
    ['3', "Parere del Comitato economico e sociale europeo in merito alla «Proposta di regolamento del Parlamento europeo e del Consiglio che modifica il regolamento (UE) 2018/848 relativo alla produzione biologica per quanto riguarda la sua data di applicazione e alcune altre date in esso previste» [COM(2020) 483 final – 2020/0231 (COD)]", ['56 AGRICOLTURA, SILVICOLTURA E PESCA, 28 QUESTIONI SOCIALI, 20 SCAMBI ECONOMICI E COMMERCIALI, 10 UNIONE EUROPEA, 60 AGROALIMENTARE']],
    ['4', "Causa C-675/15 P: Impugnazione proposta il 15 dicembre 2015 dalla The Tea Board avverso la sentenza del Tribunale (Ottava Sezione) del 2 ottobre 2015, causa T-626/13, The Tea Board/Ufficio per l’armonizzazione nel mercato interno (marchi, disegni e modelli)", ['68 INDUSTRIA, 64 PRODUZIONE, TECNOLOGIA E RICERCA, 32 ISTRUZIONE E COMUNICAZIONE']],
    ['5', "Decisione del Parlamento europeo del 12 dicembre 2013 di non sollevare obiezioni al regolamento delegato della Commissione, del 30 ottobre 2013, che modifica gli allegati I, II e IV del regolamento (UE) n. 978/2012 relativo all'applicazione di un sistema di preferenze tariffarie generalizzate (C(2013)07167 — 2013/2929(DEA))", "76 ORGANIZZAZIONI INTERNAZIONALI, 72 GEOGRAFIA, 16 ECONOMIA, 20 SCAMBI ECONOMICI E COMMERCIALI"]
   ]
    


with gr.Blocks() as interface:

    gr.Markdown("""
        # Classificatore domini EuroVoc

        Strumento che permette la classificazione di un documento rispetto alle 546 classi del livello superiore della [classificazione EuroVoc](https://eur-lex.europa.eu/browse/eurovoc.html?locale=it). 
        """)

    with gr.Row():
      with gr.Column():

        model_name = gr.Dropdown(choices=["gianma/classifierEUtopLevelBDWdIntRebalanced", "gianma/classifierEUtopLevelLongerTrainBDWdInt", "gianma/classifierEUtopLevelLongerTrain", "gianma/classifierEUtopLevelLongerTrainAugmented", "gianma/classifierEUtopLevelRoberta" ], value="gianma/classifierEUtopLevelBDWdIntRebalanced", label='selezione modello')

        input_text = gr.Textbox(label='Testo da classificare', lines=3, placeholder='inserire qui il testo che si desidera classificare')

        with gr.Accordion("Opzioni"):

          with gr.Box():
            filter_strategy = gr.Radio(['top k', 'soglia di confidenza',], value='top k', visible=False)
            gr.Markdown("""
            ### Strategia di filtraggio dei risultati

            Indica se mostrare solamente le k classi ritenute più probabili oppure se riportare le classi associate ad un punteggio maggiore o uguale alla soglia di confidenza indicata
            """)
            with gr.Tab("top k") as tab_top_k:
              s_k = gr.Slider(1, 546, step= 1, value=6, label='k')
            with gr.Tab("soglia di confidenza") as tab_confidence:
              s_confidence = gr.Slider(0, 1, value=0.5,label='valore soglia',)
            tab_confidence.select(lambda :"soglia di confidenza", None, filter_strategy)
            tab_top_k.select(lambda:'top k', None, filter_strategy)

          with gr.Box():
            gr.Markdown("""
            ### Strategia di elaborazione del documento
            """)

            # simplification
            document_reading_strategy = gr.Radio(['solo inizio','intero documento'], value = 'solo inizio', label="Indica se analizzare solamente la prima porzione del documento (più veloce) o se eseguire più classificazioni rispetto alle diverse porzioni del documento unendone i risultati")
            
              
            #document_reading_strategy = gr.Radio(['solo inizio', 'intero documento'], value='solo inizio', visible=False)
            #with gr.Tab("solo inizio") as tab_beginning_only:
            #  gr.Markdown("Opzione migliore come velocità di elaborazione")
            #with gr.Tab("intero documento") as tab_whole_document:
            #  gr.Markdown("Attenzione: questa opzione allunga i tempi di elaborazione")
            chunk_overlap = gr.Slider(0, 0.5, value=0.5, label='sovrapposizione porzioni documento', info='indica la percentuale di sovrapposizione tra le diverse porzioni di testo analizzate', visible=False)
            combine_strategy = gr.Radio(['avg', 'max'], value='max', label='strategia per la combinazione dei risultati dei chunk', info='avg indica di considerare il valore medio delle classificazioni eseguite sulle diverse porzioni del documento, mentre max prenderà i valori massimi (aumenta il numero di classi rilevanti)', visible=False)
            exclude_last = gr.Checkbox(value=True, label='escludi ultimo chuck', info='permette di rimuovere l\' ultima porzione del testo quando viene analizzato compretamente per evitare di elaborare un porzione troppo ridotta', visible=False)
            #tab_beginning_only.select(lambda:'solo inizio', None, document_reading_strategy)
            #tab_whole_document.select(lambda:'intero documento', None, document_reading_strategy)


        with gr.Accordion("Esempi", open=False):
            nome_esempio = gr.Textbox(visible=False)
            titolo = gr.Textbox(visible=False)
            ground_truth_esempio = gr.Textbox(visible=False)

            data = gr.Dataset(components=[nome_esempio, titolo, ground_truth_esempio], samples=samples, label='Cliccare su un esempio per caricarlo', headers=["#", "Titolo", "Ground Truth"])
    

        data.click(load_example_from_list, data, input_text)

        #examples = gr.Dropdown(
        #    [' ', 'esempio 1', 'esempio 2'],
        #    label="Esempi"
        #)
        #examples.select(load_example, inputs=None, outputs=input_text)

            
            
        with gr.Row():
          clear_button = gr.Button("Cancella testo")
          classify_button = gr.Button("Classifica", variant='primary')                        
        
      with gr.Column():
        outputs = gr.Label()


    clear_button.click(fn=lambda: None, inputs=None, outputs=input_text)
    classify_button.click(app, inputs=[input_text,filter_strategy, s_confidence, s_k, document_reading_strategy, chunk_overlap, combine_strategy, exclude_last, model_name], outputs=outputs)

interface.launch()