ratsgo/BranchingEntropy.py

## BranchingEntropy.py
class BranchingEntropy:
    def __init__(self, min_length=2, max_length=7):

        self.min_length = min_length
        self.max_length = max_length

        self.encoder = IntegerEncoder()

        self.L = defaultdict(lambda: defaultdict(int))
        self.R = defaultdict(lambda: defaultdict(int))

    def get_all_access_variety(self):

        av = {}
        #words = set(self.L.keys())
        #words += set(self.R.keys())
        words = set(list(set(self.L.keys())) + list(set(self.R.keys()))) # 내가 수정

        for word in words:
            av[word] = self.get_access_variety(word)

        return av

    def get_access_variety(self, word, ignore_space=False):

        return (len(self.get_left_branch(word, ignore_space)), len(self.get_right_branch(word, ignore_space)))

    def get_all_branching_entropies(self, ignore_space=False):

        be = {}
        words = set(self.L.keys())
        for word in self.R.keys():
            words.add(word)

        for word in words:
            be[self.encoder.decode(word)] = self.get_branching_entropy(word, ignore_space)

        return be

    def get_branching_entropy(self, word, ignore_space=False):

        be_l = self.entropy(self.get_left_branch(word, ignore_space))
        be_r = self.entropy(self.get_right_branch(word, ignore_space))
        return (be_l, be_r)

    def entropy(self, dic):

        if not dic:
            return 0.0

        sum_count = sum(dic.values())
        entropy = 0

        for freq in dic.values():
            prob = freq / sum_count
            entropy += prob * math.log(prob)

        return -1 * entropy

    def get_left_branch(self, word, ignore_space=False):

        if isinstance(word, int):
            word_index = word
        else:
            word_index = self.encoder.encode(word)

        if (word_index == -1) or (not word_index in self.L):
            return {}

        branch = self.L[word_index]

        if ignore_space:
            return {w: f for w, f in branch.items() if not ' ' in self.encoder.decode(w, unknown=' ')}
        else:
            return branch

    def get_right_branch(self, word, ignore_space=False):

        if isinstance(word, int):
            word_index = word
        else:
            word_index = self.encoder.encode(word)

        if (word_index == -1) or (not word_index in self.R):
            return {}

        branch = self.R[word_index]

        if ignore_space:
            return {w: f for w, f in branch.items() if not ' ' in self.encoder.decode(w, unknown=' ')}
        else:
            return branch

    def counter_size(self):
        return (len(self.L), len(self.R))

    def prune_extreme_case(self, min_count):

        # TODO: encoder remove & compatify
        before_size = self.counter_size()
        self.L = defaultdict(lambda: defaultdict(int),
                             {word: dic for word, dic in self.L.items() if sum(dic.values()) > min_count})
        self.R = defaultdict(lambda: defaultdict(int),
                             {word: dic for word, dic in self.R.items() if sum(dic.values()) > min_count})
        after_size = self.counter_size()

        return (before_size, after_size)

    def train(self, sents, min_count=5, num_for_pruning=10000):

        for num_sent, sent in enumerate(sents):

            sent = sent.strip()
            if not sent:
                continue

            sent = ' ' + sent.strip() + ' '
            length = len(sent)

            for i in range(1, length - 1):
                for window in range(self.min_length, self.max_length + 1):

                    if i + window - 1 >= length:
                        continue

                    word = sent[i:i + window]
                    if ' ' in word:
                        continue

                    word_index = self.encoder.fit(word)

                    if sent[i - 1] == ' ':
                        left_extension = sent[max(0, i - 2):i + window]
                    else:
                        left_extension = sent[i - 1:i + window]

                    if sent[i + window] == ' ':
                        right_extension = sent[i:min(length, i + window + 2)]
                    else:
                        right_extension = sent[i:i + window + 1]

                    if left_extension == None or right_extension == None:
                        print(sent, i, window)

                    left_index = self.encoder.fit(left_extension)
                    right_index = self.encoder.fit(right_extension)

                    self.L[word_index][left_index] += 1
                    self.R[word_index][right_index] += 1

            if (num_for_pruning > 0) and ((num_sent + 1) % num_for_pruning == 0):
                before, after = self.prune_extreme_case(min_count)
                sys.stdout.write('\rnum sent = %d: %s --> %s' % (num_sent, str(before), str(after)))

        if (num_for_pruning > 0) and ((num_sent + 1) % num_for_pruning == 0):
            self.prune_extreme_case(min_count)
            sys.stdout.write('\rnum_sent = %d: %s --> %s' % (num_sent, str(before), str(after)))

    def load(self, model_fname, encoder_fname):

        self.encoder.load(encoder_fname)

        try:
            with open(model_fname, encoding='utf-8') as f:

                next(f)  # SKIP: parameters (min_length, max_length)
                token = next(f).split()
                self.min_length = int(token[0])
                self.max_length = int(token[1])

                next(f)  # SKIP: left side extension
                is_right_side = True

                for line in f:

                    if '# right side extension' in line:
                        is_right_side = True
                        continue

                    token = line.split();
                    word = int(token[0])
                    extension = int(token[1])
                    freq = int(token[2])

                    if is_right_side:
                        self.R[word][extension] = freq
                    else:
                        self.L[word][extension] = freq

        except Exception as e:
            print(e)

    def save(self, model_fname, encoder_fname):

        self.encoder.save(encoder_fname)

        try:
            with open(model_fname, 'w', encoding='utf-8') as f:

                f.write("# parameters (min_length max_length)\n")
                f.write('%d %d\n' % (self.min_length, self.max_length))

                f.write('# left side extension\n')
                for word, extension_dict in self.L.items():
                    for extension, freq in extension_dict.items():
                        f.write('%d %d %d\n' % (word, extension, freq))

                f.write('# right side extension\n')
                for word, extension_dict in self.R.items():
                    for extension, freq in extension_dict.items():
                        f.write('%d %d %d\n' % (word, extension, freq))

        except Exception as e:
            print(e)

    def words(self):
        return set(self.encoder.inverse)


class MaxScoreTokenizer:
    def __init__(self, max_length=10, scores={}, default_score=0.0):
        self.max_length = max_length
        self.scores = scores
        self.ds = default_score

    def tokenize(self, sentence):
        #return [self._recursive_tokenize(token) for token in sentence.split()]
        tmp = [self._recursive_tokenize(token) for token in sentence.split()]
        result = []
        for words in tmp:
            for tokens in words:
                result.append(tokens[0])
        return result

    def _recursive_tokenize(self, token, range_l=0, debug=False):

        length = len(token)
        if length <= 2:
            return [(token, 0, length, self.ds, length)]

        if range_l == 0:
            range_l = min(self.max_length, length)

        scores = self._initialize(token, range_l, length)
        if debug:
            pprint(scores)

        result = self._find(scores)

        adds = self._add_inter_subtokens(token, result)

        if result[-1][2] != length:
            adds += self._add_first_subtoken(token, result)

        if result[0][1] != 0:
            adds += self._add_last_subtoken(token, result)

        return sorted(result + adds, key=lambda x: x[1])

    def _initialize(self, token, range_l, length):
        scores = []
        for b in range(0, length - 1):
            for r in range(2, range_l + 1):
                e = b + r

                if e > length:
                    continue

                subtoken = token[b:e]
                score = self.scores.get(subtoken, self.ds)
                scores.append((subtoken, b, e, score, r))

        #return sorted(scores, key=lambda x: (x[3], x[4]), reverse=True)
        return sorted(scores, key=lambda x: (x[0], x[1]), reverse=True)

    def _find(self, scores):
        result = []
        num_iter = 0

        while scores:
            word, b, e, score, r = scores.pop(0)
            result.append((word, b, e, score, r))

            if not scores:
                break

            removals = []
            for i, (_1, b_, e_, _2, _3) in enumerate(scores):
                if (b_ < e and b < e_) or (b_ < e and e_ > b):
                    removals.append(i)

            for i in reversed(removals):
                del scores[i]

            num_iter += 1
            if num_iter > 100: break

        return sorted(result, key=lambda x: x[1])

    def _add_inter_subtokens(self, token, result):
        adds = []
        for i, base in enumerate(result[:-1]):
            if base[2] == result[i + 1][1]:
                continue

            b = base[2]
            e = result[i + 1][1]
            subtoken = token[b:e]
            adds.append((subtoken, b, e, self.ds, e - b))

        return adds

    def _add_first_subtoken(self, token, result):
        b = result[-1][2]
        subtoken = token[b:]
        score = self.scores.get(subtoken, self.ds)
        return [(subtoken, b, len(token), score, len(subtoken))]

    def _add_last_subtoken(self, token, result):
        e = result[0][1]
        subtoken = token[0:e]
        score = self.scores.get(subtoken, self.ds)
        return [(subtoken, 0, e, score, e)]
	class BranchingEntropy:
	def __init__(self, min_length=2, max_length=7):

	self.min_length = min_length
	self.max_length = max_length

	self.encoder = IntegerEncoder()

	self.L = defaultdict(lambda: defaultdict(int))
	self.R = defaultdict(lambda: defaultdict(int))

	def get_all_access_variety(self):

	av = {}
	#words = set(self.L.keys())
	#words += set(self.R.keys())
	words = set(list(set(self.L.keys())) + list(set(self.R.keys()))) # 내가 수정

	for word in words:
	av[word] = self.get_access_variety(word)

	return av

	def get_access_variety(self, word, ignore_space=False):

	return (len(self.get_left_branch(word, ignore_space)), len(self.get_right_branch(word, ignore_space)))

	def get_all_branching_entropies(self, ignore_space=False):

	be = {}
	words = set(self.L.keys())
	for word in self.R.keys():
	words.add(word)

	for word in words:
	be[self.encoder.decode(word)] = self.get_branching_entropy(word, ignore_space)

	return be

	def get_branching_entropy(self, word, ignore_space=False):

	be_l = self.entropy(self.get_left_branch(word, ignore_space))
	be_r = self.entropy(self.get_right_branch(word, ignore_space))
	return (be_l, be_r)

	def entropy(self, dic):

	if not dic:
	return 0.0

	sum_count = sum(dic.values())
	entropy = 0

	for freq in dic.values():
	prob = freq / sum_count
	entropy += prob * math.log(prob)

	return -1 * entropy

	def get_left_branch(self, word, ignore_space=False):

	if isinstance(word, int):
	word_index = word
	else:
	word_index = self.encoder.encode(word)

	if (word_index == -1) or (not word_index in self.L):
	return {}

	branch = self.L[word_index]

	if ignore_space:
	return {w: f for w, f in branch.items() if not ' ' in self.encoder.decode(w, unknown=' ')}
	else:
	return branch

	def get_right_branch(self, word, ignore_space=False):

	if isinstance(word, int):
	word_index = word
	else:
	word_index = self.encoder.encode(word)

	if (word_index == -1) or (not word_index in self.R):
	return {}

	branch = self.R[word_index]

	if ignore_space:
	return {w: f for w, f in branch.items() if not ' ' in self.encoder.decode(w, unknown=' ')}
	else:
	return branch

	def counter_size(self):
	return (len(self.L), len(self.R))

	def prune_extreme_case(self, min_count):

	# TODO: encoder remove & compatify
	before_size = self.counter_size()
	self.L = defaultdict(lambda: defaultdict(int),
	{word: dic for word, dic in self.L.items() if sum(dic.values()) > min_count})
	self.R = defaultdict(lambda: defaultdict(int),
	{word: dic for word, dic in self.R.items() if sum(dic.values()) > min_count})
	after_size = self.counter_size()

	return (before_size, after_size)

	def train(self, sents, min_count=5, num_for_pruning=10000):

	for num_sent, sent in enumerate(sents):

	sent = sent.strip()
	if not sent:
	continue

	sent = ' ' + sent.strip() + ' '
	length = len(sent)

	for i in range(1, length - 1):
	for window in range(self.min_length, self.max_length + 1):

	if i + window - 1 >= length:
	continue

	word = sent[i:i + window]
	if ' ' in word:
	continue

	word_index = self.encoder.fit(word)

	if sent[i - 1] == ' ':
	left_extension = sent[max(0, i - 2):i + window]
	else:
	left_extension = sent[i - 1:i + window]

	if sent[i + window] == ' ':
	right_extension = sent[i:min(length, i + window + 2)]
	else:
	right_extension = sent[i:i + window + 1]

	if left_extension == None or right_extension == None:
	print(sent, i, window)

	left_index = self.encoder.fit(left_extension)
	right_index = self.encoder.fit(right_extension)

	self.L[word_index][left_index] += 1
	self.R[word_index][right_index] += 1

	if (num_for_pruning > 0) and ((num_sent + 1) % num_for_pruning == 0):
	before, after = self.prune_extreme_case(min_count)
	sys.stdout.write('\rnum sent = %d: %s --> %s' % (num_sent, str(before), str(after)))

	if (num_for_pruning > 0) and ((num_sent + 1) % num_for_pruning == 0):
	self.prune_extreme_case(min_count)
	sys.stdout.write('\rnum_sent = %d: %s --> %s' % (num_sent, str(before), str(after)))

	def load(self, model_fname, encoder_fname):

	self.encoder.load(encoder_fname)

	try:
	with open(model_fname, encoding='utf-8') as f:

	next(f) # SKIP: parameters (min_length, max_length)
	token = next(f).split()
	self.min_length = int(token[0])
	self.max_length = int(token[1])

	next(f) # SKIP: left side extension
	is_right_side = True

	for line in f:

	if '# right side extension' in line:
	is_right_side = True
	continue

	token = line.split();
	word = int(token[0])
	extension = int(token[1])
	freq = int(token[2])

	if is_right_side:
	self.R[word][extension] = freq
	else:
	self.L[word][extension] = freq

	except Exception as e:
	print(e)

	def save(self, model_fname, encoder_fname):

	self.encoder.save(encoder_fname)

	try:
	with open(model_fname, 'w', encoding='utf-8') as f:

	f.write("# parameters (min_length max_length)\n")
	f.write('%d %d\n' % (self.min_length, self.max_length))

	f.write('# left side extension\n')
	for word, extension_dict in self.L.items():
	for extension, freq in extension_dict.items():
	f.write('%d %d %d\n' % (word, extension, freq))

	f.write('# right side extension\n')
	for word, extension_dict in self.R.items():
	for extension, freq in extension_dict.items():
	f.write('%d %d %d\n' % (word, extension, freq))

	except Exception as e:
	print(e)

	def words(self):
	return set(self.encoder.inverse)




	class MaxScoreTokenizer:
	def __init__(self, max_length=10, scores={}, default_score=0.0):
	self.max_length = max_length
	self.scores = scores
	self.ds = default_score

	def tokenize(self, sentence):
	#return [self._recursive_tokenize(token) for token in sentence.split()]
	tmp = [self._recursive_tokenize(token) for token in sentence.split()]
	result = []
	for words in tmp:
	for tokens in words:
	result.append(tokens[0])
	return result

	def _recursive_tokenize(self, token, range_l=0, debug=False):

	length = len(token)
	if length <= 2:
	return [(token, 0, length, self.ds, length)]

	if range_l == 0:
	range_l = min(self.max_length, length)

	scores = self._initialize(token, range_l, length)
	if debug:
	pprint(scores)

	result = self._find(scores)

	adds = self._add_inter_subtokens(token, result)

	if result[-1][2] != length:
	adds += self._add_first_subtoken(token, result)

	if result[0][1] != 0:
	adds += self._add_last_subtoken(token, result)

	return sorted(result + adds, key=lambda x: x[1])

	def _initialize(self, token, range_l, length):
	scores = []
	for b in range(0, length - 1):
	for r in range(2, range_l + 1):
	e = b + r

	if e > length:
	continue

	subtoken = token[b:e]
	score = self.scores.get(subtoken, self.ds)
	scores.append((subtoken, b, e, score, r))

	#return sorted(scores, key=lambda x: (x[3], x[4]), reverse=True)
	return sorted(scores, key=lambda x: (x[0], x[1]), reverse=True)

	def _find(self, scores):
	result = []
	num_iter = 0

	while scores:
	word, b, e, score, r = scores.pop(0)
	result.append((word, b, e, score, r))

	if not scores:
	break

	removals = []
	for i, (_1, b_, e_, _2, _3) in enumerate(scores):
	if (b_ < e and b < e_) or (b_ < e and e_ > b):
	removals.append(i)

	for i in reversed(removals):
	del scores[i]

	num_iter += 1
	if num_iter > 100: break

	return sorted(result, key=lambda x: x[1])

	def _add_inter_subtokens(self, token, result):
	adds = []
	for i, base in enumerate(result[:-1]):
	if base[2] == result[i + 1][1]:
	continue

	b = base[2]
	e = result[i + 1][1]
	subtoken = token[b:e]
	adds.append((subtoken, b, e, self.ds, e - b))

	return adds

	def _add_first_subtoken(self, token, result):
	b = result[-1][2]
	subtoken = token[b:]
	score = self.scores.get(subtoken, self.ds)
	return [(subtoken, b, len(token), score, len(subtoken))]

	def _add_last_subtoken(self, token, result):
	e = result[0][1]
	subtoken = token[0:e]
	score = self.scores.get(subtoken, self.ds)
	return [(subtoken, 0, e, score, e)]