Рубрики
Без рубрики

Построение конечных автоматов с помощью сопрограмм Python

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

Автор оригинала: Arpit Bhayani.

Конечный автомат-это математическая модель вычислений, которая моделирует последовательную логику. FSM состоит из конечного числа состояний, функций перехода, входных алфавитов, начального состояния и конечного состояния(состояний). В области компьютерных наук FSMS используются при проектировании Компиляторов, обработке лингвистики, пошаговых рабочих процессах, разработке игр, процедурах протоколов (таких как TCP/IP), программировании на основе событий, разговорном ИИ и многом другом.

Чтобы понять, что такое конечная машина, мы рассмотрим Сигнал светофора. Конечный автомат для сигнала светофора разработан и представлен ниже. Зеленый – это начальное/начальное состояние, которое при получении триггера переходит в Желтый , который, в свою очередь, при получении триггера переходит в Красный . Красный затем возвращается к Зеленому , и цикл продолжается.

сигнал светофора fsm

FSM должен находиться точно в одном из конечных состояний в любой данный момент времени, а затем в ответ на вход, который он получает, машина переходит в другое состояние. В приведенном выше примере сигнал светофора находится точно в одном из 3 состояний – Зеленый , Желтый или Красный . Правила перехода определены для каждого состояния, которое определяет, какая последовательная логика будет воспроизводиться при вводе.

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

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

Генераторы

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

def fib():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a+b

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

>>> fgen = fib()
>>> [next(fgen) for _ in range(10)]
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

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

Сопрограммы

Сопрограммы, как и генераторы, являются возобновляемыми функциями, но вместо генерации значений они потребляют значения на лету. Работа его очень похожа на генератор, и опять же оператор yield – это то, где происходит волшебство. Когда сопрограмма приостанавливается в операторе yield , мы можем отправить значение с помощью функции send , и значение может быть использовано с помощью оператора присваивания = on yield , как показано ниже

def grep(substr):
    while True:
        line = yield
        if substr in line:
            print(f"found {substr}")

В приведенном выше примере мы написали простую утилиту grep , которая проверяет наличие подстроки в заданном потоке текста. Когда сопрограмма grep приостанавливается в операторе yield , используя функцию send , мы отправляем ей текст, и на него будет ссылаться переменная line . Затем сопрограмма продолжает выполнение, чтобы проверить, находится ли substr в строке или нет. Как только поток снова достигает оператора yield , сопрограмма приостанавливается и ждет, пока вызывающий объект отправит ему новое значение.

Обратите внимание, что это не поток, который продолжает работать и загружает процессор. Это просто функция, выполнение которой приостанавливается в операторе 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.")

В приведенных выше вызовах функций мы видим, как мы могли бы продолжать отправлять текст в сопрограмму, и она продолжает выплевывать, если она нашла заданную подстроку users/created в тексте. Эта способность сопрограммы приостанавливать выполнение и принимать ввод на лету помогает нам моделировать FSM очень интуитивно понятным способом.

При построении FSMS самое важное-это то, как мы решаем моделировать и реализовывать состояния и функции перехода. Состояния могут быть смоделированы как сопрограммы Python, которые запускают бесконечный цикл, в котором они принимают входные данные, решают переход и обновляют текущее состояние FSM. Функция перехода может быть такой же простой, как набор операторов if и elif , а в более сложной системе это может быть функция принятия решений.

Чтобы погрузиться в детали низкого уровня, мы создаем FSM для регулярного выражения ab*c , что означает, что если данная строка соответствует регулярному выражению, то машина должна завершиться в конечном состоянии, только тогда мы говорим, что строка соответствует регулярному выражению.

fsm для ab*c

Государство

Из приведенного выше FSM мы моделируем состояние 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

Сопрограмма работает как бесконечный цикл, в котором она ожидает входного токена в операторе yield . При получении входных данных, скажем, b он изменяет текущее состояние FSM на q2 , а при получении c изменяет состояние на q3 , и это именно то, что мы видим на диаграмме FSM.

Класс FSM

Чтобы сохранить инкапсуляцию, мы определим класс для FSM, который содержит все состояния и поддерживает текущее состояние машины. Он также будет иметь метод под названием send , который перенаправляет полученные входные данные в текущее состояние. Текущее состояние после получения этого ввода принимает решение и обновляет current_state 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 . Вы можете найти полную модель FSM по адресу arpitbbhayani/fsm/regex-1

Функция драйвера

Мотивом этой постановки задачи является определение функции с именем grep_regex , которая проверяет данный текст на соответствие регулярному выражению ab*c . Функция внутренне создаст экземпляр FSM и передаст ему поток символов. Как только все символы исчерпаны, мы вызываем функцию does_match в FSM, которая подсказывает, соответствует ли данный текст регулярному выражению ab*c или нет.

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

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

Мы видели, насколько интуитивно понятно создавать FSMS регулярных выражений с использованием сопрограмм Python, но если наша гипотеза верна, все должно быть одинаково интуитивно понятным, когда мы реализуем FSMs для других случаев использования, и здесь мы рассмотрим два примера и посмотрим, как состояние реализуется в каждом из них.

Делимость на 3

Здесь мы строим FSM, который сообщает, делится ли данный поток цифр числа на 3 или нет. Государственная машина выглядит так, как показано ниже.

div3

Мы можем реализовать состояние 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

Мы видим сходство между реализацией сопрограммы и функцией перехода для состояния. Всю реализацию этого FSM можно найти по адресу arpitbbhayani/fsm/делимость на 3 .

Валидатор SQL-запросов

Здесь мы создаем FSM для валидатора SQL-запросов, который для данного SQL-запроса сообщает, является ли он допустимым SQL-запросом или нет. FSM для валидатора, который охватывает все SQL-запросы, будет огромным, поэтому мы просто имеем дело с его подмножеством, в котором мы поддерживаем следующие SQL-запросы

SELECT * from TABLE_NAME;
SELECT column, [...columns] from TABLE_NAME;
fsm для проверки запросов sql

Мы можем реализовать состояние 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

Опять же, сопрограмма, с помощью которой реализуется состояние, очень похожа на функцию перехода состояния, сохраняя вещи интуитивно понятными. Всю реализацию этого FSM можно найти по адресу arpitbbhayani/fsm/sql-query-validator .

Хотя это может быть не самый эффективный способ реализации и построения FSM, но это действительно самый интуитивно понятный способ. Ребра и переходы состояний хорошо переводятся в операторы if и elif или функции принятия решений, в то время как каждое состояние моделируется как независимая сопрограмма, и мы все еще делаем вещи последовательным образом. Все исполнение похоже на эстафету, в которой эстафетная палочка передается от одной сопрограммы к другой.

Другие статьи, которые вам могут понравиться:

Эта статья была первоначально опубликована на моем блог – Создание конечных автоматов с сопрограммами Python .

Если вам понравилось то, что вы прочитали, подпишитесь на мою рассылку, получите сообщение прямо в свой почтовый ящик и дайте мне знать @arpit_bhayani .

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