import re
from collections import defaultdict
from math import *
import itertools
import queue
import copy
from math import factorial
import time

from corpustools.exceptions import FuncLoadError
from .io import save_minimal_pairs

def matches(first, second):
"""
Determine if two neutralized transcriptions are a minimal pair or not

Parameters
----------
first : tuple
Tuple of the neutralized sequence, the spelling of the word,
and the unneutralized sequence
second : tuple
Tuple of the neutralized sequence, the spelling of the word,
and the unneutralized sequence

Returns
-------
bool
Returns True if the neutralized sequences match, they both contain
neutralized segments, and the spellings and original transcriptions
are different; otherwise returns False
"""
return (first[0] == second[0] and first[1] != second[1]
and 'NEUTR:' in first[0] and 'NEUTR:' in second[0]
and first[2] != second[2])

[docs]def minpair_fl(corpus_context, segment_pairs, relative_count = True, distinguish_homophones = False, stop_check = None, call_back = None): """Calculate the functional load of the contrast between two segments as a count of minimal pairs. Parameters ---------- corpus_context : CorpusContext Context manager for a corpus segment_pairs : list of length-2 tuples of str The pairs of segments to be conflated. relative_count : bool, optional If True, divide the number of minimal pairs by the total count by the total number of words that contain either of the two segments. distinguish_homophones : bool, optional If False, then you'll count sock~shock (sock=clothing) and sock~shock (sock=punch) as just one minimal pair; but if True, you'll overcount alternative spellings of the same word, e.g. axel~actual and axle~actual. False is the value used by Wedel et al. stop_check : callable, optional Optional function to check whether to gracefully terminate early call_back : callable, optional Optional function to supply progress information during the function Returns ------- tuple(int or float, list) Tuple of: 0. if relative_count==False, an int of the raw number of minimal pairs; if relative_count==True, a float of that count divided by the total number of words in the corpus that include either s1 or s2; and 1. list of minimal pairs. """ if stop_check is not None and stop_check(): return all_segments = list(itertools.chain.from_iterable(segment_pairs)) neutralized = [] if call_back is not None: call_back('Finding and neutralizing instances of segments...') call_back(0, len(corpus_context)) cur = 0 for w in corpus_context: if stop_check is not None and stop_check(): return if call_back is not None: cur += 1 if cur % 100 == 0: call_back(cur) tier = getattr(w, corpus_context.sequence_type) if any([s in tier for s in all_segments]): n = [neutralize_segment(seg, segment_pairs) for seg in tier] neutralized.append(('.'.join(n), w, tier)) if stop_check is not None and stop_check(): return minpairs = [] if call_back is not None: call_back('Counting minimal pairs...') call_back(0,factorial(len(neutralized))/(factorial(len(neutralized)-2)*2)) cur = 0 for first,second in itertools.combinations(neutralized, 2): if stop_check is not None and stop_check(): return if call_back is not None: cur += 1 if cur % 100 == 0: call_back(cur) if not matches(first,second): continue ordered_pair = sorted([(first[1],first[2]), (second[1], second[2])], key = lambda x: x[1]) minpairs.append(tuple(ordered_pair)) if not distinguish_homophones: actual_minpairs = {} for pair in minpairs: key = (pair[0][1], pair[1][1]) # Keys are tuples of tiers if key not in actual_minpairs: actual_minpairs[key] = (pair[0][0], pair[1][0]) # Values are words else: pair_freq = pair[0][0].frequency + pair[1][0].frequency existing_freq = actual_minpairs[key][0].frequency + \ actual_minpairs[key][1].frequency if pair_freq > existing_freq: actual_minpairs[key] = (pair[0][0], pair[1][0]) result = sum((x[0].frequency + x[1].frequency)/2 for x in actual_minpairs.values()) else: result = sum((x[0][0].frequency + x[1][0].frequency)/2 for x in minpairs) if relative_count and len(neutralized) > 0: result /= sum(x[1].frequency for x in neutralized) return (result, minpairs)
[docs]def deltah_fl(corpus_context, segment_pairs, stop_check = None, call_back = None): """Calculate the functional load of the contrast between between two segments as the decrease in corpus entropy caused by a merger. Parameters ---------- corpus_context : CorpusContext Context manager for a corpus segment_pairs : list of length-2 tuples of str The pairs of segments to be conflated. stop_check : callable, optional Optional function to check whether to gracefully terminate early call_back : callable, optional Optional function to supply progress information during the function Returns ------- float The difference between a) the entropy of the choice among non-homophonous words in the corpus before a merger of s1 and s2 and b) the entropy of that choice after the merger. """ if call_back is not None: call_back('Finding instances of segments...') call_back(0, len(corpus_context)) cur = 0 freq_sum = 0 original_probs = defaultdict(float) for w in corpus_context: if stop_check is not None and stop_check(): return if call_back is not None: cur += 1 if cur % 20 == 0: call_back(cur) f = w.frequency original_probs[getattr(w, corpus_context.sequence_type)] += f freq_sum += f original_probs = {k:v/freq_sum for k,v in original_probs.items()} if stop_check is not None and stop_check(): return preneutr_h = entropy(original_probs.values()) neutralized_probs = defaultdict(float) if call_back is not None: call_back('Neutralizing instances of segments...') call_back(0, len(list(original_probs.keys()))) cur = 0 for k,v in original_probs.items(): if stop_check is not None and stop_check(): return if call_back is not None: cur += 1 if cur % 100 == 0: call_back(cur) neutralized_probs['.'.join([neutralize_segment(s, segment_pairs) for s in k])] += v postneutr_h = entropy(neutralized_probs.values()) if stop_check is not None and stop_check(): return result = preneutr_h - postneutr_h if result < 1e-10: result = 0.0 return result
[docs]def relative_minpair_fl(corpus_context, segment, relative_count = True, distinguish_homophones = False, output_filename = None, stop_check = None, call_back = None): """Calculate the average functional load of the contrasts between a segment and all other segments, as a count of minimal pairs. Parameters ---------- corpus_context : CorpusContext Context manager for a corpus segment : str The target segment. relative_count : bool, optional If True, divide the number of minimal pairs by the total count by the total number of words that contain either of the two segments. distinguish_homophones : bool, optional If False, then you'll count sock~shock (sock=clothing) and sock~shock (sock=punch) as just one minimal pair; but if True, you'll overcount alternative spellings of the same word, e.g. axel~actual and axle~actual. False is the value used by Wedel et al. stop_check : callable, optional Optional function to check whether to gracefully terminate early call_back : callable, optional Optional function to supply progress information during the function Returns ------- int or float If relative_count==False, returns an int of the raw number of minimal pairs. If relative_count==True, returns a float of that count divided by the total number of words in the corpus that include either s1 or s2. """ all_segments = corpus_context.inventory segment_pairs = [(segment,other) for other in all_segments if other != segment and other != '#'] results = [] to_output = [] for sp in segment_pairs: res = minpair_fl(corpus_context, [sp], relative_count = relative_count, distinguish_homophones = distinguish_homophones, stop_check = stop_check, call_back = call_back) results.append(res[0]) if output_filename is not None: to_output.append((sp, res[1])) if output_filename is not None: save_minimal_pairs(output_filename, to_output) return sum(results)/len(segment_pairs)
[docs]def relative_deltah_fl(corpus_context, segment, stop_check = None, call_back = None): """Calculate the average functional load of the contrasts between a segment and all other segments, as the decrease in corpus entropy caused by a merger. Parameters ---------- corpus_context : CorpusContext Context manager for a corpus segment : str The target segment. stop_check : callable, optional Optional function to check whether to gracefully terminate early call_back : callable, optional Optional function to supply progress information during the function Returns ------- float The difference between a) the entropy of the choice among non-homophonous words in the corpus before a merger of s1 and s2 and b) the entropy of that choice after the merger. """ all_segments = corpus_context.inventory segment_pairs = [(segment,other) for other in all_segments if other != segment and other != '#'] results = [] for sp in segment_pairs: results.append(deltah_fl(corpus_context, [sp], stop_check = stop_check, call_back = call_back)) return sum(results)/len(segment_pairs)
def collapse_segpairs_fl(corpus_context, **kwargs): func_type = kwargs.get('func_type') segment_pairs = kwargs.get('segment_pairs') relative_count = kwargs.get('relative_count') distinguish_homophones = kwargs.get('distinguish_homophones') if func_type == 'min_pairs': fl = minpair_fl(corpus_context, segment_pairs, relative_count, distinguish_homophones) elif func_type == 'entropy': fl = deltah_fl(corpus_context, segment_pairs) def individual_segpairs_fl(corpus_context, **kwargs): func_type = kwargs.get('func_type') segment_pairs = kwargs.get('segment_pairs') relative_count = kwargs.get('relative_count') distinguish_homophones = kwargs.get('distinguish_homophones') results = [] for pair in segment_pairs: if func_type == 'min_pairs': fl = minpair_fl(corpus_context, [pair], relative_count, distinguish_homophones) elif func_type == 'entropy': fl = deltah_fl(corpus_context, [pair]) results.append(fl) def entropy(probabilities): """Calculate the entropy of a choice from the provided probability distribution. Parameters --------- probabilities : list of floats Contains the probability of each item in the list. Returns ------- float Entropy """ return -(sum([p*log(p,2) if p > 0 else 0 for p in probabilities])) def neutralize_segment(segment, segment_pairs): for sp in segment_pairs: try: s = segment.symbol except AttributeError: s = segment if s in sp: return 'NEUTR:'+''.join(str(x) for x in sp) return s def all_pairwise_fls(corpus_context, relative_fl = False, algorithm = 'minpair', relative_count = True, distinguish_homophones = False): """Calculate the functional load of the contrast between two segments as a count of minimal pairs. Parameters ---------- corpus_context : CorpusContext Context manager for a corpus relative_fl : bool If False, return the FL for all segment pairs. If True, return the relative (average) FL for each segment. algorithm : str {'minpair', 'deltah'} Algorithm to use for calculating functional load: "minpair" for minimal pair count or "deltah" for change in entropy. relative_count : bool, optional If True, divide the number of minimal pairs by the total count by the total number of words that contain either of the two segments. distinguish_homophones : bool, optional If False, then you'll count sock~shock (sock=clothing) and sock~shock (sock=punch) as just one minimal pair; but if True, you'll overcount alternative spellings of the same word, e.g. axel~actual and axle~actual. False is the value used by Wedel et al. Returns ------- list of tuple(tuple(str, st), float) OR list of (str, float) Normally returns a list of all Segment pairs and their respective functional load values, as length-2 tuples ordered by FL. If calculating relative FL, returns a dictionary of each segment and its relative (average) FL, with entries ordered by FL. """ fls = {} total_calculations = ((((len(corpus_context.inventory)-1)**2)-len(corpus_context.inventory)-1)/2)+1 ct = 1 t = time.time() if '' in corpus_context.inventory: raise Exception('Warning: Calculation of functional load for all segment pairs requires that all items in corpus have a non-null transcription.') for i, s1 in enumerate(corpus_context.inventory[:-1]): for s2 in corpus_context.inventory[i+1:]: if s1 != '#' and s2 != '#': print('Performing FL calculation {} out of {} possible'.format(str(ct), str(total_calculations))) ct += 1 print('Duration of last calculation: {}'.format(str(time.time() - t))) t = time.time() if type(s1) != str: s1 = s1.symbol if type(s2) != str: s2 = s2.symbol if algorithm == 'minpair': fl = minpair_fl(corpus_context, [(s1, s2)], relative_count=relative_count, distinguish_homophones=distinguish_homophones)[0] elif algorithm == 'deltah': fl = deltah_fl(corpus_context, [(s1, s2)]) fls[(s1, s2)] = fl if not relative_fl: ordered_fls = sorted([(pair, fls[pair]) for pair in fls], key=lambda p: p[1], reverse=True) return ordered_fls elif relative_fl: rel_fls = {} for s in corpus_context.inventory: if type(s) != str: s = s.symbol if s != '#': total = 0.0 for pair in fls: if s == pair[0] or s == pair[1]: total += fls[pair] rel_fls[s] = total / (len(corpus_context.inventory) - 1) ordered_rel_fls = sorted([(s, rel_fls[s]) for s in rel_fls], key=lambda p: p[1], reverse=True) return ordered_rel_fls