<commit_before>import logging
import operator

from mapreduce.driver import MapReduceDriver
from mapreduce.emblem_freq import EmblemFreq
from nlp.emblem import Emblem
from processor.data_source import MongoDataSource

data_source = MongoDataSource()
songci_list = data_source.find()

logger = logging.getLogger('emblem_stat')
logging.basicConfig(level=logging.INFO)


def stat_freq():
    freq_threshold = 1

    driver_freq = MapReduceDriver(EmblemFreq.map_fn, EmblemFreq.reduce_fn)
    emblem_list = Emblem(songci_list).emblem_list()
    emblem_freq_stat = driver_freq(emblem_list)
    emblem_freq_stat.sort(key=operator.itemgetter(1), reverse=True)

    total_len = len(emblem_freq_stat)
    logger.info('Total length of emblem_freq_stat is %d', total_len)
    prev_freq = prev_freq_rate = 0  # cache previous quotient (a.k.a. freq) to improve performance
    for emblem_freq in emblem_freq_stat:  # todo: optimize this using concurrency
        emblem_name, freq = emblem_freq
        if freq <= freq_threshold:
            break
        freq_rate = prev_freq_rate if freq == prev_freq else freq / total_len
        data_source.save('emblem', {'name': emblem_name}, {'freq_rate': freq_rate})
        prev_freq = freq
        prev_freq_rate = freq_rate

    data_source.create_index('emblem', 'name', unique=True)
    data_source.create_index('emblem', 'freq_rate')


if __name__ == '__main__':
    stat_freq()<commit_msg>optimize freq_stat.<commit_after>import logging
import multiprocessing
import operator
from itertools import repeat

from mapreduce.driver import MapReduceDriver
from mapreduce.emblem_freq import EmblemFreq
from nlp.emblem import Emblem
from processor.data_source import MongoDataSource

data_source = MongoDataSource()
songci_list = data_source.find()

logger = logging.getLogger('emblem_stat')
logging.basicConfig(level=logging.INFO)


def stat_freq():
    driver_freq = MapReduceDriver(EmblemFreq.map_fn, EmblemFreq.reduce_fn)
    emblem_list = Emblem(songci_list).emblem_list()
    emblem_freq_stat = driver_freq(emblem_list)
    emblem_freq_stat.sort(key=operator.itemgetter(1), reverse=True)

    total_len = len(emblem_freq_stat)
    logger.info('Total length of emblem_freq_stat is %d', total_len)

    workers = (multiprocessing.cpu_count() or 1) * 4
    pool = multiprocessing.Pool(processes=workers)
    emblem_freq_chunks = MapReduceDriver.chunks(emblem_freq_stat, int(total_len / workers))
    pool.starmap(save_freq_stat, zip(emblem_freq_chunks, repeat(total_len)))

    data_source.create_index('emblem', 'name', unique=True)
    data_source.create_index('emblem', 'freq_rate')


def save_freq_stat(freq_stat, total_len):
    freq_threshold = 1
    prev_freq = prev_freq_rate = 0  # cache previous quotient (a.k.a. freq) to improve performance
    for emblem_freq in freq_stat:
        emblem_name, freq = emblem_freq
        if freq <= freq_threshold:
            break
        freq_rate = prev_freq_rate if freq == prev_freq else freq / total_len
        data_source.save('emblem', {'name': emblem_name}, {'freq_rate': freq_rate})
        prev_freq = freq
        prev_freq_rate = freq_rate


if __name__ == '__main__':
    stat_freq()
{'fraction_non_alphanumeric': 0.09002203336480957, 'fraction_numerical': 0.0025180988353792886, 'mean_word_length': 4.565674255691769, 'pattern_counts': {'":': 0, '<': 5, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7357568', 'n_tokens_mistral': 1080, 'n_tokens_neox': 1048, 'n_words': 222}