Всем привет, въезд в новый проект и работу отожрал все силы и на блог не оставалось времени. Буду потихоньку возвращаться в неторопливый режим) Сегодня перевод интересной заметки, посвященной конечным автоматам и корутинам.
_________________________________________________________________________________________
Конечные автоматы (КА) это математическая модель вычислений, представляющая последовательностную логику[1] КА состоит из конечного набора состояний, функций перехода, входящих алфавитов, стартового состояния и конечных состояний. В computer science, КА используются при проектировании Компиляторов, Лингвистических расчетов, Геймдизайне, Протокольных процедурах (tcp/ip например), Событийной разработке (Event-driven programming), и т.д.
Чтобы понять что же такое КА, построим для примера модель светофора (рисунок ниже). Зеленый это начальное/ стартовое состояние, которое может быть переведено в желтый и затем в красный, а затем опять в зеленый.
В каждый момент времени КА должен быть в одном из своих состояний. В примере выше автомат имеет 3 состояния: зеленый, желтый, красный. Для каждого состояния свои правила перехода, определяющие воспроизводимую логику.
Реализация КА имеет решающее значение для решения некоторых наиболее интересных проблем в Computer Science. В этой статье мы рассмотрим вариант моделирования КА с помощью сопрограмм(корутин) в Python.
Сопрограммы/корутины в Python
Перед тем как рассматривать реализацию, давайте немного рассмотрим устройство генераторов и сопрограмм, их место в общей схеме.
Генераторы
Генераторы - функции, к исполнению которых вы можете возвращаться. Они генерируют значение, пока их продолжают запрашивать с помощью функции next(). Когда возможные значения заканчиваются, генератор инициирует исключение StopIteration.
def fib():
a, b = 0, 1
while True:
yield a
a, b = b, a+b
Основная магия начинается, когда функция доходит до yield. Исполнение генератора в этот момент приостанавливается, он отдает значение вызывающему и вызывающий продолжает исполнение. Основная программа возвращается к генератору только в момент, когда необходимо следующее значение. Как только в генератор приходит запрос, он начинает с момента остановки, т.е. с прошлого yield.
>>> fgen = fib()
>>> [next(fgen) for _ in range(10)]
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
Использование генератора для расчета чисел Фибоначчи позволяет эффективно распоряжаться памятью, т.к. нет необходимости держать в ней весь объем рассчитанных чисел, если мы их не собираем в список. Мы можем запросить любое количество значений и генератор будет выдавать их одно за другим.
Корутины
Сопрограммы как и генераторы позволяют возвращаться к исполнению функции, но вместо генерирования значений они их принимают. Это очень похоже на работу с генераторами и опять вся магия кроется в команде yield. Когда сопрограмма остановлена, мы можем отправить ей значение с помощью функции send, как в примере ниже:
def grep(substr):
while True:
line = yield
if substr in line:
print(f"found {substr}")
В примере мы реализовали небольшую grep функцию, которая ищет заданную подстроку в потоке текста. Когда сопрограмма находится на паузе, мы используем функцию send, чтобы отправить текст и она продолжает выполнение. Когда функция снова достигает команды yield, сопрограмма ставится на паузу и ожидает получения нового значения.
Важно заметить, что это не поток, который продолжает работу и потребление ресурсов ЦПУ. Это просто функция, которая прерывает свое выполнение, сохраняя состояние до следующего вызова. Контроль на это время передается вызывающему.
Перед тем, как послать значение в сопрограмму, нам нужно "запустить" её, чтобы поток достиг команды yield и остановился, ожидая значение.
>>> g = grep("users/created")
>>> next(g) # priming the generator
>>>
>>> g.send("users/get api took 1 ms.")
>>> g.send("users/created api took 3 ms.")
found users/created
>>> g.send("users/get api took 1 ms.")
>>> g.send("users/created api took 4 ms.")
found users/created
>>> g.send("users/get api took 1 ms.")
В примере выше мы продолжаем посылать текст в сопрограмму и получать результаты поиска подстроки. Эта способность останавливаться и взаимодействовать с неограниченным потоком входных значений очень поможет нам при построении КА.
Строим конечный автомат
При построении конечного автомата самое главное - определить модель состояний и функции перехода. В нашем случае состояния могут быть смоделированы как Python-сопрограммы, находящихся в бесконечном ожидании входящих данных, а функция перехода может быть реализована как набор if/elif операторов. В более сложных системах может быть прописана отдельная функция, учитывающая все особенности.
Чтобы немного погрузиться в детали мы построим конечный автомат для регулярного выражения ab*c. Если входящая строка будет соответствовать выражению - автомат должен перейти в конечное состояние.
State
From the FSM above we model the state q2 as
Определим состояние q2 с рисунка выше:
def _create_q2():
while True:
# Wait till the input is received.
# once received store the input in `char`
char = yield
# depending on what we received as the input
# change the current state of the fsm
if char == 'b':
# on receiving `b` the state moves to `q2`
current_state = q2
elif char == 'c':
# on receiving `c` the state moves to `q3`
current_state = q3
else:
# on receiving any other input, break the loop
# so that next time when someone sends any input to
# the coroutine it raises StopIteration
break
Сопрограмма запускается в бесконечном ожидании входящего значения. После получения значения b состояние сменяется на q2 и на q3 после получения значения с.
Класс КА
Чтобы поддерживать уровень инкапсуляции, определим класс FSM, который содержит все состояния и обслуживает текущее. Так же в нем есть метод, который пересылает входящее значение в текущее состояние.
В зависимости от задач FSM может так же содержать функцию, которая отвечает на какой-либо вопрос вроде "соответствует ли строка регулярному выражению" или "делится ли это число на 3".
Класс FSM для регулярного выражения ab*c может выглядеть так:
class FSM:
def __init__(self):
# initializing states
self.start = self._create_start()
self.q1 = self._create_q1()
self.q2 = self._create_q2()
self.q3 = self._create_q3()
# setting current state of the system
self.current_state = self.start
# stopped flag to denote that iteration is stopped due to bad
# input against which transition was not defined.
self.stopped = False
def send(self, char):
"""The function sends the curretn input to the current state
It captures the StopIteration exception and marks the stopped flag.
"""
try:
self.current_state.send(char)
except StopIteration:
self.stopped = True
def does_match(self):
"""The function at any point in time returns if till the current input
the string matches the given regular expression.
It does so by comparing the current state with the end state `q3`.
It also checks for `stopped` flag which sees that due to bad input the iteration of FSM had to be stopped.
"""
if self.stopped:
return False
return self.current_state == self.q3
...
@prime
def _create_q2(self):
while True:
# Wait till the input is received.
# once received store the input in `char`
char = yield
# depending on what we received as the input
# change the current state of the fsm
if char == 'b':
# on receiving `b` the state moves to `q2`
self.current_state = self.q2
elif char == 'c':
# on receiving `c` the state moves to `q3`
self.current_state = self.q3
else:
# on receiving any other input, break the loop
# so that next time when someone sends any input to
# the coroutine it raises StopIteration
break
...
Похожим с _create_q2 образом определяем функции для остальных состояний: start, q1, q3. Полный код можно найти на гитхабе автора
Функция запуска
Определим функцию grep_regex, которая будет проверять входной текст на соответствие выражению ab*c. Алгоритм следующий: создаем внутри функции экземпляр КА и пропускаем через него поток символов. Как только последовательность символов кончилась, мы вызываем метод does_match.
def grep_regex(text):
evaluator = FSM()
for ch in text:
evaluator.send(ch)
return evaluator.does_match()
>>> grep_regex("abc")
True
>>> grep_regex("aba")
False
Другие варианты КА
Если наша гипотеза верна, то будет легко использовать конечный автомат для решения других задач, кроме соответствия регулярным выражениям. Разберем ещё пару примеров:
Делимость на 3
Схема КА, который определяет является ли поток цифр числом делящимся на 3.
Вариант реализации состояния q1:
def _create_q1(self):
while True:
digit = yield
if digit in [0, 3, 6, 9]:
self.current_state = self.q1
elif digit in [1, 4, 7]:
self.current_state = self.q2
elif digit in [2, 5, 8]:
self.current_state = self.q0
Можно заметить схожие моменты между реализацией корутины и функции смены состояний. Полный код можно посмотреть на гитхабе
Валидатор SQL запросов
Ещё один вариант задачи - построить валидатор SQL запросов. Полноценный валидатор будет слишком большим, поэтому предположим, что мы обрабатываем запросы подобного формата:
SELECT * from TABLE_NAME;
SELECT column, [...columns] from TABLE_NAME;
Вариант определения состояния explicit_cols:
def _create_explicit_cols(self):
while True:
token = yield
if token == 'from':
self.current_state = self.from_clause
elif token == ',':
self.current_state = self.more_cols
else:
break
полный код на гитхабе
Заключение
Возможно построение конечных автоматов это не самый эффективный путь к решения задач, но один из самых интуитивно понятных. Переходы состояний хорошо переносятся на if/elif команды или в отдельные "функции решений". Каждое состояние может быть смоделировано как независимая корутина, но при этом все они будут решать задачу в последовательно передавая исполнение из одной в другую.
__________________________________________________________________________________________
Фух, всё руки не доходили до этого перевода, лежал сиротливо в черновиках. От себя могу добавить, что конечные автоматы - это очень интересная штука как минимум на уровне концепции. Выглядит подходящим инструментом для имитации искуственного интеллекта в играх, например.
[1] последовательностная логика
[оригинал] тык