<commit_before>#!/usr/bin/python

"""Read segmentation overlap measures of individual pairwise registrations
from CSV files generated by mirtk evaluate-overlap and compute the average
overlap for each class.
"""

import sys
import csv
import numpy as np
import argparse


def abbreviate(measure):
    """Get abbreviation of measure."""
    lstr = measure.lower()
    if lstr in ('sensitivity', 'truepositiverate', 'true positive rate'):
        return 'TPR'
    if lstr in ('specificity', 'truenegativerate', 'true negative rate'):
        return 'TNR'
    if lstr in ('positivepredictivevalue', 'positive predictive value', 'precision'):
        return 'PPV'
    if lstr == 'negativepredictivevalue':
        return 'NPV'
    if lstr == 'falsepositiverate':
        return 'FPR'
    if lstr == 'falsediscoveryrate':
        return 'FDR'
    if lstr == 'accuracy':
        return 'ACC'
    if lstr in ('informedness', 'bookmaker informedness'):
        return 'BM'
    if lstr in ('matthewscorrelation', 'matthewscorrelationcoefficient',
                'matthews correlation', 'matthews correlation coefficient'):
        return 'MCC'
    return measure


def evaluate_overlap(tp, fp, fn, tn, measure='Dice'):
    """Evaluate overlap measure given entries of confusion matrix."""
    measure = abbreviate(measure).lower()
    if measure == 'tpr':
        return float(tp) / float(tp + fn)
    if measure == 'tnr':
        return float(tn) / float(fp + tn)
    if measure == 'ppv':
        return float(tp) / float(tp + fp)
    if measure == 'npv':
        return float(tn) / float(tn + fn)
    if measure == 'fpr':
        return float(fp) / float(fp + tn)
    if measure == 'fdr':
        return float(fp) / float(fp + tp)
    if measure == 'fnr':
        return float(fn) / float(tp + fn)
    if measure == 'acc':
        return float(tp + tn) / float(tp + fn + fp + tn)
    if measure == 'bm':
        return float(tp) / float(tp + fn) + float(tn) / float(fp + tn) - 1.
    if measure == 'mcc':
        denom = sqrt(float(tp + fp) * float(tp + fn) * float(tn + fp) * float(tn + fn))
        return (float(tp) * float(tn) - float(fp) * float(fn)) / denom
    if measure.lower() == 'markedness':
        return float(tp) / float(tp + fp) + float(tn) / float(tn + fn) - 1.
    if measure.lower() in ('f1score', 'f1-score', 'f-score', 'fmeasure', 'f-measure', 'dice'):
        return float(2 * tp) / float((fp + tp) + (tp + fn))
    if measure == 'jaccard':
        return float(tp) / float(fp + tp + fn)
    else:
        raise Exception("Unknown overlap measure: {}".format(measure))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('tables', nargs='+',
                        help="List of CSV files.")
    parser.add_argument('--micro', action='store_true',
                        help="Compute micro-average of each measure instead of a macro-average")
    args = parser.parse_args()
    header = []
    labels = []
    with open(args.tables[0], 'rb') as f:
        reader = csv.reader(f)
        header = reader.next()
        for row in reader:
            if len(row) != len(header):
                raise Exception("Rows of CSV tables must have equal number of columns")
            labels.append(int(row[0]))
    tp_col = -1
    fp_col = -1
    fn_col = -1
    tn_col = -1
    for c in range(1, len(header)):
        lstr = header[c].lower()
        if lstr == 'tp':
            tp_col = c
        elif lstr == 'fp':
            fp_col = c
        elif lstr == 'fn':
            fn_col = c
        elif lstr == 'tn':
            tn_col = c
    if args.micro and (tp_col == -1 or fp_col == -1 or fn_col == -1 or tn_col == -1):
        raise Exception("Missing one or more of TP,FP,FN,TN columns needed for micro-averaging")
    sums = np.zeros((len(labels), len(header) - 1), dtype=np.float)
    for csv_name in args.tables:
        sums += np.genfromtxt(csv_name,
                              delimiter=',',
                              skip_header=1,
                              usecols=range(1, len(header)),
                              dtype=np.float)
    num = len(args.tables)
    out = sys.stdout
    try:
        if args.micro:
            cols = []
            for c in range(1, len(header)):
                if c not in (tp_col, fp_col, fn_col, tn_col):
                    cols.append(c)
            out.write(header[0])
            out.write(',')
            out.write(','.join([header[c] for c in cols]))
            out.write('\n')
            for l in range(0, len(labels)):
                out.write(str(labels[l]))
                for c in cols:
                    avg = evaluate_overlap(sums[l, tp_col - 1],
                                           sums[l, fp_col - 1],
                                           sums[l, fn_col - 1],
                                           sums[l, tn_col - 1],
                                           measure=header[c])
                    out.write(',')
                    out.write('{:.5f}'.format(avg))
                out.write('\n')
        else:
            out.write(','.join(header))
            out.write('\n')
            for l in range(0, len(labels)):
                out.write(str(labels[l]))
                for c in range(0, len(header) - 1):
                    avg = sums[l, c] / num
                    out.write(',')
                    out.write('{:.5f}'.format(avg))
                out.write('\n')
    finally:
        if out != sys.stdout:
            out.close()<commit_msg>enh: average-overlap -segment and -measure options [Applications]<commit_after>#!/usr/bin/python

"""Read segmentation overlap measures of individual pairwise registrations
from CSV files generated by mirtk evaluate-overlap and compute the average
overlap for each class.
"""

import re
import math
import sys
import csv
import numpy as np
import argparse


def abbreviate(measure):
    """Get abbreviation of measure."""
    lstr = measure.lower().replace(' ', '').replace('-', '')
    if lstr in ('dice', 'dicecoefficient', 'dicesimilaritycoefficient',
                'sorensendicecoefficient', 'sorensendiceindex', 'sorensenindex',
                'sorensondicecoefficient', 'sorensondiceindex', 'sorensonindex'):
        return 'DSC'
    if lstr in ('jaccard', 'jaccardindex', 'jaccardsimilaritycoefficient'):
        return 'JSC'
    if lstr in ('sensitivity', 'truepositiverate'):
        return 'TPR'
    if lstr in ('specificity', 'truenegativerate'):
        return 'TNR'
    if lstr in ('positivepredictivevalue', 'precision'):
        return 'PPV'
    if lstr == 'negativepredictivevalue':
        return 'NPV'
    if lstr == 'falsepositiverate':
        return 'FPR'
    if lstr == 'falsediscoveryrate':
        return 'FDR'
    if lstr == 'accuracy':
        return 'ACC'
    if lstr in ('informedness', 'bookmakerinformedness'):
        return 'BM'
    if lstr in ('matthewscorrelation', 'matthewscorrelationcoefficient'):
        return 'MCC'
    if lstr in ('f1score', 'fscore', 'fmeasure'):
        return 'F1'
    return measure


def evaluate_overlap(tp, fp, fn, tn, measure='DSC'):
    """Evaluate overlap measure given entries of confusion matrix."""
    abbr = abbreviate(measure).lower()
    if abbr == 'tpr':
        return float(tp) / float(tp + fn)
    if abbr == 'tnr':
        return float(tn) / float(fp + tn)
    if abbr == 'ppv':
        return float(tp) / float(tp + fp)
    if abbr == 'npv':
        return float(tn) / float(tn + fn)
    if abbr == 'fpr':
        return float(fp) / float(fp + tn)
    if abbr == 'fdr':
        return float(fp) / float(fp + tp)
    if abbr == 'fnr':
        return float(fn) / float(tp + fn)
    if abbr == 'acc':
        return float(tp + tn) / float(tp + fn + fp + tn)
    if abbr == 'bm':
        return float(tp) / float(tp + fn) + float(tn) / float(fp + tn) - 1.
    if abbr == 'mcc':
        denom = math.sqrt(float(tp + fp) * float(tp + fn) * float(tn + fp) * float(tn + fn))
        return (float(tp) * float(tn) - float(fp) * float(fn)) / denom
    if abbr == 'markedness':
        return float(tp) / float(tp + fp) + float(tn) / float(tn + fn) - 1.
    if abbr in ('f1', 'dsc'):
        return float(2 * tp) / float((fp + tp) + (tp + fn))
    if abbr == 'jsc':
        return float(tp) / float(fp + tp + fn)
    else:
        raise Exception("Unknown overlap measure: {} (abbr: {})".format(measure, abbr))


class ParseSegmentOptionArguments(argparse.Action):
    """Custom argparse action for --segment option."""

    def __init__(self, option_strings, dest, nargs=None, **kwargs):
        if nargs is not '+':
            raise ValueError("nargs must be +")
        self.re_range = re.compile('^(\d+)\.\.(\d+)$')
        super(ParseSegmentOptionArguments, self).__init__(option_strings, dest, nargs, **kwargs)

    def __call__(self, parser, namespace, values, option_string=None):
        if len(values) <= 1:
            raise argparse.ArgumentError(self, 'At least two arguments required')
        labels = []
        self.re_range = re.compile('^(\d+)\.\.(\d+)$')
        for value in values[1:]:
            try:
                m_range = self.re_range.match(value)
                if m_range:
                    a = int(m_range.group(1))
                    b = int(m_range.group(2))
                    labels.extend(range(a, b + 1))
                else:
                    labels.append(int(value))
            except ValueError:
                argparse.ArgumentTypeError("Label argument must be integer or range specification, i.e., 1..5")
        segments = getattr(namespace, self.dest)
        if not segments:
            segments = []
        segments.append((values[0], set(labels)))
        setattr(namespace, self.dest, segments)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('tables', nargs='+',
                        help="List of CSV files written by evaluate-overlap -table option.")
    parser.add_argument('--micro', '-micro', action='store_true',
                        help="Compute micro-average of each measure instead of a macro-average.")
    parser.add_argument('--measure', '-measure', '--metric', '-metric', dest='measure', nargs='+', default=[], action='append',
                        help="Overlap measure(s) to include in output, all by default.")
    parser.add_argument('--segment', '-segment', dest='segments', nargs='+', default=[], metavar='NAME LABEL...', action=ParseSegmentOptionArguments,
                        help="Average the overlap measures for the specified labels, LABEL must be integer or range '5..10'.")
    parser.add_argument('--output', '-output',
                        help="Name of output text file.")
    args = parser.parse_args()
    header = []
    labels = []
    with open(args.tables[0], 'rb') as f:
        reader = csv.reader(f)
        header = reader.next()
        for row in reader:
            if len(row) != len(header):
                raise Exception("Rows of CSV tables must have equal number of columns")
            labels.append(int(row[0]))
    header = [abbreviate(name) for name in header]
    measures = [abbreviate(measure) for measure in args.measure]
    tp_col = -1
    fp_col = -1
    fn_col = -1
    tn_col = -1
    for c in range(1, len(header)):
        lstr = header[c].lower()
        if lstr == 'tp':
            tp_col = c
        elif lstr == 'fp':
            fp_col = c
        elif lstr == 'fn':
            fn_col = c
        elif lstr == 'tn':
            tn_col = c
    if args.micro and (tp_col == -1 or fp_col == -1 or fn_col == -1 or tn_col == -1):
        raise Exception("Missing one or more of TP,FP,FN,TN columns needed for micro-averaging")
    if args.micro:
        usecols = (tp_col, fp_col, fn_col, tn_col)
    else:
        usecols = range(1, len(header))
    if args.micro and measures:
        header = [header[0]]
        header.extend(measures)
        cols = range(1, len(header))
    else:
        cols = []
        for measure in measures:
            if measure not in header:
                raise Exception("Requested measure {} not found in input tables, consider --micro average if TP,FP,FN,TN columns available".format(measure))
        for col in range(1, len(header)):
            if measures and header[col] not in measures:
                continue
            if args.micro and col in usecols:
                continue
            cols.append(col)
    sums = np.zeros((len(labels), len(usecols)), dtype=np.float)
    for csv_name in args.tables:
        sums += np.genfromtxt(csv_name, delimiter=',', skip_header=1, usecols=usecols, dtype=np.float)
    num = len(args.tables)
    out = sys.stdout
    if args.output:
        out = open(args.output, 'w')
    try:
        if args.segments:
            out.write('Segment')
        else:
            out.write(header[0])
        out.write(',')
        out.write(','.join([header[col] for col in cols]))
        out.write('\n')
        if args.segments:
            for segment in args.segments:
                out.write(segment[0])
                for col in cols:
                    avg = 0
                    for label in segment[1]:
                        row = labels.index(label)
                        if args.micro:
                            avg += evaluate_overlap(sums[row, tp_col - 1],
                                                    sums[row, fp_col - 1],
                                                    sums[row, fn_col - 1],
                                                    sums[row, tn_col - 1],
                                                    measure=header[col])
                        else:
                            avg += sums[row, col - 1] / num
                    avg /= len(segment[1])
                    out.write(',')
                    out.write('{:.5f}'.format(avg))
                out.write('\n')
        else:
            for row in range(0, len(labels)):
                out.write(str(labels[row]))
                for col in cols:
                    if args.micro:
                        avg = evaluate_overlap(sums[row, tp_col - 1],
                                               sums[row, fp_col - 1],
                                               sums[row, fn_col - 1],
                                               sums[row, tn_col - 1],
                                               measure=header[col])
                    else:
                        avg = sums[row, col - 1] / num
                    out.write(',')
                    out.write('{:.5f}'.format(avg))
                out.write('\n')
    finally:
        if out != sys.stdout:
            out.close()
{'fraction_non_alphanumeric': 0.1341803111624431, 'fraction_numerical': 0.005571030640668524, 'mean_word_length': 1.7705627705627707, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7570134', 'n_tokens_mistral': 4492, 'n_tokens_neox': 4328, 'n_words': 1215}