import math import random import multiprocessing as mp from multiprocessing.managers import ListProxy from config import ConfigParser def get_symbol(alphabet: dict): """ Генератор создающий последовательность из полученных символов с заданной веростяность и при каждом запросе возвращает случайный :param alphabet словарь символов из которых будет генерится строка с заданной вероятностью для каждого из символов """ seq = sum([[key] * alphabet[key] for key in alphabet], []) while True: yield random.choice(seq) def generate_str(symbols, str_len: int): """ Генерирует строку из заданных символов с указанной вероятностью :param alphabet словарь символов из которых будет генерится строка с заданной вероятностью для каждого из символов :param str_len длинна генерируемой строки """ return ''.join([next(symbols) for _ in range(str_len)]) def calc_entropy(alphabet: dict): return sum([-p / 100 * math.log(p / 100, 2) for p in alphabet.values()]) def calc_parts(str_len: int): """ Считает какой длинны генерировать строку каждому из процессов :param str_len: общая длинна строки """ parts = [str_len // mp.cpu_count()] * mp.cpu_count() diff = str_len - sum(parts) if diff != 0: parts[-1] += diff return parts def wrapper_generate_str(symbols, str_len: int, out: ListProxy): out.append(generate_str(symbols, str_len)) conf = ConfigParser() symbols = get_symbol(conf.alphabet) all_process = [] manager = mp.Manager() out = manager.list() for length in calc_parts(conf.length): process = mp.Process( target=wrapper_generate_str, args=(symbols, length, out) ) process.start() all_process.append(process) for i in all_process: i.join() text = ''.join(out) # print(text) print('Энтропия равна: {}'.format(calc_entropy(conf.alphabet)))