Как вызывать функции Flask каждые несколько секунд и передавать между ними переменные

Question

Начал разработку одного проекта на Raspberry PI, который через сериал порты передавал бы ардуинке json строку по нажатию на кнопку, но параллельно с этим все время мониторил бы сам порт и ждал получения json от ардуинки. И если приходила бы строка, то отправляла нужные переменные в функцию для кнопки. Попытался как то сам это реализовать через ajax jquery, но почему то не работало.

Сам я месяц назад начал изучать python и сразу вмести с ним flask и может еще чего-то не понимаю. Поэтому может кто-нибудь подсказать что написал не так?

Первая функция для кнопки:

...
@app.route('/que', methods=['POST'])
def butt(action = None, skaz = None):
    import serial, time, cgi, cgitb, pygame
    ser = serial.Serial(port='/dev/ttyUSB0',baudrate=115200, timeout = 100)
    ser.flush()
    comand = "Listen"
    phrases = [1,2,3,4,5,6,7]
    json_cumand = {"comand":comand,"phrases": phrases}
    print(type(json_cumand))
    json_string = json.dumps(json_cumand)
    print(json_string)
    ser.write(str.encode(json_string))
    if skaz == "Da":
        print("Peredal")
    else:
        action = request.form['action']
        skaz = "Net"
...

Вторая функция для мониторинга:

@app.route("/myStatus")
def getStatus():
    import serial, time, cgi, cgitb, pygame
    from jsonschema import validate, exceptions
    ser = serial.Serial(port='/dev/ttyUSB0',baudrate=115200, timeout = 100)
    ser.flush()
    success_validate = False
    json_result = "1"

    schema = {
    "type": "object",
    "properties": {
        "cumand": {
            "type": "number"
            },
        },
    }

    while not success_validate:
        success_validate = True
        json_result = ser.readline()[:-2] 
        try:
            json_str2 = str(json_result)[2:-1]
            print(json.loads(json_str2))
            success_validate = not validate(json.loads(json_str2),schema=schema)
        except: 
            print("GJ")
            success_validate = False

    if success_validate:
        data = '%s' % (str(json_result)[2:-1])
        d = json.loads(data)
        butt("BT1", "Da")
    return d

скрипт на сайте:

    <script>
        $(document).ready(function() {
            setInterval("ajaxd()",10000);
        });
        
        function ajaxd() { 
          load("/myStatus");
        
        }
    </script>

Answer 1

Тебе нужно двигаться в сторону многопоточных приложений. Вам будет нужно создать отдельный поток для метода getStatus(). Основной поток приложения будет обрабатывать нажатия кнопки, а второй, поток созданный вами, ожидать строку от ардуинку.

Первое нужно вынести код с getStatus() в отдельный файл:

Status_From_Arduino.py

import serial, time, cgi, cgitb, pygame
from jsonschema import validate, exceptions
ser = serial.Serial(port='/dev/ttyUSB0',baudrate=115200, timeout = 100)

# Импорт библиотеки для роботы с потоками
from threading import Thread 

class Status():

    def __init__(self) -> None:
        pass
    
    def start_monitor(self):
        # Создаём нвоый поток
        th = Thread(target=self.get_status, args=(1, ), name='monitor')
        
        # Если поток не запущеный, то запускаем
        if not th.is_alive():
            th.start()
    
    def get_status(self):
        ser.flush()
        success_validate = False
        json_result = "1"

        schema = {
        "type": "object",
        "properties": {
            "cumand": {
                "type": "number"
                },
            },
        }

        while not success_validate:
            success_validate = True
            json_result = ser.readline()[:-2] 
            try:
                json_str2 = str(json_result)[2:-1]
                print(json.loads(json_str2))
                success_validate = not validate(json.loads(json_str2),schema=schema)
            except: 
                print("GJ")
                success_validate = False

        if success_validate:
            data = '%s' % (str(json_result)[2:-1])
            d = json.loads(data)
            butt("BT1", "Da")

Потом импортируйте этот класс и в методе getStatus добавьте следующее:

from Status_From_Arduino import Status

@app.route("/myStatus")
def getStatus():
    status = Status()
    status.start_monitor()

Примерно код должен быть таким, вам остаётся добавить логику для обработки данных. Код не запускал, но думаю что суть ясна. Больше инфы здесь:

Ссылка-1

Ссылка-2

БЛОГ НА HUSL

Как вызывать функции Flask каждые несколько секунд и передавать между ними переменные

Ответы (1 шт):