gleu.py

# Copyright 2020 The HuggingFace Datasets Authors and the current dataset script contributor.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import evaluate
import datasets
from collections import Counter
from math import log, exp
from random import seed, randint
from numpy import mean, std, round


# TODO: Add BibTeX citation
_CITATION = """\
@InProceedings{napoles-EtAl:2015:ACL-IJCNLP,
  author    = {Napoles, Courtney  and  Sakaguchi, Keisuke  and  Post, Matt  and  Tetreault, Joel},
  title     = {Ground Truth for Grammatical Error Correction Metrics},
  booktitle = {Proceedings of the 53rd Annual Meeting of the Association for Computational Linguistics and the 7th International Joint Conference on Natural Language Processing (Volume 2: Short Papers)},
  month     = {July},
  year      = {2015},
  address   = {Beijing, China},
  publisher = {Association for Computational Linguistics},
  pages     = {588--593},
  url       = {http://www.aclweb.org/anthology/P15-2097}
}
"""

# TODO: Add description of the module here
_DESCRIPTION = """\
 GLEU metric can be used for any monolingual "translation" task, that is it can be used for Grammatical Error Correction and other text re-writing tasks. BLEU  computes n-gram precisions over the reference but assigns more weight to n-grams that have been correctly changed from the source. GLEU rewards corrections while also correctly crediting unchanged source text. 
"""


# TODO: Add description of the arguments of the module here
_KWARGS_DESCRIPTION = """
Calculates how good are predictions given some references, using certain scores
Args:
    sources: Source language reference sentences. This is assumed to be same as references if not provided.
    references: Reference for each prediction. Each reference should be a string with tokens separated by spaces.
    predictions: list of predictions to score. Each prediction should be a string with tokens separated by spaces.
Returns:
    mean_gleu_score: Average gleu_score over all predictions.
    SD: standard deviation

Examples:

    >>> my_new_module = evaluate.load("my_new_module")
    >>> references=["We may in actual fact be communicating with a hoax Facebook acccount of a cyberfriend , which we assume to be real but in reality , it is a fake account ."]
    >>> results = my_new_module.compute(references=references, predictions=["We may of actual fact communicating with a hoax Facebook acccount of a cyber friend , which we assumed to be real but in reality , it is a fake account ."])
    >>> print(results)
    {'mean_gleu_score': 0.6}
    
    >>> results = my_new_module.compute(references=references, predictions=["We may be in actual fact communicating with a hoax Facebook acccount of a cyber friend , we assume to be real but in reality , it is a fake account ."])
    >>> print(results)
    {'mean_gleu_score': 0.62}
    
    >>> results = my_new_module.compute(references=references, predictions=["We may in actual fact communicating with a hoax Facebook account of a cyber friend , which we assume to be real but in reality , it is a fake accounts ."])
    >>> print(results)
    {'mean_gleu_score': 0.64}
    
"""

class GLEU():
    def __init__(self, order=4) :
        self.order = order
        
    def load_hypothesis_sentence(self,hypothesis) :
        self.hlen = len(hypothesis)
        self.this_h_ngrams = [self.get_ngram_counts(hypothesis,n)
                                for n in range(1,self.order+1) ]
    
    def load_sources(self,source_sents) :
        self.all_s_ngrams = [[self.get_ngram_counts(source_sent.split(),n)
                                for n in range(1,self.order+1) ]
                                    for source_sent in source_sents ]
    
    def load_references(self,ref_sents) :
        self.refs = [ [] for i in range(len(self.all_s_ngrams)) ]
        self.rlens = [ [] for i in range(len(self.all_s_ngrams)) ]
        for i, ref_sent in enumerate(ref_sents) :
            self.refs[i].append(ref_sent.split())
            self.rlens[i].append(len(ref_sent.split()))
    
        # count number of references each n-gram appear sin
        self.all_rngrams_freq = [ Counter() for i in range(self.order) ]
    
        self.all_r_ngrams = [ ]
        for refset in self.refs :
            all_ngrams = []
            self.all_r_ngrams.append(all_ngrams)
    
            for n in range(1,self.order+1) :
                ngrams = self.get_ngram_counts(refset[0],n)
                all_ngrams.append(ngrams)
    
                for k in ngrams.keys() :
                    self.all_rngrams_freq[n-1][k]+=1
    
                for ref in refset[1:] :
                    new_ngrams = self.get_ngram_counts(ref,n)
                    for nn in new_ngrams.elements() :
                        if new_ngrams[nn] > ngrams.get(nn,0) :
                            ngrams[nn] = new_ngrams[nn]
    
    def get_ngram_counts(self,sentence,n) :
        return Counter([tuple(sentence[i:i+n]) for i in range(len(sentence)+1-n)])
    
    # returns ngrams in a but not in b
    def get_ngram_diff(self,a,b) :
        diff = Counter(a)
        for k in (set(a) & set(b)) :
            del diff[k]
        return diff
    
    def normalization(self,ngram,n) :
        return 1.0*self.all_rngrams_freq[n-1][ngram]/len(self.rlens[0])
    
    # Collect BLEU-relevant statistics for a single hypothesis/reference pair.
    # Return value is a generator yielding:
    # (c, r, numerator1, denominator1, ... numerator4, denominator4)
    # Summing the columns across calls to this function on an entire corpus
    # will produce a vector of statistics that can be used to compute GLEU
    def gleu_stats(self,i,r_ind=None):
    
      hlen = self.hlen
      rlen = self.rlens[i][r_ind]
      
      yield hlen
      yield rlen
    
      for n in range(1,self.order+1):
        h_ngrams = self.this_h_ngrams[n-1]
        s_ngrams = self.all_s_ngrams[i][n-1]
        r_ngrams = self.get_ngram_counts(self.refs[i][r_ind],n)
    
        s_ngram_diff = self.get_ngram_diff(s_ngrams,r_ngrams)
    
        yield max([ sum( (h_ngrams & r_ngrams).values() ) - \
                    sum( (h_ngrams & s_ngram_diff).values() ), 0 ])
    
        yield max([hlen+1-n, 0])
    
    # Compute GLEU from collected statistics obtained by call(s) to gleu_stats
    def compute_gleu(self,stats,smooth=False):
        # smooth 0 counts for sentence-level scores
        if smooth :
            stats = [ s if s != 0 else 1 for s in stats ]
        if len(list(filter(lambda x: x==0, stats))) > 0:
            return 0
        (c, r) = stats[:2]
        log_gleu_prec = sum([log(float(x)/y) for x,y in zip(stats[2::2],stats[3::2])]) / 4
        return exp(min([0, 1-float(r)/c]) + log_gleu_prec)
    


@evaluate.utils.file_utils.add_start_docstrings(_DESCRIPTION, _KWARGS_DESCRIPTION)
class gleu(evaluate.Metric):
    """TODO: Short description of my evaluation module."""

    def _info(self):
        # TODO: Specifies the evaluate.EvaluationModuleInfo object
        return evaluate.MetricInfo(
            # This is the description that will appear on the modules page.
            module_type="metric",
            description=_DESCRIPTION,
            citation=_CITATION,
            inputs_description=_KWARGS_DESCRIPTION,
            # This defines the format of each prediction and reference
            features=datasets.Features(
                {
                    "predictions": datasets.Value("string", id="sequence"),
                    "references": datasets.Value("string", id="sequence"),
                }
            ),
            codebase_urls=["https://github.com/cnap/gec-ranking/"],
        )

    def _download_and_prepare(self, dl_manager):
        """Optional: download external resources useful to compute the scores"""
        # TODO: Download external resources if needed
        pass

    def _compute(self, references, predictions):
        """Returns the scores"""
        
        num_iterations = 500
        
        order=4
        
        if len(references)==1:
            num_iterations = 1
            
        gleu_calculator = GLEU(order=order)
        
        # if sources:
        #     gleu_calculator.load_sources(sources)
        # else:
        #     
        gleu_calculator.load_sources(references)
        gleu_calculator.load_references(references)
        
        # first generate a random list of indices, using a different seed
        # for each iteration
        indices = []
        for j in range(num_iterations) :
            seed(j*101)
            indices.append([randint(0,len(references)-1) for i in range(len(predictions))])
        
        iter_stats = [[0 for i in range(2*order+2)] for j in range(num_iterations) ]
        
        for i,h in enumerate(predictions) :
            gleu_calculator.load_hypothesis_sentence(h)
            
            # we are going to store the score of this sentence for each ref
            # so we don't have to recalculate them 500 times
            
            stats_by_ref = [ None for r in range(len(references)) ]
            
            for j in range(num_iterations) :
                ref = indices[j][i]
                this_stats = stats_by_ref[ref]
            
                if this_stats is None :
                    this_stats = [ s for s in gleu_calculator.gleu_stats(i,r_ind=ref) ]
                    stats_by_ref[ref] = this_stats
                    
                iter_stats[j] = [sum(scores) for scores in zip(iter_stats[j], this_stats)]
        
        sent_scores = [gleu_calculator.compute_gleu(stats) for stats in iter_stats]
        mean_score = mean(sent_scores)
        std_score = round(std(sent_scores),2)
        print(mean_score, sent_scores)
        return {"mean_gleu_score": mean_score}