Давайте начнём с азов, если брать определение из всем известной и всеми любимой Википедии, то L-система (или же система Линденмайера) это параллельная система переписывания и вид формальной грамматики.

Если говорить простым языком, то L-система состоит из алфавита символов, которые могут быть использованы для создания строк, набора порождающих правил, которые задают правила подстановки вместо каждого символа, начальной строки ( аксиомы ), с которой начинается построение, и механизм перевода образованной строки в геометрические структуры. Самым простым примером L-системы может служить задача построения дерева.

Вводные данные:

Строка (далее Аксиома): A B

Переменные (которые мы можем задействовать в построении дерева): A B C

Правило (правило по которому каждая переменная на последующие строке меняется):

• A - > AB

• B - > AC

• C - > A

Получаются такие преобразования:

Основным направлением, в котором применяются L-системы, это моделирование процессов роста как живых организмов, так и неживых объектов (кристаллов, раковин моллюсков или пчелиных сот).

Пример:

Для моделирования подобных процессов, мы с вами будем использовать такой язык программирования как Python, в нём есть встроенная библиотека turtle.

Итак, приступим:

1. Здесь мы импортируем библиотеку Turtle в наш проект:

   import turtle

2. Далее мы подключаем все необходимые конфигурации для нашей черепашки:

   turtle.hideturtle()

   turtle.tracer(1)

   turtle.penup()

   turtle.setposition(-300,340)

   turtle.pendown()

   turtle.pensize(1)

3. Далее мы задаём значение непосредственно самой аксиомы и сопутствующих параметров, необходимых для задания последовательности:

   axiom = "F+F+F+F"

   tempAx = ""

   itr = 3

   (itr- значения итераций цикла, оно нам понадобится в следующем шаге при написании нашей программы)

4. В данном цикле, где как раз нам и понадобится переменная itr-, мы занимаемся обработкой и "выращиванием" непосредственно генома нашего фрактала/растения:

   for k in range(itr):

   for ch in axiom:

   if ch == "+":

   tempAx = tempAx + "+"

   elif ch == "-":

   tempAx = tempAx + "-"

   elif ch == "F": #F

   tempAx = tempAx + "F+F-f-F+F"

   else:

   tempAx = tempAx + "f"

   axiom = tempAx

   tempAx = " "

   print(axiom)

   Если мы с вами пробежимся по циклу, то сразу же в первом условии мы увидим фильтр на символ "+":

   if ch == "+":

   tempAx = tempAx + "+"

   Здесь мы ищем в аксиоме (изначальной строке) знак + и при его появлении мы добавляем символ + в последующую строчку. Так же происходит и с символом - и f, мы просто добавляем в последующую строчку символы - и f соответственно. Но при появлении в нашей аксиоме символа F, мы поступим немного иначе и добавим в последующую строчку уже последовательность символов F + F f F + F, для увеличения длины каждой последующей строки. Данное действие в принципе не принципиально, но для быстроты генерации генома, я решил поступить именно так. В конце мы обязательно приравниваем аксиому к нашему геному и обнуляем его:

   axiom = tempAx

   tempAx = " "

   Результат (значение аксиомы):

   

5. Ну что, вроде по генерации генома всё понятно, давайте уже приступим непосредственно к визуализации. Для этого нам понадобиться простенький цикл, который будет переводить встречающиеся в геноме символы в движения черепашки:

   for ch in axiom:

   if ch == "+":

   turtle.right(45)

   turtle.forward(10)

   turtle.right(45)

   elif ch == "-":

   turtle.left(45)

   turtle.forward(10)

   turtle.left(45)

   else:

   turtle.forward(20)

   Сразу можно заметить, что по аналогии с предыдущим пунктом, мы детектируем символы "+", "-", "F" и "f". Только теперь в момент, когда мы встречаем символ "+":

   if ch == "+":

   turtle.right(45)

   turtle.forward(10)

   turtle.right(45)

   Мы поворачиваем сначала на право, на 45 градусов, потом проезжаем расстояние, равное 10, и потом мы снова поворачиваем на право, на 45 градусов. Когда же мы встречаем символ "-":

   elif ch == "-":

   turtle.left(45)

   turtle.forward(10)

   turtle.left(45)

   Мы делаем все те же самые действия, что и при символе "+", только на этот раз поворачиваем уже не вправо, а уже влево. Но если же мы встретим символы "F" или "f", то мы просто буем проезжать вперёд на 20 пикселей:

   else:

   turtle.forward(20)

   В итоге мы получаем вот такой фрактал-снежинку:

   

6. Если же мы захотим раскрасить наш фрактал-снежинку, то нам понадобиться перед циклом из предыдущего пункта добавить строчки:

   turtle.fillcolor("#99BBFF")

   turtle.begin_fill()

   Где #99BBFF - это кодировка цвета (RGB: 153, 187, 255), а begin_fill() это начало заполнения цветом. И в конце, уже после цикла, добавить строчки:

   turtle.end_fill()

   turtle.update()

   turtle.done()

   А end_fill() означает конец заполнения. Далее мы обновляем и выключаем нашу "черепашку". И на выходе мы получаем вот такой фрактал-снежинку:

   

Вы так же можете посмотреть, "потыкать" код, изменяя параметры и данные в нём.

В конце, хочу также добавить, что мне очень понравилось работать и писать по данной тематике, возможно, в будущем, я напишу ещё ряд статей по теме "L-системы", но а пока, хочу представить вам результаты моего тыканья творчества:

Вчера я зарелизил свой первый Python фреймворк. Нет, не еще один. Это в мире - еще один. А для меня пока что первый. И я допускаю, что он первый в своем роде. Это фреймворк для создания кастомных серверов. И создаваться они будут через конфиг. Ух, насоздаем сейчас...

Вначале был конфиг

Итак, конфиг. Поскольку к этому моменту фреймворк мы уже установили. А если нет, то это легко и просто делается командой:

pip3 install idewavecore==0.0.1

Это при условии наличия у вас Python 3.6+, интернета и компьютера.

Сам конфиг при этом выглядит примерно вот так:

# settings.ymlsettings:  servers: !inject config_dir/servers.yml  db_connections:    sqlite:      host: ~      username: ~      password: ~      # default mysql 3306, postgresql 5432, sqlite don't need port      port: ~      # currently allowed: mysql, postgresql, sqlite      dialect: sqlite      # supported drivers:      # mysql: mysqlconnector, pymysql, pyodbc      # postgresql: psycopg2, pg8000, pygresql      driver: ~      # to use with sqlite this should be absolute db path      # can be empty to keep db in memory (sqlite only)      db_name: ~      charset: ~

Где !inject - это специальный тэг для импорта других yaml файлов. Что очень удобно, если нужно разбить огромную yaml-простыню на набор аккуратных мини-конфигов.

# servers.ymlsample_server:  connection:    host: 1.2.3.4    port: 1234    # possible values: tcp, websocket    connection_type: tcp  # optional  proxy:    host: ~    port: ~    # possible values: tcp, websocket    connection_type: tcp  options:    server_name: Sample Server    is_debug: false  middlewares: !pipe    - !fn native.test.mock_middleware    - !fn native.test.mock_middleware    - !infinite_loop        - !fn native.test.mock_middleware        - !fn native.test.mock_middleware        - !fn native.test.mock_middleware        - !router            ROUTE_1: !fn native.test.mock_middleware            ROUTE_2: !fn native.test.mock_middleware            ROUTE_3:              - !fn native.test.mock_middleware              - !fn native.test.mock_middleware              - !fn native.test.mock_middleware  # optional  db_connection: sqlite

Здесь уже побольше тэгов.

!pipe - это специальный тэг, который создает обертку над функциями в массиве. На данный момент все функции, используемые в разделе middlewares (о них - чуть ниже), должны быть обернуты в пайп. Вкратце - пайп выполняет поочередно переданные в него функции и пробрасывает в них необходимые параметры.

!infinite_loop - это специальный тэг, который автоматически создает пайп из переданных функций и затем выполняет их внутри вечного цикла. Что может быть полезно для создания непрерывного соединения (например, по websocket).

!router - это специальный тэг, который создает маршрутизируемый пайп. Фактически, создается словарь пайпов и какой из них выполнить, роутер определяет по специальному параметру (route).

И, наконец, !fn - специальный тэг, который позволяет импортировать функцию (далее - middleware) по указанному пути. Мой фреймворк предоставляет некоторое количество миддлвэров, но можно написать свои - достаточно создать в корне своего проекта одноименную папку - middlewares - и далее создать там файлы с необходимыми функциями. Далее, чтобы использовать вашу функцию, достаточно будет вызвать тэг:

!fn <имя_вашего_файла>.<имя_функции>

И строка будет преобразована в миддлвэр. Название корневой папки (middlewares) указывать не нужно - оно будет подставлено автоматически при импорте. Если же вы хотите использовать нативные миддлвэры фреймворка (есть! я использовал три заимствованных слова подряд!), достаточно указать префикс native в пути к функции, например:

!fn native.test.mock_middleware

В целом, большой акцент сделан именно на работу с конфигом.

Middle where

Миддлвэры - это место, где будет сосредоточена вся кастомизируемая логика вашего приложения. Фактически, это - центр. Сердце каждого отдельно созданного сервера.

В общих чертах, миддлвэр - это асинхронная функция, имеющая доступ ко всем хранилищам (о них - ниже). Каждая такая функция должна выглядеть примерно вот так:

from idewavecore.session import Storage, ItemFlagasync def sample_middleware(**kwargs):    global_storage: Storage = kwargs.pop('global_storage')    server_storage: Storage = kwargs.pop('server_storage')    session_storage: Storage = kwargs.pop('session_storage')    session_storage.set_items([        {            'key1': {                'value': 'some_tmp_value'            }        },        {            'key2': {                'value': 'some_persistent_value',                'flags': ItemFlag.PERSISTENT            }        },        {            'key3': {                'value': 'some_persistent_constant_value',                'flags': ItemFlag.PERSISTENT | ItemFlag.FROZEN            }        },        {            'key4': {                'value': 'some_constant_value',                'flags': ItemFlag.FROZEN            }        }    ])    value_of_key3 = session_storage.get_value('key3')

Каждый миддлвэр имеет доступ к одному из трех типов хранилищ (storage). Хранилища бывают трех типов: глобальное (global storage), хранилище сервера (server storage) и хранилище сессии (session storage).

Глобальное хранилище связывает все части приложения между собой, позволяя серверам получить доступ к общим данным.

Хранилище сервера используется для хранения соединений и в основном используется для броадкаста.

Хранилище сессий создается персонально для каждого клиентского соединения и хранит данные в пределах этого соединения.

В основном в миддлвэрах будет использоваться именно хранилище сессии.

(После третьего раза я понял, что больше не могу так. Простите за такой топорный перевод термина. Я не придумал, как по-другому. Предлагайте свои варианты в комментариях - возможно, эту ситуацию еще не поздно исправить.)

Запускаем...

Теперь наконец пора применить то, без чего не обходится ни одно нормальное приложение на базе моего фреймворка - Assembler. Не совсем тот, но тоже производит сборку. Примерно так:

# run.pyimport asynciofrom idewavecore import Assemblerfrom idewavecore.session import Storageif __name__ == '__main__':    loop = asyncio.get_event_loop()    global_storage = Storage()    assembler = Assembler(        global_storage=global_storage,        # ваш путь к конфигу        config_path='settings.yml'    )    servers = assembler.assemble()    for server in servers:        server.start()    loop.run_until_complete(        asyncio.gather(*[server.get() for server in servers])    )    try:        loop.run_forever()    except KeyboardInterrupt:        pass    finally:        loop.close()

Запускаем в терминале - и можем лицезреть (при условии, что вы все сделали правильно) сообщения о том, что серверы запущены. Теперь к ним можно пробовать достучаться всеми возможными способами - браузер, curl, клиент mmo rpg игры...

Что теперь

А теперь я хочу сказать спасибо всем, кто прочитал мой пост до конца. Я буду благодарен за любые конструктивные замечания. Даже если мой фреймворк - не первый в своем роде, я хочу его максимально приблизить к этой планке.

Присоединяйтесь https://github.com/idewave/idewavecore.

При проведении проверок с использованием моделей машинного обучения одной из часто решаемых задач является задача кластеризации. Например, необходимо разбить на несколько кластеров отзывы клиентов на мобильное приложение (задача тематического моделирования). Для задач кластеризации часто используют модель k-means. Это обусловлено её простотой и понятностью. Однако, у этого алгоритма есть один большой недостаток - необходимость изначально задать число кластеров. Эта проблема прекрасно решается с помощью расширяющегося нейронного газа.

Расширяющийся нейронный газ строит граф, пытаясь приблизить распределение данных. Не связанные подграфы этого графа это наши искомые кластеры. Он строится по следующему алгоритму:

1) Генерация первых двух нейронов случайным образом

2) На каждом шаге итерационного процесса берется один элемент данных. Два ближайших к нему нейрона двигаются в его сторону

3) Между наиболее часто перемещающемуся нейроном и его ближайшим соседом создается новый нейрон

4) Удаляются связи, если соединенные им нейроны вместе не передвигаются, и нейроны без связей

Рассмотрим этот итерационный алгоритм на примере со следующими данными:

В самом начале построения графа случайным образом задаются первые два нейрона s1 и s2.

После этого начинается итерационный процесс:

1. Выбирается один элемент наших данных v1.

2. Выбирается два ближайших нейрона. Они перемещаются на r1 и r2 соответственно ближе к данному элементу, где r1 > r2.

3. Следующие три итерации s2 поменяет свое местоположение сильнее, чем s1. Значит в окрестности s2 большая плотность данных, и нужно создать новый нейрон s3 посередине между s2 и его ближайшим соседом s1. Связь между s1 и s2 удаляется.

4. После еще 3 итераций нейрон s1 никаким образом не изменит своего положения. Значит он не помогает приблизить распределение наших данных. Сначала удаляется его связь с s3, а потом и он сам

5. За следующие 3 итерации мы столкнемся с такой же проблемой, как в пункте 3 и нам понадобится создать s4. В результате получится граф s2-s3-s4, приближающий распределение наших данных

В результате получается граф с несколькими не связанными подграфами, повторяющие распределение наших данных. Их число можно использовать как искомое количество кластеров для k-means.

Эту гипотезу нужно проверить на реальных данных.

Для начала возьмем стандартный набор данных sklearn c двумя полумесяцами:

from sklearn.datasets import make_moonsdata, _ = make_moons(10000, noise=0.06, random_state=0)plt.scatter(*data.T)plt.show()

Переобучением нам потребуется создать несколько своих функций для визуализации графа и подсчета несвязанных графов в нем:

import copyfrom neupy import algorithms, utilsdef draw_image(graph, show=True):    for node_1, node_2 in graph.edges:        weights = np.concatenate([node_1.weight, node_2.weight])        line, = plt.plot(*weights.T, color='black')        plt.setp(line, linewidth=0.2, color='black')    plt.xticks([], [])    plt.yticks([], [])        if show:       plt.show()def create_gng(max_nodes, step=0.2, n_start_nodes=2, max_edge_age=50):    return algorithms.GrowingNeuralGas(        n_inputs=2,        n_start_nodes=n_start_nodes,        shuffle_data=True,        verbose=True,        step=step,        neighbour_step=0.005,        max_edge_age=max_edge_age,        max_nodes=max_nodes,        n_iter_before_neuron_added=100,        after_split_error_decay_rate=0.5,        error_decay_rate=0.995,        min_distance_for_update=0.01,    )def extract_subgraphs(graph):    subgraphs = []    edges_per_node = copy.deepcopy(graph.edges_per_node)        while edges_per_node:        nodes_left = list(edges_per_node.keys())        nodes_to_check = [nodes_left[0]]        subgraph = []                while nodes_to_check:           node = nodes_to_check.pop()            subgraph.append(node)            if node in edges_per_node:                nodes_to_check.extend(edges_per_node[node])                del edges_per_node[node]                    subgraphs.append(subgraph)            return subgraphs

В результате обучения расширяющийся нейронный газ точно сохранил форму данных всего лишь с 500 элементами, хотя изначальный набор данных состоял из 10000, и точно определил, что здесь два кластера.

utils.reproducible()gng = create_gng(max_nodes=500)for epoch in range(20):    gng.train(data, epochs=1)draw_image(gng.graph)    print("Found {} clusters".format(len(extract_subgraphs(gng.graph))))

К сожалению, такие хорошо структурированные данные редко встречаются в реальных задачах.

Для искусственной имитации реальной ситуации создадим набор данных с 3 кластерами со случайным образом разбросанными элементами:

X = -0.7 - 2.5 * np.random.rand(900,2)X1 = 0.7 + 2.5 * np.random.rand(375,2)X2 = -0.5 + 1.7 * np.random.rand(50,2)X[475:850, :] = X1X[850:900, :] = X2plt.scatter(X[ : , 0], X[ :, 1])plt.show()

Несмотря на не структурированность данных и неявные границы между ними, расширяющийся нейронный газ и здесь смог правильно приблизить распределение и определить количество кластеров.

utils.reproducible()gng = create_gng(max_nodes=300)for epoch in range(40):    gng.train(X, epochs=1)    draw_image(gng.graph)    print("Found {} clusters".format(len(extract_subgraphs(gng.graph))))

Привет, Хабр! Хочу вам рассказать о своём исследовательском проекте, в котором я создал игрового ботеца для ВКонтакте.

Ахтунг!

Я не являюсь профессиональным разработчиком, я обычный девятиклассник, любящий иногда покодить на совершенно разных язык. Здесь я просто рассказываю о своём проекте и опыте участия в конкурсе проектных работ.

Эта статья не является пошаговым руководством по созданию бота для ВКонтакте - их достаточно и на Хабре, и за его пределами.

Что за проект?

Я, как администратор немаленькой беседы во ВКонтакте (на тот момент это было ~670 человек), столкнулся с проблемой ужасной активности. Ну серьёзно, человек много, а не пишет никто, очень много незаинтересованных молчунов, которые лишь цифру создают для беседы. Подавляющее большинство пользователей из тех 670 человек просто входили и не появлялись в списке сообщений ни разу.

Чтобы решить эту проблему, я решил попробовать написать своего бота с некоторыми игровыми механиками. Сразу хочу оговориться, что пусть и код писал я сам, мне два раза проводил код ревью мой замечательный приятель, поэтому, пусть сам проект, сама задумка довольно простая, - это не помешало мне набраться опыта от более продвинутого в кодинге человека.

С ботом и своей темой я отправился прямиком на конкурс "Большие вызовы", съездил в лагерь, где с экспертами (а с одним из них мы общаемся до сих пор) я допиливал презентацию, чутка сменил цель, всё подкорректировал, успешно защитился. Сейчас меня ждёт только последний тур заключительного этапа, но это уже совсем другая история.

Что же за системы есть в боте?

В общем и целом, по сути я вдохновлялся другим ботом, в которого любил играть продолжительное время (и закончил только потому, что они его отрубили и делают сейчас игру в приложении ВК).

У меня бот намного проще, но подошёл для проекта в школе и для проекта в конкурсе. Основными характеристиками пользователя являются игровая валюта и опыт. Фактически, всё сводится к фарму опыта и денег ради фарма опыта и денег. За опыт можно устроиться на работу, каждая последующая работа будет давать всё больше денег в 24 часа (без фарма, получение зарплаты только командой раз в сутки), в свою очередь за деньги можно купить машину получше, которая будет давать больший множитель к опыту. Опыт даётся за каждое сообщение кроме команд посимвольно, затем умножается на множитель от автомобиля. Это самое основное, чего я стал требовать от бота. Также есть выдача предупреждений (максимальное их количество - 4).

Да, такого бота я делал около 6 месяцев, постоянно что-то изменяя, добавляя и удаляя.

Давай о реализации уже!

Делал я бота на языке программирования Python с использованием асинхронной библиотеки для написания ботов vkbottle. Работает на CallBack API, в качестве сервера использую aiohttp.

Во входном файле bot.py происходит инициализация сервера, прописаны все роуты (их всего 4 и они очень маленькие), к основному боту добавляются blueprint'ы, о которых пойдёт речь позже.

import pathlibimport aiohttpimport aiohttp_jinja2import jinja2from aiohttp import webimport utils.constsfrom config import SECRET, WEBHOOK_ACCEPT, CONFIRMATION_TOKENfrom routes import actions, admin_realize, global_admin_realize, users_realize, economic_realizefrom utils.db_methods import init_databasefrom middlewares import ExpMiddleware # dead import for include middlewareINDEX_DIR = str(pathlib.Path(__file__).resolve().parent) + '/index_page'utils.consts.BOT.loop.run_until_complete(init_database())utils.consts.BOT.set_blueprints(    actions.bp, admin_realize.bp, global_admin_realize.bp,    users_realize.bp, economic_realize.bp)APP = aiohttp.web.Application()ROUTES = aiohttp.web.RouteTableDef()if not WEBHOOK_ACCEPT:    aiohttp_jinja2.setup(APP, loader=jinja2.FileSystemLoader(str(INDEX_DIR)))    APP.router.add_static('/static/',                          path=str('./index_page/'),                          name='static')@ROUTES.get("/")@aiohttp_jinja2.template('index.html')async def hello(request):    """Root site response"""    return {}@ROUTES.get("/when_update")@aiohttp_jinja2.template('whenupdate.html')async def whenupdate(request):    """When update site response"""    return {}

Все конфиги хранятся в config.py, точнее, там инициализируются константы. Сами значения хранятся в файле .env и с помощью библиотеки dotenv берутся из виртуального окружения по ключу.

import osfrom dotenv import load_dotenvdotenv_path = os.path.join(os.path.dirname(__file__), '.env')if os.path.exists(dotenv_path):    load_dotenv(dotenv_path)# Loading token from .envACCESS_TOKEN = os.getenv("ACCESS_TOKEN")SECRET = os.getenv("SECRET")USER_ACCESS_TOKEN = os.getenv("USER_ACCESS_TOKEN")WEBHOOK_ACCEPT = bool(int(os.getenv("WEBHOOK_ACCEPT", 0)))CONFIRMATION_TOKEN = os.getenv("CONFIRMATION_TOKEN")NEW_START = bool(int(os.getenv("NEW_START", 0)))ADMINS_IN_CONV = list(map(int, os.getenv("ADMINS_IN_CONV").split(',')))

Теперь о том, где хранятся все обработчики команд.

Я их разделил логически на 5 видов: обработчик событий (пользователь вошел в беседу), обработчик сообщений для всех (например, команда /profile), для администраторов беседы (например, /пред чтобы выдать предупреждение пользователю), для администраторов и модераторов бота (например, /бд добавить, чтобы добавить, как ни странно, новый экземпляр какой-то модельки, например, создать новую машину, не взаимодействуя напрямую с БД), и реализация системы экономики (купить или продать машину, поступить на работу и пр.).

Всё это хранится в пяти разных файлах в папке routes:

Вот пример команды покупки машины:

@bp.on.message_handler(AccessForAllRule(), Registered(), text="/купить_машину <c_id>")async def buy_car(message: Message, user: User, c_id: str = None):    if c_id.isdigit():        c_id = int(c_id)        car = await Car.get(id=c_id)        buy_car_user_status = status_on_buy_car(user, car)        if buy_car_user_status == BuyCarUserStatuses.APPROVED:            chat = await Conversation.get(peer_id=message.peer_id)            await User.get(user_id=message.from_id, chat=chat).update(                coins=user.coins - car.cost, car=car            )            await message(f"Машина {car} куплена!")        elif buy_car_user_status == BuyCarUserStatuses.NOT_ENOUGH_MONEY:            await message("У тебя недостаточно денег!")        elif buy_car_user_status == BuyCarUserStatuses.NOT_ENOUGH_EXP:            await message("У тебя недостаточно опыта!")        else:            await message("У тебя уже есть машина!")    else:        await message("Введите цифру-ID машины!")

Все обработчики в пределах одного файла объединяются blueprint'ом, а все "чертежи" подключаются к боту во входном файле.

В обработчиках команд используются различные служебные функции, например, в команде выше есть функция status_on_buy_car, которая говорит, можно ли пользователю купить машину, а если нельзя, то причину этого.

Все подобные функции хранятся в отдельно в папке utils в файле main.py. В этой же папке лежат файлы с константами, функциями для работы с БД, правила и ошибки, которые функции могут raise'ить иногда.

def status_on_buy_car(user: User, car: Car) -> BuyCarUserStatuses:    if user.coins >= car.cost and user.exp >= car.exp_need and user.car is None:        return BuyCarUserStatuses.APPROVED    elif user.coins < car.cost:        return BuyCarUserStatuses.NOT_ENOUGH_MONEY    elif user.exp < car.exp_need:        return BuyCarUserStatuses.NOT_ENOUGH_EXP    else:        return BuyCarUserStatuses.NOW_HAVE_CAR

В качестве ОРМки я использую Tortoise ORM, потому что асинхронно (а смысл в асинхронности фреймворка, если вся работа с БД синхронная?), потому что удобно лично для меня.

Что по итогу?

По итогу я имею бота, который некоторое время работал у меня в беседе, с помощью которого я показал небольшой прирост количества сообщений. В него иногда игрались, люди даже зарабатывали опыт общением и покупали себе имущество, но эффект, на самом деле, заметен не так сильно, так что это больше просто для развлечения, чем реально для поднятия активности с 0 до миллиона сообщений в секунду.

Конечно, в беседе, где бот был, наблюдалось больше сообщений, чем в беседе, где его не было, было больше активности (пусть и не сильно). Это я и записал в вывод.

В результате я получил небольшой научно-исследовательский проект, целью которого было исследовать возможность с помощью бота с игровой механикой повышения активности в беседах. Цель полностью выполнена, это частично возможно, но нужно пробовать различные подходы к решению поставленной проблемы.

Постскриптум

Я открыт к критике в комментариях, за неё отдельное спасибо, ведь именно критика, может даже очень жёсткая, может даже с заминусованной статьёй поможет развиваться мне как разработчику и создателю статей на Хабре.

Бот на GitHub

Давеча, с коллегой, на работе поспорили что нельзя написать кэширующий декоратор в 4 строки, я утверждал что можно. Все началось с 4-х строк, закончилось функциональным программированием кучей lambda выражений в одну строку и декоратором в одну строку.

Дисклеймер

Такой код не попадает в мои проекты или проекты моей команды, и все что описано ниже было написано в рамках академического изыскания. Я понимает что важное преимущество языка программирования python в его читабельности. Автором, из изменяющих сознания вещества, при написании данного поста, было использовано только кофе.

Пролог

Идея написать декоратор в 4 строки меня изначально никак не трогала. Было просто интересно написать декоратор. Но в процессе спортивный интерес взял верх. Изначально все начиналось с простого кэширующего декоратора.

data = {}  # словарь для сохарения кэшируемых данныхdef decor_cache(func):      def wrapper(*args):        # создаем ключ из назвнии функции которую          # кэшируем и аргументов которые передаем        key = f"{func.__name__}{args}"         # проверяем кэшировли дунную функцию с аргументами         if args in data:            return data.get(key)        else:            # если кэшируем впервые            response = func(args) # запускаем функцию с аргументами             data[key] = response # кэшируем результат             return response      return wrapper

Сейчас задача из 18 строк кода, 11 если удалить пробелы и комментарии, сделать 4 строки. Первое что приходит на ум, записать конструкцию ifelse в одну строчку.

data = {}  # словарь для сохарения кэшируемых данныхdef decor_cache(func):      def wrapper(*args):        # создаем ключ из назвнии функции которую          # кэшируем и аргументов которые передаем        key = f"{func.__name__}{args}"        if not args in data            # если кэшируем впервые            response = func(args) # запускаем функцию с аргументами             data[key] = response # кэшируем результат         return data.get(key) if args in data else response         return wrapper

Теперь у нас 15 строк кода против 18, и появился ещё один if, что создает дополнительную вычислительную нагрузку, но сегодня мы собрались не для улучшению performance. Давайте добавим в этот мир энтропии и немного copy-paste и упростим переменную key.

data = {}  # словарь для сохарения кэшируемых данныхdef decor_cache(func):      def wrapper(*args):        if not args in data            # если кэшируем впервые            response = func(args) # запускаем функцию с аргументами             data[f"{func.__name__}{args}"] = response # кэшируем результат         return data.get(f"{func.__name__}{args}") if args in data else response         return wrapper

Теперь мы имеем 12 строк, без пробелов и комментариев 8 строк. Нам пока этого не достаточно, цель 4 строчки, и надо упростить ещё. Мы помним что декораторэто функция которая должна возвращать callable объект (функцию). Функцией может быть и lambda! Значит мы можем упростить и функцию wrapper и заменить её на lambdaанонимную функцию. И возвращать из функции "декоратора", анонимную функцию.

data = {}  # словарь для сохарения кэшируемых данныхdef decor_cache(func):  cache = labda *args: data.get(f"{func.__name__}{args}") if args in data else data[f"{func.__name__}{args}"] = func(args)     return labda *args: cache(*args) if cache(*args) else data.get(f"{func.__name__}{args}")

Цель достигнута! Декоратор в 4 строки, чистого кодаполучился. Как можно увидеть одной lambda функцией не обошлось, пришлось создать две lambda функцию. Первая lambda делает две вещи: если объект уже был закешировав возвращаем ранее закешировнанное значение и кэширует объект если он не был ранее кэширован но в таком случае мы нечего не возвращаем.

Первая lambda функция теперь у нас не анонимная её пришлось записать в переменную чтобы к ней можно было обращаться по имени переменной. Вторая lambda делает примерно тоже самое что и первая только обращалась к первой lambda по имени, кеширует и получает значения из кеша, в случае если первая lambda не вернула значение, значит объект новый и он был закэширован и значит вторая lambda вернет входящие значение.

Конструкция получилась громоздкой, но все равно читаемая, более или менеепри условии если вы легко читаете lambda выражения. Много условностей, но это ещё не ад но мы к нему приближаемся. После всего проделанного пути все ещё кажется что можно ещё сократить количество строк кода. Мы уже вошли во вкус. Например те-же две lambda выражения можно совместить в одно выражение. Давайте объеденим две lambda функции в одну.

Для этого нам надо пойти на некоторое ухищрение, использовать тернарный оператор or. Тернарный оператор принимает два значения, справа и слева относительно себя, и пытается получить логический ответ True или False. Как оператор сравнения. Для того чтобы вычислить конструкцию слева и справа интерпретатор python выполнит код справа и слева. Слева у нас конструкция memory.update({f"{func.name}_{args[0]}": func(args[0])}) данное выражение вернет нам None метод update всегда будет возвращать нам None тернарный оператор воспримит этого как False и не будет это выводить, но главное что он выполнит этот код и мы обновим переменную memory. Справа у нас конструкция получения элемента по индексу из tupla, выражение простое и всегда будет давать результат, если в tuple будет запрашиваемый индекс.

data = {}  # словарь для сохарения кэшируемых данныхdef decor_cache(func):    return lambda *args: memory.get(f"{func.__name__}_{args[0]}") if f"{func.__name__}_{args[0]}" in memory else (lambda: memory.update({f"{func.__name__}_{args[0]}": func(args[0])}) or args[0])()

Отлично мы почти получили что хотели, меньше строчек кода две lambda функции в одной строке. Появляется вопрос, зачем нам здесь тогда функция decorator_cache, давайте её тоже сделаем lambda выражением но сохраним в переменную, чтобы ей можно было пользоваться по имени.

data = {}  # словарь для сохарения кэшируемых данныхdecor_cache = lambda func: lambda *args: memory.get(f"{func.__name__}_{args[0]}") if f"{func.__name__}_{args[0]}" in memory else (lambda: memory.update({f"{func.__name__}_{args[0]}": func(args[0])}) or args[0])()

Нарушая все паттерны, нам удалось создать кэширующий декоратор, почти в одну строку. Почти, потому что формально у нас есть строка объявления переменной data. Это мне не давало покоя... примерно 10 минут, пока не вспомнил что в python есть функция globals().

Функцияglobals()возвращает словарь с глобальной таблицей символов, определённых в модуле. По сути выдает словарь глобальных переменных (ключимя переменной, значениессылка на объект). Так мы получаем возможность создавать переменные в одно выражение, одной строкой. Давайте тогда для создания переменной с пустым словарем, будем использовать следующую конструкцию:

globals().update({memory: {}})

И для получения значения переменной конструкцию с get:

globals().get(memory)

После чего мы применили это на наш декоратор, и теперь действительно все получилось уместить в одну строку.

decor_cache = lambda func: lambda *args: globals().get("memory").get(f"{func.__name__}_{args[0]}") if f"{func.__name__}_{args[0]}" in globals().get("memory") else (lambda : globals().get("memory").update({f"{func.__name__}_{args[0]}": func(args[0])}) or args[0])()

Отлично, казалось мы сделали не невозможное говнокод описали декоратор в одну строку. Немного сложно, и плохо читаймо, но конкретный пример нарачито спецально был доведен до предела, для демонстрации что lambda выражение это не сложный инструмент, и в качестве доказательства продемонстривать что любой код можно записать в одну строку.

Итоги

Все выше описанное служит академическим целям. Показать что функциональный подход бывает удобным и веселым, в сравнении со скучным декларативным ООП подход. Не надо бояться или недооценивать функциональное программирование. Это отличный инструмент, который можно использовать чаще, при повседневных задачах.

Надеюсь данный пост, поможет вам или вашим коллегам полюбить функциональный подход и лучше понять как работают lambda выражения.

Настройка программного обеспечения

Без промедления начнём. Нам нужно установить следующее ПО:

• Windows 10

• Anaconda 3 (Python 3.8)

• Visual Studio 2019 (Community) - объясню позже, зачем она понадобится.

Открываем Anaconda Prompt (Anaconda3) и устанавливаем следующие пакеты:

pip install opencv-pythonpip install dlibpip install face_recognition

И уже на этом моменте начнутся проблемы с dlib.

Решаем проблему с dlib

Я перепробовал все решения, что нашёл в интернете и они оказались неактуальными - раз, два, три, официальное руководство и видео есть. Поэтому будем собирать пакет вручную.

Итак, первая же ошибка говорит о том, что у нас не установлен cmake.

ERROR: CMake must be installed to build dlib

Не закрывая консоль, вводим следующую команду:

pip install cmake

Пробуем установить пакет той же командой (pip install dlib), но на этот раз получаем новую ошибку:

Ошибка явно указывает, что у меня, скорее всего, стоит студия с элементами только для C# - и она оказывается права. Открываем Visual Studio Installer, выбираем "Изменить", в вкладке "Рабочие нагрузки" в разделе "Классические и мобильные приложения" выбираем пункт "Разработка классических приложений на С++":

"Изменить"Разработка классических приложений на С++Ждем окончания установки

Почему важно оставить все галочки, которые предлагает Visual Studio. У меня с интернетом плоховато, поэтому я решил не скачивать пакет SDK для Windows, на что получил следующую ошибку:

Я начал искать решение этой ошибки, пробовать менять тип компилятора (cmake -G " Visual Studio 16 2019"), но только стоило установить SDK, как все проблемы ушли.

Я пробовал данный метод на двух ПК и отмечу ещё пару подводных камней. Самое главное - Visual Studio должна быть 2019 года. У меня под рукой был офлайн установщик только 2017 - я мигом его поставил, делаю команду на установку пакета и получаю ошибку, что нужна свежая Microsoft Visual C++ версии 14.0. Вторая проблема была связана с тем, что даже установленная студия не могла скомпилировать проект. Помогла дополнительная установка Visual C++ 2015 Build Tools и Microsoft Build Tools 2015.

Открываем вновь Anaconda Prompt, используем ту же самую команду и ждём, когда соберется проект (около 5 минут):

Всё прошло успешно

Управляем громкостью

Вариантов оказалось несколько (ссылка), но чем проще - тем лучше. На русском язычном StackOverflow предложили использовать простую библиотеку от Paradoxis - ей и воспользуемся. Чтобы установить её, нам нужно скачать архив, пройти по пути C:\ProgramData\Anaconda3\Lib и перенести файлы keyboard.py, sound.py из архива. Проблем с использованием не возникало, поэтому идём дальше

Собираем события мыши

Самым популярным модулем для автоматизации управления мышью/клавиатурой оказался pynput. Устанавливаем так же через (pip install dlib). У модуля в целом неплохое описание - https://pynput.readthedocs.io/en/latest/mouse.html . Но у меня возникли сложности при получении событий. Я написал простую функцию:

from pynput import mousedef func_mouse():        with mouse.Events() as events:            for event in events:                if event == mouse.Events.Scroll or mouse.Events.Click:                    #print('Переместил мышку/нажал кнопку/скролл колесиком: {}\n'.format(event))                    print('Делаю половину громкости: ', time.ctime())                    Sound.volume_set(volum_half)                    break

Самое интересное, что если раскомментировать самую первую строчку и посмотреть на событие, которое привело выходу из цикла, то там можно увидеть Move. Если вы заметили, в условии if про него не слово. Без разницы, делал я только скролл колесиком или только нажатие любой клавиши мыши - все равно просто движение мыши приводит к выходу из цикла. В целом, мне нужно все действия (Scroll, Click, Move), но такое поведение я объяснить не могу. Возможно я где-то ошибаюсь, поэтому можете поправить.

А что в итоге?

Adam Geitgey, автор библиотеки face recognition, в своём репозитории имеет очень хороший набор примеров, которые многие используют при написании статей: https://github.com/ageitgey/face_recognition/tree/master/examples

Воспользуемся одним из них и получим следующий код, который можно скачать по ссылке: Activity.ipynb, Activity.py

# Подключаем нужные библиотекиimport cv2import face_recognition # Получаем данные с устройства (веб камера у меня всего одна, поэтому в аргументах 0)video_capture = cv2.VideoCapture(0) # Инициализируем переменныеface_locations = []from sound import SoundSound.volume_up() # увеличим громкость на 2 единицыcurrent = Sound.current_volume() # текущая громкость, если кому-то нужноvolum_half=50  # 50% громкостьvolum_full=100 # 100% громкостьSound.volume_max() # выставляем сразу по максимуму# Работа со временем# Подключаем модуль для работы со временемimport time# Подключаем потокиfrom threading import Threadimport threading# Функция для работы с активностью мышиfrom pynput import mousedef func_mouse():        with mouse.Events() as events:            for event in events:                if event == mouse.Events.Scroll or mouse.Events.Click:                    #print('Переместил мышку/нажал кнопку/скролл колесиком: {}\n'.format(event))                    print('Делаю половину громкости: ', time.ctime())                    Sound.volume_set(volum_half)                    break# Делаем отдельную функцию с напоминаниемdef not_find():    #print("Cкрипт на 15 секунд начинается ", time.ctime())    print('Делаю 100% громкости: ', time.ctime())    #Sound.volume_set(volum_full)    Sound.volume_max()        # Секунды на выполнение    #local_time = 15    # Ждём нужное количество секунд, цикл в это время ничего не делает    #time.sleep(local_time)        # Вызываю функцию поиска действий по мышке    func_mouse()    #print("Cкрипт на 15 сек прошел")# А тут уже основная часть кодаwhile True:    ret, frame = video_capture.read()        '''    # Resize frame of video to 1/2 size for faster face recognition processing    small_frame = cv2.resize(frame, (0, 0), fx=0.50, fy=0.50)    rgb_frame = small_frame[:, :, ::-1]    '''    rgb_frame = frame[:, :, ::-1]        face_locations = face_recognition.face_locations(rgb_frame)        number_of_face = len(face_locations)        '''    #print("Я нашел {} лицо(лица) в данном окне".format(number_of_face))    #print("Я нашел {} лицо(лица) в данном окне".format(len(face_locations)))    '''        if number_of_face < 1:        print("Я не нашел лицо/лица в данном окне, начинаю работу:", time.ctime())        '''        th = Thread(target=not_find, args=()) # Создаём новый поток        th.start() # И запускаем его        # Пока работает поток, выведем на экран через 10 секунд, что основной цикл в работе        '''        #time.sleep(5)        print("Поток мыши заработал в основном цикле: ", time.ctime())                #thread = threading.Timer(60, not_find)        #thread.start()                not_find()        '''        thread = threading.Timer(60, func_mouse)        thread.start()        print("Поток мыши заработал.\n")        # Пока работает поток, выведем на экран через 10 секунд, что основной цикл в работе        '''        #time.sleep(10)        print("Пока поток работает, основной цикл поиска лица в работе.\n")    else:        #все хорошо, за ПК кто-то есть        print("Я нашел лицо/лица в данном окне в", time.ctime())        Sound.volume_set(volum_half)            for top, right, bottom, left in face_locations:        cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 255), 2)        cv2.imshow('Video', frame)        if cv2.waitKey(1) & 0xFF == ord('q'):        breakvideo_capture.release()cv2.destroyAllWindows()

Суть кода предельно проста: бегаем в цикле, как только появилось хотя бы одно лицо (а точнее координаты), то звук делаем 50%. Если не нашёл никого поблизости, то запускаем цикл с мышкой.

Ожидание и реальность

Если вы посмотрели видео, то поняли, что результат ещё далёк от реальной эксплуатации.

Признаю честно - до этого момента никогда не сталкивался с многопоточностью на Python, поэтому "с наскоку" тему взять не удалось и результат по видео понятен. Есть неплохая статья на Хабре, описывающая различные методы многопоточности, применяемые в языке. Пока у меня решения нету по этой теме нету - будет повод разобраться лучше и дописать код/статью с учетом этого.

Так же возникает закономерный вопрос - а если вместо живого человека поставить перед монитором картинку? Да, она распознает, что, скорее всего, не совсем верно. Мне попался очень хороший материал по поводу определения живого лица в реальном времени - https://www.machinelearningmastery.ru/real-time-face-liveness-detection-with-python-keras-and-opencv-c35dc70dafd3/ , но это уже немного другой уровень и думаю новичкам это будет посложнее. Но эксперименты с нейронными сетями я чуть позже повторю, чтобы тоже проверить верность и повторяемость данного руководства.

Немаловажным фактором на качество распознавания оказывает получаемое изображение с веб-камеры. Предложение использовать 1/4 изображения (сжатие его) приводит только к ухудшению - моё лицо алгоритм распознать так и не смог. Для повышения качества предлагают использовать MTCNN face detector (пример использования), либо что-нибудь посложнее из абзаца выше.

Другая интересная особенность - таймеры в Питоне. Я, опять же, признаю, что ни разу до этого не было нужды в них, но все статьях сводится к тому, чтобы ставить поток в sleep(кол-во секунд). А если мне нужно сделать так, чтобы основной поток был в работе, а по истечению n-ое количества секунд не было активности, то выполнялась моя функция? Использовать демонов (daemon)? Так это не совсем то, что нужно. Писать отдельную программу, которая взаимодействует с другой? Возможно, но единство программы пропадает.

Заключение

На своём примере могу точно сказать - не все руководства одинаково полезны и простая задача может перерасти в сложную. Я понимаю, что большинство материалов является переводом/простым копированием документации. Но если ты пишешь руководство, то проверь хотя бы на тестовой системе, что написанные тобой действия точно повторимы. Пробуйте, экспериментируйте - это хороший повод изучить и узнать новое.

P.S. Предлагаю вам, читатели, обсудить в комментариях статью - ваши идеи, замечания, уточнения.

Предисловие

Я при работе с микроконтроллерами часто сталкивался с бинарными протоколами. Особенно, когда есть несколько контроллеров. Или же используется bluetooth low energy и необходимо написать код для обработки бинарных данных в характеристике. Помимо кода всегда требуется понятная документация.

Всегда возникает вопрос - а можно ли описать как-то протокол и сгенерировать на все платформы код и документацию? В этом может помочь IDL.

1. Что такое IDL

Определение IDL довольно простое и уже представлено на wikipedia

IDL, илиязык описания интерфейсов(англ.Interface Description LanguageилиInterface Definition Language)язык спецификацийдля описанияинтерфейсов, синтаксически похожий на описание классов в языкеC++.

Самое главное в IDL - он должен хорошо описывать интерфейс взаимодействия, API, протокол. Он должен быть достаточно понятен, чтобы служить другим инженерам документацией.

Бонус также является - генерация документации, структур, кода.

2. Мотивация

В процессе работы я попробовал разные кодогенераторы и IDL. Среди тех, что попробовал были - QFace (http://personeltest.ru/aways/github.com/Pelagicore/qface), swagger (Это не IDL, а API development tool). Также существует коммерческое решение проблемы: https://www.protlr.com/.

Swagger больше подходит к REST API. Поэтому сразу был отметён. Однако его можно использовать если применяется cbor (бинарный аналог json с кучей крутых фич).

В QFace давно не было коммитов, хотелось некоторых "наворотов" для применения в embedded, возникли сложности при написании шаблона. Он не ищет символы сам, не умеет считать поля enum-ов.

Бесплатные решения было найти сложно, чтобы можно было комфортно использовать при разработке бинарных протоколов.

Поэтому я отказался от генераторов кода и IDL в пользу написания некоторых "автоматизаций" в коде, позволяющих проще писать адаптер протокола. Но протокол с коллегами продолжили описывать при помощи QFace. Решил в свободное время попробовать сделать что-то более или менее годное.

2.1 Обзор QFace

IDL, которая являлась источником вдохновения, имеет простой синтаксис:

module <module> <version>import <module> <version>interface <Identifier> {    <type> <identifier>    <type> <operation>(<parameter>*)    signal <signal>(<parameter>*)}struct <Identifier> {    <type> <identifier>;}enum <Identifier> {    <name> = <value>,}flag <Identifier> {    <name> = <value>,}

Для генерации используется jinja2. Пример:

{% for module in system.modules %}    {%- for interface in module.interfaces -%}    INTERFACE, {{module}}.{{interface}}    {% endfor -%}    {%- for struct in module.structs -%}    STRUCT , {{module}}.{{struct}}    {% endfor -%}    {%- for enum in module.enums -%}    ENUM   , {{module}}.{{enum}}    {% endfor -%}{% endfor %}

Концепция интересная. Можно было просто "подпилить" для комфорта "напильником", что конечно и сделал мой коллега. Но мне показалось интересным взять библиотеку sly и просто написать IDL с нужными фичами.

3. Обзор sly

Почему именно sly - библиотека очень проста для описания грамматики.

Сначала надо написать лексер. Он токенизирует код чтобы далее было проще обрабатывать парсером. Код из документации:

class CalcLexer(Lexer):    # Set of token names.   This is always required    tokens = { ID, NUMBER, PLUS, MINUS, TIMES,               DIVIDE, ASSIGN, LPAREN, RPAREN }    # String containing ignored characters between tokens    ignore = ' \t'    # Regular expression rules for tokens    ID      = r'[a-zA-Z_][a-zA-Z0-9_]*'    NUMBER  = r'\d+'    PLUS    = r'\+'    MINUS   = r'-'    TIMES   = r'\*'    DIVIDE  = r'/'    ASSIGN  = r'='    LPAREN  = r'\('    RPAREN  = r'\)'

Нужно наследовать класс Lexer, в переменную tokens - добавить свои использованные токены. Само определение токенов делается в теле класса - достаточно просто описать регулярное выражение, соответсвующее токену.

Парсер - делает работу по преобразованию набора токенов по определенным правилам. С помощью его и осуществляется основная работа. В случае компиляторов - преобразование в байт-код/объектный файл итд. Для интерпретаторов - можно сразу выполнять вычисления. При реализации кодогенератора - можно преобразовать в дерево классов.

Также парсер задается очень простым способом (пример из документации):

class CalcParser(Parser):    # Get the token list from the lexer (required)    tokens = CalcLexer.tokens    # Grammar rules and actions    @_('expr PLUS term')    def expr(self, p):        return p.expr + p.term    @_('expr MINUS term')    def expr(self, p):        return p.expr - p.term    @_('term')    def expr(self, p):        return p.term    @_('term TIMES factor')    def term(self, p):        return p.term * p.factor    @_('term DIVIDE factor')    def term(self, p):        return p.term / p.factor    @_('factor')    def term(self, p):        return p.factor    @_('NUMBER')    def factor(self, p):        return p.NUMBER    @_('LPAREN expr RPAREN')    def factor(self, p):        return p.expr

Каждый метод класса отвечает за парсинг конкретной конструкции. В декораторе @_ указывается правило, которое обрабатывается. Имя метода sly распознает как название правила.

В этом примере сразу происходят вычисления.

Подробнее можно прочитать в официальной документации: https://sly.readthedocs.io/en/latest/sly.html

4. Процесс создания

В самом начале программа получает yml файл с настройками. Затем при помощи sly преобразовывает код в древо классов. Далее выполняются вычисления и поиски объектов. После вычисления - передается в jinja2 шаблон и дерево символов.

Читать как был определен список токенов для лексера может быть скучно, поэтому перейдем сразу к парсеру.

Вначале определили, что модуль состоит из списка термов:

    @_('term term')    def term(self, p):        t0 = p.term0        t1 = p.term1        t0.extend(t1)        return t0

Затем определим, что терм состоит из определений структуры, энумератора или интерфейса разделенные символом ";"(SEPARATOR):

   @_('enum_def SEPARATOR')    def term(self, p):        return [p.enum_def]    @_('statement SEPARATOR')    def term(self, p):        return [p.statement]    @_('interface SEPARATOR')    def term(self, p):        return [p.interface]    @_('struct SEPARATOR')    def term(self, p):        return [p.struct]

Здесь терм сразу паковался в массив для удобства. Чтобы список термов (term term правило) работал уже сразу с листами и собрал в один лист.

Ниже представлен набор правил для описания структуры:

    @_('STRUCT NAME LBRACE struct_items RBRACE')    def struct(self, p):        return Struct(p.NAME, p.struct_items, lineno=p.lineno)    @_('decorator_item STRUCT NAME LBRACE struct_items RBRACE')    def struct(self, p):        return Struct(p.NAME, p.struct_items, lineno=p.lineno, tags=p.decorator_item)    @_('struct_items struct_items')    def struct_items(self, p):        si0 = p.struct_items0        si0.extend(p.struct_items1)        return si0    @_('type_def NAME SEPARATOR')    def struct_items(self, p):        return [StructField(p.type_def, p.NAME, lineno=p.lineno)]    @_('type_def NAME COLON NUMBER SEPARATOR')    def struct_items(self, p):        return [StructField(p.type_def, p.NAME, bitsize=p.NUMBER, lineno=p.lineno)]    @_('decorator_item type_def NAME SEPARATOR')    def struct_items(self, p):        return [StructField(p.type_def, p.NAME, lineno=p.lineno, tags=p.decorator_item)]    @_('decorator_item type_def NAME COLON NUMBER SEPARATOR')    def struct_items(self, p):        return [StructField(p.type_def, p.NAME, bitsize=p.NUMBER, lineno=p.lineno, tags=p.decorator_item)]

Если описать простым языком правила - структура (struct) содержит поля структур (struct_items). А поля структур могут определяться как:

• тип (type_def), имя (NAME), разделитель (SEPARATOR)

• тип (type_def), имя, двоеточие (COLON), число (NUMBER - для битфилда, означает количество бит), разделитель

• список декораторов (decorator_item), тип, имя, разделитель

• список декораторов, тип, имя, двоеточие (COLON), число (NUMBER - для битфилда), разделитель

Новшество относительно QFace (однако есть в protlr) - была введена возможность описывать специальные условные ссылки на структуры. Было решено назвать эту фичу - alias.

    @_('DECORATOR ALIAS NAME COLON expr struct SEPARATOR')    def term(self, p):        return [Alias(p.NAME, p.expr, p.struct), p.struct]

Это было сделано чтобы поддерживалась следующая конструкция:

enum Opcode {    Start =  0x00,    Stop = 0x01};@alias Payload: Opcode.Startstruct StartPayload {...};@alias Payload: Opcode.Stopstruct StopPayload {...};struct Message {    Opcode opcode: 8;    Payload<opcode> payload;};

Данная конструкция обозначает, что если opcode = Opcode.Start (0x00) - payload будет соответствовать структуре StartPayload. Если opcode = Opcode.Stop (0x01) - payload будет иметь структуру StopPayload. То есть создаем ссылку структуры с определенными условиями.

Следующее что было сделано - отказался от объявления модуля. Показалось это избыточным так как - имя файла уже содержит имя модуля, а версию писать бессмысленно так как есть git. Хороший протокол имеет прямую и обратную совместимость и в версии нуждаться не должен. Был выкинут тип flag так как есть enum, и добавил возможность описания битфилдов. Убрал возможность определения сигналов так как пока что низкоуровневого примера, демонстрирующего пользу, не было.

Была добавлена возможность python-подобных импортов. Чтобы можно было импортировать из другого модуля только конкретный символ. Это полезно для генерации документации.

Для вычислений был создан класс - Solvable. Его наследует каждый объект, которому есть что посчитать. Например, для SymbolType (тип поля класса или интерфейса). В данном классе этот метод ищет по ссылке тип, чтобы добавить его в поле reference. Чтобы в jinja можно было сразу на месте обратиться к полям enum или структуры. Класс Solvable должен искать во вложенных символах вычислимые и вызывать solve. Т.е. вычисления происходят рекурсивно.

Пример реализации метода solve для структуры:

    def solve(self, scopes: list):        scopes = scopes + [self]        for i in self.items:            if isinstance(i, Solvable):                i.solve(scopes=scopes)

Как видно, в методе solve есть аргумент - scopes. Этот аргумент отвечает за видимость символов. Пример использования:

struct SomeStruct {i32someNumber;@setter: someNumber;void setInteger(i32 integer);};

Как видно из примера - это позволяет производить поиск символа someNumber в области видимости структуры, вместо явного указания SomeStruct.someNumber.

Заключение

По сравнению с QFace мне удалось - упростить написание шаблона за счет поиска типов, вычисления перечислений. Также полезно иметь импорт символов и возможность условно ссылаться на разные структуры.

В папке examples/uart - находится пример генерации заголовков, кода и html документации. Пример иллюстрирует типичный uart протокол с применением новых фич. Подразумевается, что функции типа put_u32 итд - определит сам пользователь исходя из порядка байт и архитектуры MCU.

Ознакомиться подробнее с реализацией можно по ссылке: https://gitlab.com/volodyaleo/volk-idl

P.S.

Это моя первая статья на Хабр. Буду рад получить отзывы - интересна ли данная тематика или нет. Если у кого-то есть хорошие примеры кодо+доко-генераторов бинарных протоколов для Embedded, было бы интересно прочитать в комментариях. Или какая-то успешная практика внедрения похожих систем для описания бинарных протоколов.

В данном проекте я не обращал особого внимания на скорость работы. Некоторые вещи делал чтобы "быстрее решить задачу". Было важнее получить рабочий код, который можно уже пробовать применять к разным проектам.

Модуль logging в питоне - это мощный инструмент в разработки. Он помогает отследить ошибки, наблюдать за работой приложения и даже собирать статистику об использовании вашего сервиса. В этой статье я расскажу, как можно расширить возможности этого модуля и причем тут телеграмм.

Intro

Давным-давно, а точнее несколько месяцев назад, накануне Нового года, я сидел дома и решал задачу по машинному обучению. Связана она была с нейронными сетями и классификацией текстов, поэтому я естественно пользовался бесплатным GPU от гугла (colab). За окном шел снег, а модели обучались ну уж очень долго. Обучать модель оставалось всего несколько минут, как вдруг появляется уведомление, что подключение к runtime потеряно, а это значит, что обученную модель и сабмиты из этого runtime скачать я не смогу, и все придется начинать заново.

Данная ситуация меня расстроила и заставила задуматься от том, что надо как-то через код сохранять файлы, потому что процесс выполнения кода не прекращается, когда возникает такая ошибка. Почему-то я сразу подумал про телеграм ботов и, написав несколько функций, смог сохранять самбиты и промежуточную информацию о том, как обучается модель и сколько ей осталось.

Копировал я этот код из ноутбука (jupyter notebook) в ноутбук, а потом осознал, что это можно встроить в модуль logging и завернуть в библиотеку, чтобы не таскать каждый раз большие куски кода, а использовать всего пару строк.

Logging.handlers

Понятно было, что за основу надо взять модуль для логирования, встроенный в питон. Очевидно, потому что изобретать свой велосипед мне не хотелось, а еще потому что, для использования в приложениях пришлось бы переписывать весь код логирования. Надо было найти способ перехватывать логи, и пересылать их в телеграмм. К счастью, в logging есть такой класс, как StreamHandler. Не буду вдаваться в подробности, но с помощью него можно сделать свою функцию, которая будет запускаться для каждой новой строки в логах и отправлять ее в телеграмм.

Tg-logger

И вот, держа в голове все вышеописанные идеи, я отправился писать свою библиотеку. Вот, что получилось.

Для тех, кому лень запускать код, но хочется понять, как это будет работать, я сделал бота @tg_logger_demo_bot.

Чтобы воспользоваться библиотекой нужно:

• создать телеграмм бота (как это сделать описано здесь)

• получить свой user_id (это можно сделать через @tg_logger_demo_bot с помощью команды /id)

Установим библиотеку через pip.

pip install tg-logger

Рассмотрим код примера

import loggingimport tg_logger# Telegram datatoken = "1234567890:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"users = [1111111111]# Base loggerlogger = logging.getLogger('foo')logger.setLevel(logging.INFO)# Logging bridge setuptg_logger.setup(logger, token=token, users=users)# Testlogger.info("Hello from tg_logger by otter18")

Особо интересна для нас строка, в которой подключается логирование в телеграмм.

# Logging bridge setuptg_logger.setup(logger, token=token, users=users)

В функцию setup() нужно просто передать тот logger, к которому вы хотите подключить мост. Если заглянуть в документацию, то можно посмотреть на другие параметры функции setup(). С помощью них можно, в частности, настроить формат, в котором логи будут отправлены.

Outro

• Весь исходный код есть на гитхабе: github.com/otter18/tg_logger.

• Документация на английском есть на Read the Docs.

• Библиотека доступна для скачивания через pip: pypi.org/project/tg-logger/.

Введение

После того, как мы закончили разработку веб-приложения, оно должно быть размещено на хосте, чтобы общественность могла получить доступ к нему из любого места. Мы посмотрим, как развернуть и разместить приложение на экземпляре AWS EC2, используя Nginx в качестве веб-сервера и Gunicorn в качестве WSGI.

AWS EC2

Amazon Elastic Compute Cloud (Amazon EC2) - это веб-сервис, обеспечивающий масштабируемость вычислительных мощностей в облаке. Мы устанавливаем и размещаем наши веб-приложения на экземпляре EC2 после выбора AMI (OS) по нашему усмотрению. Подробнее об этом мы поговорим в следующих разделах.

NGINX

Nginx - это веб-сервер с открытым исходным кодом. Мы будем использовать Nginx для сервера наших веб-страниц по мере необходимости.

GUNICORN

Gunicorn - это серверная реализация интерфейса шлюза Web Server Gateway Interface (WSGI), который обычно используется для запуска веб-приложений Python.

WSGI - используется для переадресации запроса с веб-сервера на Python бэкэнд.

Мы не будем использовать сервер, который поставляется с django по умолчанию в производстве.

Развертывание приложения

Мы запустим EC2 экземпляр на AWS, для этого войдите в консоль aws.

• Выберите EC2 из всех сервисов

• Выберите запуск New instance и выберите Ubuntu из списка.

• Выберите любой из экземпляров, каждый из них имеет различные конфигурации, мы выберем тот, который имеет свободный уровень.

• Теперь настройте группы безопасности и откройте порты 8000 и 9000, так как мы будем использовать эти порты . Просмотрите и запустите ваш экземпляр, может потребоваться некоторое время, чтобы он запустился.

Подключение к Экземпляру

Мы можем подключиться к экземпляру, используя опцию 'connect' в консоли (или с помощью putty или любого другого подобного инструмента ). После подключения запустите следующие команды

sudo apt-get update

Установите python , pip и django

sudo apt install pythonsudo apt install python3-pippip3 install django

Теперь, когда мы установили наши зависимости, мы можем создать папку, в которую мы скопируем наше приложение django.

cd  /home/ubuntu/  mkdir Projectcd Projectmkdir ProjectNamecd ProjectName

Теперь мы поместим наш код по следующему пути.
/home/ubuntu/Project/ProjectName

GitHub

Убедитесь, что ваш код находится в репозитории, чтобы мы могли легко втянуть его в наш экземпляр ec2.

• Перейдите в только что созданную папку (/home/ubuntu/Project/ProjectName/)

• git clone <repository-url>

Это клонирует репозиторий в папку, и в следующий раз мы сможем просто вытащить изменения с помощью git pull.

Settings.py Файл.

Мы должны внести некоторые изменения в settings.py в нашем проекте.

• Вставьте свои секретные ключи и пароли в переменные окружения

• Установить Debug = False

• Добавте Ваш домейн в ALLOWED_HOSTS

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))STATIC_ROOT = os.path.join(BASE_DIR, static)

Выполните следующие действия, чтобы миграция модели произошла и все статические файлы были собраны в общую папку (путь указан в STATIC_ROOT).

manage.py makemigrationsmanage.py migratemanage.py collectstatic

Установка Nginx

Для установки Nginx выполните команду

 sudo apt install nginx

Есть конфигурационный файл с именем по умолчанию в /etc/nginx/sites-enabled/, который имеет базовую настройку для NGINX, мы отредактируем этот файл.

sudo vi default

Файл будет выглядеть так после добавления необходимой конфигурации, а остальная часть файла останется такой же.

мы добавим proxy_pass http://0.0.0.0:9000 и укажем путь к нашей статической папке, добавив путь внутри каталога /static/, как указано выше. Убедитесь, что вы собрали все статические файлы в общую папку, запустив команду

manage.py collectstatic

Теперь запустите сервер nginx

sudo service nginx start             #to start nginxsudo service nginx stop              #to stop nginxsudo service nginx restart           #to restart nginx

Установка Gunicorn

pip install gunicorn

Убедитесь, что Вы находитесь в папке проекта, например: /home/ubuntu/Project, и запустите следующую команду, чтобы запустить gunicorn

gunicorn ProjectName.wsgi:application- -bind 0.0.0.0:9000

Теперь, когда мы установили и настроили nginx и gunicorn, к нашему приложению можно получить доступ через DNS экземпляра ec2.

Привет, Хабр!

Сегодня хочу поделиться своим небольшим домашним проектом:

ModulationPy (GiHub)

- модуль для моделирования цифровых схем модуляции (это которые PSK, QAM и т.п.). Проект был вдохновлен другой питоновской библиотекой: CommPy; однако, в рассмотренном классе задач с ней удалось даже немного посоревноваться!

На данный момент доступны два класса схем модуляции:

• M-PSK: Phase Shift Keying (фазовая цифровая модуляция)

• M-QAM: Quadratured Amplitude Modulation (квадратурная амплитудная модуляция)

  где M - это порядок модуляции.

Интересен модуль может быть, скорее всего, в разрезе образовательных целей в сфере беспроводной связи (подбор модуляций исходил именно из нее), однако, вдруг кому-то пригодится и для научных изысканий. Не MatLab'ом насущным едины!

Примеры использования

Итак, для начала скачиваем библиотеку с PyPI:

$ pip install ModulationPy

Либо устанавливаем из исходников, но на PyPI на момент написания статьи все-таки актуальная версия.

Зависимости две:

• numpy>=1.7.1

• matplotlib>=2.2.2 (для построения сигнальных созвездий)

Продемонстрируем на деле.

Итак, например, мы хотим использовать QPSK с поворотом фазы pi/4, двоичным входом и наложением по Грею (Gray mapping). Импортируем и инициализируем:

import numpy as npfrom ModulationPy import PSKModemmodem = PSKModem(4, np.pi/4,                 gray_map=True,                 bin_input=True)

Проверяем то ли мы инициализировали, нарисовав сигнальное созвездие методом plot_const():

modem.plot_const()

То же самое сделаем для 16-QAM (но для десятичных чисел на входе; указывать фазовый сдвиг не нужно - подразумевается наиболее распространенная прямоугольная QAM):

from ModulationPy import QAMModemmodem = QAMModem(16,                 gray_map=True,                  bin_input=False)modem.plot_const()

На данный момент модуляция QAM реализована по примеру функции qammod в Octave [4]. И, да, реализованы только четные (в том смысле, что результат log2(M) - четное число) схемы модуляции (4-QAM, 16-QAM, 64-QAM). Пусть и не совсем полный набор, но как бы то ни было, в популярных стандартах беспроводной связи все равно нет "нечетных" схем модуляции (насколько я знаю).

Далее предлагаю перейти, собственно, к главному в модемах: к модуляции и демодуляции. Для этого нам понадобятся два метода:modulate() и demodulate() , доступные в обоих классах.

Метод modulate() принимает на вход всего один аргумент:

• вектор входных значений (1-D ndarray of ints) - либо единиц и нулей, если выбрана опция bin_input=True , либо целых десятичных чисел от 0 до M-1, если bin_input=False ;

Методdemodulate() ожидает максимум два аргумента:

• вектор, который должен быть демодулирован (1-D ndarray of complex symbols) ;

• значение дисперсии аддитивного шума (float, по умолчанию 1.0).

Например, вот как это будет выглядеть для QPSK (двоичный вход/выход):

import numpy as npfrom ModulationPy import PSKModemmodem = PSKModem(4, np.pi/4,                  bin_input=True,                 soft_decision=False,                 bin_output=True)msg = np.array([0, 0, 0, 1, 1, 0, 1, 1]) # input messagemodulated = modem.modulate(msg) # modulationdemodulated = modem.demodulate(modulated) # demodulationprint("Modulated message:\n"+str(modulated))print("Demodulated message:\n"+str(demodulated)) >>>  Modulated message:   [0.70710678+0.70710678j  0.70710678-0.70710678j    -0.70710678+0.70710678j  -0.70710678-0.70710678j]>>> Demodulated message:   [0. 0. 0. 1. 1. 0. 1. 1.]

Или тоже QPSK, но уже с недвоичным входом / выходом:

import numpy as npfrom ModulationPy import PSKModemmodem = PSKModem(4, np.pi/4,                  bin_input=False,                 soft_decision=False,                 bin_output=False)msg = np.array([0, 1, 2, 3]) # input messagemodulated = modem.modulate(msg) # modulationdemodulated = modem.demodulate(modulated) # demodulationprint("Modulated message:\n"+str(modulated))print("Demodulated message:\n"+str(demodulated))>>> Modulated message:[ 0.70710678+0.70710678j -0.70710678+0.70710678j  0.70710678-0.70710678j -0.70710678-0.70710678j] >>> Demodulated message:[0, 1, 2, 3]

Пример для 16-QAM (десятичный вход / выход):

import numpy as npfrom ModulationPy import QAMModemmodem = PSKModem(16,                  bin_input=False,                 soft_decision=False,                 bin_output=False)msg = np.array([i for i in range(16)]) # input messagemodulated = modem.modulate(msg) # modulationdemodulated = modem.demodulate(modulated) # demodulationprint("Demodulated message:\n"+str(demodulated))>>> Demodulated message:[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]

В общем и целом, я думаю, понятно. Доступные опции старался делать по примеру доступных в рамках матлабовского Communication Toolbox. Подробное описание приведено в README.md проекта.

BER performance

Продемонстрируем адекватно ли модемы работают в случае присутствия шума (возьмем классический АБГШ, он же AWGN), используя простейшую модель приема-передачи:

Сверяться будем с теоретическими кривыми [5] (если кому интересно, все формулы описаны тоже в README.md).

Исходные коды для моделирования представлены по ссылкам: M-PSK, M-QAM.

Результаты:

Да, с огрехами на малых значениях битовых ошибок (из-за относительно небольшого количества усреднений, я полагаю), однако.... it works!

Производительность

А теперь я хотел бы вернуться к вопросу "соревнования" с упомянутой выше CommPy.

Что нас отличает:

• организация кода, стилистические различия (побочное, но не пренебрежимое);

• я использовал более быстрый алгоритм демодуляции [6] (подробно описан в матлабовской документации [7], ну, и я добавил все в тот же README.md).

И вот, что получилось "забенчмаркать":

• Скрипт: CommPy_vs_ModulationPy.ipynb

• Платформа: https://jupyter.org/try (Classic Notebook)

• Длина фрейма: 10 000 (элементов)

Результаты:

Разработчикам CommPy результаты, вроде, понравились (см. данный issue) - поэтому возможно в обозримом будущем что-то из моего ModulationPy будет перекочевывать в CommPy (я не против, главное, чтобы пользу приносило). Но это, как говорится, поживем - увидим.

И, да, пусть результаты производительности и не дотянули до MatLab (по крайней мере исходя из данного примера: см. вкладку "Examples"), я все равно считаю достигнутое неплохим стартом!

Послесловие

Наверное, проекту не хватает еще некоторых видов модуляции (тех же 32-QAM и 128-QAM или же используемой в DVB-S2/S2X APSK), однако, честно скажу, что не могу обещать их скорого добавления.

Проект всегда был для меня в большей мере площадкой для изучения языка Python и библиотеки NumPy на практике (и сопутствующих инструментов: юнит-тесты (не успел правда в данном случае перейти на pytest - каюсь), CI (использую Travis), подготовка модуля для PyPi и т.д.), однако, теперь, слава богу, всему этому есть приложение и в рамках рабочих задач!

Однако, все же буду рад вашим issue и pull request'ам! И если возьметесь интегрировать наработки в CommPy, тоже будет очень круто!

В общем, не серчайте, если вдруг не отвечу достаточно быстро, и да пребудет с вами сила науки!

Литература и ссылки

1. Haykin S. Communication systems. John Wiley & Sons, 2008. p. 93

2. Goldsmith A. Wireless communications. Cambridge university press, 2005. p. 88-92

3. MathWorks: comm.PSKModulator (https://www.mathworks.com/help/comm/ref/comm.pskmodulator-system-object.html?s_tid=doc_ta)

4. Octave: qammod (https://octave.sourceforge.io/communications/function/qammod.html)

5. Link Budget Analysis: Digital Modulation, Part 3 (www.AtlantaRF.com)

6. Viterbi, A. J. (1998). An intuitive justification and a simplified implementation of the MAP decoder for convolutional codes. IEEE Journal on Selected Areas in Communications, 16(2), 260-264.

7. MathWorks: Approximate LLR Algorithm (https://www.mathworks.com/help/comm/ug/digital-modulation.html#brc6ymu)

Квалификацию надо иногда повышать, и вообще учиться для мозгов полезно. А потому пошел я недавно на курсы - поизучать Python и всякие его фреймворки. На днях вот до Django добрался. И тут мы в ходе обучения коллективно выловили не то чтобы баг, но дивный эффект на стыке Python 3, Sqlite 3, JSON и Win 10. Причем эффект был настолько дивен, что гугль нам не помог - пришлось собираться всей заинтересованной группой вместе с преподавателем и коллективным разумом его решать.
А дело вот в чем: изучали мы базу данных (а у Django предустановлена Sqlite 3) и, чтоб каждый раз заново руками данные не вбивать, прикрутили загрузку скриптом из json-файлов. А в файлы данные из базы штатно дампили питоновскими же методами:

python manage.py dumpdata -e contenttypes -o db.json

Внезапно те, кто работал под виндой (за все версии не поручусь, у нас подобрались только обитатели Win 10), обнаружили, что дамп у них производится в кодировке windows-1251. Более того, джейсоны в этой кодировке отлично скармливаются базе. Но стоило только переформатировать их в штатную по документам для Sqlite 3, Python 3 и особенно для JSON кодировку UTF-8, как в лучшем случае кириллица в базе превращалась в тыкву, а в худшем ломался вообще весь процесс загрузки данных.
Ничего подобного найти не удалось ни в документации, ни во всем остальном гугле, считая и англоязычный. Что самое загадочное, ручная загрузка тех же самых данных через консоль или админку проекта работала как часы, хотя уж там-то кодировка была точно UTF-8. Более того, принудительное прописывание кодировки базе никакого эффекта не дало.
Мы предположили, что причиной эффекта было взаимодействие джейсона с операционной системой - каким-то образом при записи и чтении именно джейсонов система навязывала свою родную кодировку вместо нормальной. И действительно, когда при открытии файла принудительно устанавливалась кодировка UTF-8:

open(os.path.join(JSON_PATH, file_name + '.json'), 'r', encoding="utf-8")

в базу попадали не кракозябры, а нормальные русские буквы. Но проблему с созданием дампа таким способом не решишь, а переделывать кодировку потом руками тоже как-то не по-нашему. И тогда мы решили поискать способ укротить винду.
И такой способ нашелся. Вот он:

• открываем панель управления, но не новую красивую, а старую добрую:

• открываем (по стрелке) окошко региона:

• по стрелкам переключаем вкладку "Дополнительно" и открываем окошко "Изменить язык системы":

• и в нем ставим галку по стрелке в чекбоксе "Бета-версия: Использовать Юникод (UTF-8) для поддержки языка во всем мире.

Система потребует перезагрузки, после чего проблема будет решена.

Не могу сказать, чтобы этот мелкий странный баг был так уж важен или интересен (питоновские проекты обычно живут под линуксами, где такого не бывает), но мозги он нам поломал изрядно - вследствие чего я и решил написать эту заметку. Мало ли кто еще из новичков как раз во время учебы попадется.

Незаметно от всех 12 мая 2021 вышла новая версия известного микрофреймворка Flask. Хотя казалось, что во Flask есть уже все, ну или почти все, что нужно для микрофреймворка.
Предвкушая интерес, а что же нового завезли, оставлю ссылку на Change log.

Из приглянувшихся особенностей новой версии:

• Прекращена поддержка Python версии 2. Минимальная версия Python 3.6

• Поддержка асинхронных view и других обратных вызовов, таких как обработчики ошибок, определенные с помощью async def. Обычные синхронные view продолжают работать без изменений. Функции ASGI, такие как веб-сокеты, не поддерживаются.

• Добавьте декораторы роутов для общих методов HTTP API.
  @app.post ("/ login") == @ app.route ("/ login", methods = ["POST"])

• Новая функция Config.from_file для загрузки конфигурации из файла любого формата.

• Команда flask shell включает завершение табуляции, как это делает обычная оболочка python.

• При обслуживании статических файлов браузеры будут кэшировать на основе содержимого, а не на основе 12-часового таймера. Это означает, что изменения статического содержимого, такого как стили CSS, будут немедленно отражены при перезагрузке без необходимости очистки кеша.

Рассмотрим асинхронность

Все бы было хорошо, но в самом начале после установки был не найден модуль asgiref. Доустановим руками.

Для примера напишем самое простое приложение: Ping/Pong. Оно не имеет особого смысла и сложной логики, только имитирует некоторую проверку "жив ли сервис". Также это приложение станет бенчмарком.

from flask import Flaskapp = Flask(__name__)@app.get('/')async def ping():    return {'message': 'pong'}if __name__ == '__main__':    app.run(host='0.0.0.0')

Деплой

Как было сказано в Change log: "Функции ASGI, такие как веб-сокеты, не поддерживаются."
То есть только единственный способ задеплоить приложение используя gunicorn.

Команда: gunicorn -w 8 --bind 0.0.0.0:5000 app:app
-w 8 - 8 запущенных процессов
--bind 0.0.0.0:5000 - адрес приложения

Сверим производительность

Команда для нагрузочного тестирования: wrk -t 8 -c 100 -d 5 http://localhost:5000

Асинхронное приложение Flask 2.0:
Running 5s test @ http://localhost:5000
8 threads and 100 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 17.80ms 2.37ms 36.95ms 91.92%
Req/Sec 673.44 163.80 3.86k 99.75%
26891 requests in 5.10s, 4.21MB read
Requests/sec: 5273.84
Transfer/sec: 844.69KB

Синхронное приложение Flask 2.0:
Running 5s test @ http://localhost:5000
8 threads and 100 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 4.91ms 842.62us 21.56ms 89.86%
Req/Sec 2.38k 410.20 7.64k 93.53%
95301 requests in 5.10s, 14.91MB read
Requests/sec: 18689.25
Transfer/sec: 2.92MB

Синхронное приложение Flask 1.1.2:
Running 5s test @ http://localhost:5000
8 threads and 100 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 4.98ms 823.42us 17.40ms 88.55%
Req/Sec 2.37k 505.28 12.23k 98.50%
94522 requests in 5.10s, 14.78MB read
Requests/sec: 18537.84
Transfer/sec: 2.90MB

В качестве вывода

Исходя из результатов бенчмарков можно увидеть, что 1 и 2 версия в синхронном режиме выдают одинаковые результаты(с небольшой погрешностью). Что касается асинхронности в Flask 2.0 можно сделать вывод, что пока она слишком сырая даже в dev режиме запуска асинхронный view отстает от синхронного. Но также не стоит забывать о том что ASGI пока не поддерживается, и нет возможности запустить через uvicorn. Остается только ждать обновления и следить за дальнейшим развитием.

Обновилась именно major версия, а это значит у нас есть надежда новую переосмысленную итерацию фреймворка. Разработчиков стоит похвалить как минимум за то, что проект не заброшен и старается успевать за основными тенденциями в других фреймворках. Лично мне очень нравится Flask, он совершенно не перегружен, как например Django.

Если у вас есть идеи, почему получились именно такие результаты - пожалуйста поделитесь ими в комментариях.

Введение

Привет, меня зовут Артём и я работаю бекендером в KTS. Компания уже 3 года проводит летние и зимние курсы по разработке, а в феврале этого года прошла очередная бесплатная backend-школа от KTS. В ее рамках студенты изучали инструменты и технологии, которые используют разработчики нашей компании, общались с менторами и делали итоговый проект - чат-бота в стиле Моя игра, который защищали в конце курса. После курса отличившихся студентов мы приглашали на стажировку.

Школа состояла из 6 лекций, шаг за шагом погружавших студентов в мир веб-разработки. На них были рассмотрены такие темы как сетевые протоколы, взаимодействие backend-а и frontend-а, компоненты веб-сервера и многое другое. Лейтмотивом курса было изучение асинхронного веб-программирования на Python, в частности изучение фреймворка aiohttp.

Для поступления на курс нужно было пройти комплексный тест на знания в области веба и python-а, так что студенты пришли учиться с хорошим начальным уровнем знаний. Однако, во время курса выяснилось, что не все темы даются одинаково легко. Самыми трудными для понимания темами стали:

1. Асинхронное программирование

2. Работа с СУБД

3. Деплой приложения

Студенты задавали достаточно разные по уровню понимания вопросы, начиная от Как создать отложенную задачу, используя только asyncio? и заканчивая Почему нельзя использовать Django для асинхронного программирования? (имелась в виду полностью синхронная версия Django). В коде наши менторы тоже находили ошибки, связанные с недостаточным пониманием предмета, например, использование синхронного драйвера для базы данных в асинхронном проекте.

По результатам курса, я решил написать небольшой туториал, рассказывающий о создании базового aiohttp-сервиса с нуля и затрагивающий самые сложные для студентов вопросы: как сделать асинхронное python-приложение, как работать с базой данных и как разложить свой проект в интернете.

В цикле статей мы рассмотрим следующие темы:

1. Архитектура веб-приложения

2. Асинхронная работа с базой данных и автоматические миграции

3. Работа с HTML-шаблонами с помощью Jinja2

4. Размещение нашего приложения в Интернете с помощью сервиса Heroku

5. А также сигналы, обработку ошибок, работу с Dockerом и многое другое.

Эта статья первая из трех, и ее цель помочь начинающим aiohttp-программистам написать первое hello-world приложение.

В этой статье мы напишем небольшое веб-приложение на aiohttpстену с отзывами, где пользователь может оставить мнение о продукте.

Мы пройдем по шагам:

• Создание проекта

• Структура проекта

• Создание View

• Создание Route

• Создание шаблона

• Запуск приложения
  

Создание проекта

Все команды в статье были выполнены в операционной системе OSX, но также должны работать в любой *NIX системе, например в Linux Ubuntu. Во время разработки я буду использовать Python 3.7.

Давайте создадим папку aiohttp_server, которая в дальнейшем будет называться корнем проекта. В ней создадим текстовый файл requirements.txt, который будет содержать все необходимые для работы приложения зависимости и их версии. Запишем в него следующие модули:

aiohttp==3.7.3 # наш фрейворкaiohttp-jinja2==1.4.2 # модуль для работы с HTML-шаблонами

Создадим виртуальное окружение что-то вроде песочницы, которое содержит приложение со своими библиотеками, обновление и изменение которых не затронет другие приложение, и установим в него наши зависимости:

cd {путь_до_папки}/aiohttp_serverpython3 -m venv venvsource venv/bin/activate

После этого в начале строки терминала должна появится надпись (venv)это означает что виртуальное окружение успешно активировано. Установим необходимые модули:

pip install -r requirements.txt

Структура проекта

Создадим в папке aiohttp_server следующую структуру:

 app    __init__.py    forum       __init__.py       routes.py  # тут будут пути, по которым надо отправлять запросы       views.py  # тут будут функции, обрабатывающие запросы    settings.py main.py  # тут будет точка входа в приложение requirements.txt templates    index.html  # тут будет html-шаблон страницым сайта

Теперь откроем файл main.py и добавим в него следующее:

from aiohttp import web  # основной модуль aiohttpimport jinja2  # шаблонизатор jinja2import aiohttp_jinja2  # адаптация jinja2 к aiohttp# в этой функции производится настройка url-путей для всего приложенияdef setup_routes(application):   from app.forum.routes import setup_routes as setup_forum_routes   setup_forum_routes(application)  # настраиваем url-пути приложения forumdef setup_external_libraries(application: web.Application) -> None:   # указываем шаблонизатору, что html-шаблоны надо искать в папке templates   aiohttp_jinja2.setup(application, loader=jinja2.FileSystemLoader("templates"))def setup_app(application):   # настройка всего приложения состоит из:   setup_external_libraries(application)  # настройки внешних библиотек, например шаблонизатора   setup_routes(application)  # настройки роутера приложенияapp = web.Application()  # создаем наш веб-серверif __name__ == "__main__":  # эта строчка указывает, что данный файл можно запустить как скрипт   setup_app(app)  # настраиваем приложение   web.run_app(app)  # запускаем приложение

После предварительной настройки можно создать первый View.

Первый View

Viewэто некий вызываемый объект, который принимает на вход HTTP-запросRequest и возвращает на пришедший запрос HTTP-ответResponse.

Http-запрос содержит полезную информацию, например url запроса и его контекст, переданные пользователем данные и многое другое. В контексте запроса содержатся данные, которые мы или aiohttp добавили к этому запросу. Например, мы предварительно авторизовали пользователячтобы повторно не проверять авторизацию пользователя из базы во всех View и не дублировать код, мы можем добавить объект пользователя в контекст запроса. Тогда мы сможем получить нашего пользователя во View, например, так: request['user'].

HTTP-ответ включает в себя полезную нагрузку, например, данные в json, заголовки и статус ответа. В простейшем View, который из примера выше, всю работу по формированию HTTP-ответа выполняет декоратор @aiohttp_jinja2.template("index.html") . Декоратор получает данные из View, которые возвращаются в виде словаря, находит шаблон index.html (о шаблонах написано ниже), подставляет туда данные из этого словаря, преобразует шаблон в html-текст и передает его в ответ на запрос. Браузер парсит html и показывает страницу с нашим контентом.

В файле views.py в папке app/forum напишем следующий код:

import aiohttp_jinja2from aiohttp import web# создаем функцию, которая будет отдавать html-файл@aiohttp_jinja2.template("index.html")async def index(request):   return {'title': 'Пишем первое приложение на aiohttp'}

Здесь создается функциональный View (function-based View). Определение функциональный означает, что код оформлен в виде функции, а не классом (в следующей части мы коснемся и class-based View).

Рассмотрим написанную функцию детальнее: функция обернута в декоратор @aiohttp_jinja2.template("index.html")этот декоратор передает возвращенное функцией значение в шаблонизатор Jinja2, а затем возвращает сгенерированную шаблонизатором html-страницу как http-ответ. В данном случае возвращенным значением будет словарь, значения которого подставляются в html-файл index.html.

Отдельно стоит заметить, что объект запроса request передается как аргумент функции index. Мы не используем request в этой функции, но будем использовать в дальнейшем.

HTTP-запрос отправляется на конкретный url-адрес. Для передачи HTTP-запроса в нужный View необходимо задать эту связь в приложении с помощью Route.

Первый Route

Routeэто звено, связывающее адрес, по которому был отправлен запрос и код View, в котором этот запрос будет обработан. То есть, если пользователь перейдет в корень нашего сайта (по адресу /), то объект запроса будет передан в View index и оттуда же будет возвращен ответ. Подробней про Route можно прочитать тут.

В файл routes.py необходимо добавить следующий код:

from app.forum import views# настраиваем пути, которые будут вести к нашей страницеdef setup_routes(app):   app.router.add_get("/", views.index)

Первый Template

Теперь нам осталось только добавить в templates/index.html код верстку нашей страницы. Его можно найти по этой ссылке.

Templateэто html-шаблон, в который подставляются данные, полученные в результате обработки запроса. В примере в коде View отдается словарь с ключом title, шаблонизатор Jinja2 ищет в указанном html-шаблоне строки {{title}} и заменяет их на значение из словаря по данному ключу. Это простейший пример, шаблоны позволяют делать намного больше: выполнять операции ветвления, циклы и другие операции, например, суммирование. Примеры использования можно посмотреть в документации jinja2.

Запуск приложения

Мы создали первую версию нашего приложения! Осталось запустить его следующей командой в терминале (убедитесь, что находитесь в папке aiohttp_server):

python3 main.py

Вы должны увидеть следующий текст в консоли. Он означает, что сервер запущен на порту 8080.

======== Running on http://0.0.0.0:8080 ========(Press CTRL+C to quit)

Давайте теперь посмотрим результаты нашей работы! Для этого перейдите по адресу http://0.0.0.0:8080 в браузере. Вы должны увидеть первую версию нашего приложения. При клике на кнопку Отправить должно возникнуть сообщение о том, что отзыв отправлен.

Поздравляю! Вы успешно создали первое приложение на aiohttp!

Заключение

В статье рассмотрено создание простого приложения на aiohttp, которое принимает запрос пользователя и отдает html-страницу. Мы затронули:

• Настройку виртуального окружения

• Базовую настройку проекта на aiohttp

• Создание View

• Создание Route

• Использование html-шаблонов

Наше приложение представляет собой простой веб-сервер, отдающий html-страницу по запросу - в нем нет никакого взаимодействия с базами данных, его структура максимально проста и оно недоступно пользователям в Интернете. В следующих статьях мы разберем, как вырастить из нашей заготовки настоящее веб-приложение на aiohttp и опубликовать его в Интернете.

Весь код статьи можно найти на гитхабе.

Пользуясь случаем, приглашаю всех читателей, интересующихся веб-разработкой, к нам на бесплатные занятия школу KTS. А для более опытных читателей сейчас идет запись на продвинутые курсы для backend-разработчиков, желающих повысить свои навыки в асинхронной веб-разработке. Всю информацию о всех школах можно найти на сайте, а также в нашем телеграм-чате.

Очень круто - запрограммировать механизм или программный модуль, заставив его выполнять твою волю. С похожими мыслями в конце 2018 я размышлял о том, что хочу сделать собственный WoW-сервер, который будет полностью мной управляем. Поизучав С++ исходники MANGOS, я пришел к выводу, что не смогу вот так взять и реализовать все свои идеи, не понимая, как же работает MMO RPG сервер от начала и до конца. И для этой цели я решил реализовать свой движок. С нуля.

Начну с того, что вначале из документации у меня были только исходники. Ответы же на многочисленные свои вопросы я с переменным успехом по крупицам искал на форумах (в этом плане - огромный респект сообществу MANGOS, - очень отзывчивый коллектив). Итого прошло несколько месяцев, прежде чем методом проб и ошибок я реализовал свой первый рабочий прототип Login сервера - и смог попасть на экран выбора персонажей.

Если вкратце, то Login Server (как и аутентификация в клиенте WoW) построены на использовании алгоритма SRP. Описание алгоритма выходит за пределы статьи, но если вкратце, то он позволяет идентифицировать пользователя без передачи пароля на сервер, благодаря чему пароль (даже в закэшированном виде) можно не хранить на сервере. Даже желательно.

Алгоритм шифрования же World сервера скорее всего отличается от клиента к клиенту (я сделал такой вывод при беглом изучении исходников сервера WoW 3.3.5a). В статье речь пойдет о сервере для WoW 2.4.3 (именно для этой версии я писал сервер). Там используется что-то вроде шифра Цезаря. Хотя чаще (в исходниках) можно встретить название HeaderCrypt или Wowcrypt.

В версии 2.4.3 шифруются первые несколько байт каждого пакета на World сервере (кроме первого пакета). Для расшифровки (decrypt) используются первые 6 байт: 2 байта - размер пакета + 4 байта - opcode (специальный код, по которому можно определить хэндлер(ы), что будет с этим пакетом работать). И для зашифровки (encrypt) используются первые 4 байта (только opcode). Соответственно, если перехватить пакет (sniff), то не получится определить, на какой опкод он будет отправлен. А если пакет большой, то он может быть разбит на несколько и чтобы их правильно достать из буфера - нужны эти самые первые 2 байта (размер пакета).

Процесс входа на сервер вкратце можно описать так:

Клиент последовательно обменивается данными с Login сервером и в случае успеха получает SRP токен (session key), который затем будет использован для создания токена шифрования/расшифровки пакетов (crypto key). Токен создается на этапе "send auth request" (см. схему выше). После чего на клиент отправляется auth response, суть которого - показать, что операция прошла успешно. Auth response - единственный незашифрованный пакет, последующие пакеты будут шифроваться с помощью crypto key. По поводу шифра - у меня он реализован так.

Процесс взаимодействия с World сервером простой - клиент отправляет зашифрованный пакет, на сервере он кладется в буфер, расшифровываются первые несколько байт (о чем я писал выше), получаем размер (size) и опкод (opcode) и если количество байт в буфере >= size, то мы получаем список необходимых хэндлеров (их может быть несколько) по заданному опкоду и передаем им size байт из буфера. Либо ожидаем получения остальных байт. Отдельного внимания заслуживает Update Packet. Он отличается от остальных пакетов более сложной структурой.

Расшифровывать нужно каждый пакет. Если какой-то пакет пропущен, то следующий (согласно алгоритму) будет расшифрован неправильно. На эти грабли я наступал с особой старательностью.

В чем суть моего сервера

Во-первых, я его писал на Python 3 (asyncio + SQLAlchemy). Точнее, на момент создания я еще не имел опыта с SQLAlchemy - я его обрел после, когда решил, что существующая у меня реализация работы с БД - ужасна во всех ее проявлениях (а еще попутно решил изучить новую технологию). И очередной раз переписал проект с нуля.

Во-вторых, подход к обработке данных тоже несколько отличался. Я предпочел отказаться от идеи глобального хранилища всех объектов (где каждый объект содержит также методы работы с ним) и создать специальные мэнеджеры (manager), для того, чтобы работать с каждым отдельным типом объектов: Item, Player, Unit и т.д. Т.е. если я хочу выполнить действие над Player, я использую PlayerManager, который выполняет нужное действие и удаляется из памяти. В первую очередь это сделано в угоду читабельности. К примеру, аналогичный класс в MANGOS лично для меня кажется громоздким (как и C++, несмотря на множество его преимуществ). Каждый отдельный класс объекта - это SQLAlchemy модель и он же - структура данных, хранящая текущее состояние. Таким образом, я использую такие объекты не только для взаимодействия с БД, но и для обмена структурированными данными между частями приложения.

В-третьих, я использую специальные классы - handler - совокупность которых я использую для обработки отдельно взятого опкода. Каждый хэндлер должен выполнять одну типовую задачу. И (опционально) он может вернуть респонс (опкод + данные - структура аналогична запросу), который затем будет отправлен клиенту.

В-четвертых, почти сразу я решил не делать еще один MANGOS движок, а вместо этого начал экспериментировать со свободой применения (что можно сделать помимо уже реализованного в других схожих проектах?). Начиналось все с идеи использования произвольного датасэта (данных, которые мы загружаем в базу). Короче, моя цель - скорее, творчество без границ, а не создание модифицированного (или blizzlike) сервера WoW.

В-пятых, я неоднократно переписывал код, чтобы сделать его читабельнее и, в принципе, у меня получилось сделать более-менее однородную архитектуру. Думаю, это - однозначный плюс для тех, кто будет читать исходный код. Для этих же целей при разработке упор был на ООП.

Что дальше

В процессе разработки сервера у меня возникла идея создания MMO RPG шаблонизатора, который предоставит формат описания игровых серверов (например, связка Login + World сервер, или шард-ориентированная архитектура, где каждая локация - отдельный сервер и т.д.), а так же серверов, имеющих отношение к игровым косвенно (например, web сервер, где будет сайт и форум). Именно для этого я создал фреймворк (но не совсем внятно описал его в своей предыдущей статье). И сейчас сервер переписываю на базе этого фреймворка.

Я уже наткнулся на ряд вещей, которые нужно будет отрефакторить, поэтому, к моменту завершения текущей цели я планирую выпустить ряд изменений в самом фреймворке. О чем я, возможно, напишу в будущем. Если к серии статей будет интерес.

P.S. продолжать развитие своего изначального проекта не планирую. Но, учитывая, что некоторые люди его используют в образовательных целях, то я рад, если кому-то еще он будет полезен.