izekia я даже не думала кидать камни в сторону python. Это один из моих любимых языков, просто столкнулась с ситуацией, когда выгодней и проще использовать другой язык, где без использования сторонних библиотек и в лоб задача решается оптимально по скорости и потребляемым ресурсам. Все таки для каждой задачи свой инструмент.
import math import queue import threading import terminaltables import numpy as np from config import ConfigParser def calc_entropy(alphabet: dict): return sum([-p * math.log(p, 2) for p in alphabet.values()]) def calc_parts(str_len: int, parts: int): """ Считает какой длинны генерировать строку каждому из процессов :param str_len: общая длинна строки """ parts = [str_len // parts] * parts diff = str_len - sum(parts) if diff != 0: parts[-1] += diff return parts def generate_part_text(alphabet, length, res_str, res_frequency): """ Генерирует строку с заданной вероястностью появления символов и считает сколько раз каждый из символов встречается в строке :param alphabet: :param length: длинна строки, которую надо сгенерировать :param res_str: очередь в которую помещается массив сгенеровынных символов :param res_frequency: очередь в котрую помещается словарь со списком символов и частотой его встречаемости """ part = np.random.choice( list(alphabet.keys()), p=list(alphabet.values()), size=length ) res_str.put_nowait(part) res_frequency.put_nowait(dict(zip(*np.unique(part, return_counts=True)))) def generate_text(alphabet: dict, length: int): result_str = queue.Queue() result_frequency = queue.Queue() count_thread = 6 if length < 10000000: count_thread = 1 generate_part_text(alphabet, length, result_str, result_frequency) else: all_tasks = [] for i in calc_parts(length, count_thread): thread = threading.Thread( target=generate_part_text, args=(alphabet, i, result_str, result_frequency) ) thread.start() all_tasks.append(thread) for i in all_tasks: i.join() # Объединятся результаты расчитанные в потоках text = np.array([], dtype='<U1') frequency = {} for i in range(count_thread): text = np.append(text, result_str.get_nowait()) for key, value in result_frequency.get_nowait().items(): frequency[key] = frequency.setdefault(key, 0) + value return text, frequency conf = ConfigParser() text, frequency = generate_text(conf.alphabet, conf.length) with open('out.txt', 'w') as out: out.write(text.view('U{}'.format(conf.length))[0]) data_table = [['Символ', 'Количество выпадений']] data_table.extend(frequency.items()) table = terminaltables.AsciiTable(data_table) print(table.table) print('Общая длинна строки: {}\nЭнтропия: {}'.format( conf.length, calc_entropy(conf.alphabet)) )
модуль config
import os import configparser from collections import OrderedDict BASE_DIR = os.path.abspath(os.path.dirname(__file__)) class ConfigParser: def __init__(self): self._config = configparser.ConfigParser() self.length = 0 self.alphabet = OrderedDict() self.read() def read(self): self.alphabet.clear() self._config.read(os.path.join(BASE_DIR, 'config.conf')) for section in self._config.sections(): for i in self._config[section]['alphabet'].split(', '): i = i.split(':') self.alphabet[i[0]] = float(i[1]) if sum([i for i in self.alphabet.values()]) != 1.0: print('Сумма вероятностей должна быть равна 1.0') exit() self.length = int(self._config[section]['length'])
[general]
alphabet = a:0.1, b:0.3, c:0.6
length = 100000000