How to run Python Microdot web API module, app.run() that is already an asyncio task, at the same time with other asyncio functions?

Question

I'm trying to run a Microdot app.run() that is asyncio with other functions that are also asyncio and I keep getting that the Microdot.start_server() was never awaited

In this case the other function is run_thermostat() that is an asyncio function that is running fine if I do not start at all the Microdot app.run()

I need some help on how to use Microdot app.run() to run along with other functions in a forever loop

from microdot_asyncio import Microdot
import asyncio
from asyncio import gather

class Thermostat:
    def __init__(self):
        self.current_temp = 0
        self.target_temp = 0
        ###
        
        self.loop = asyncio.get_event_loop()
    
    def call__async_main(self):
        self.loop.run_until_complete(self.__async_main())
        print(self.loop)
    
    async def __async_main(self):
    ####does not work | sys:1: RuntimeWarning: coroutine 'Microdot.start_server' was never awaited
        await gather(
            self.start_web_server(),
            self.run_thermostat(),
            )

    ##### does not work | sys:1: RuntimeWarning: coroutine 'Microdot.start_server' was never awaited
        # task1 = asyncio.create_task(self.run_thermostat())
        # task2 = asyncio.create_task(self.start_web_server())
        # # Iterate over the tasks and wait for them to complete one by one
        # for task in asyncio.as_completed([task1, task2]):
        #     await task
        
    ####### does not work | sys:1: RuntimeWarning: coroutine 'Microdot.start_server' was never awaited
        # task1 = asyncio.create_task(self.run_thermostat())
        # task2 = asyncio.create_task(self.start_web_server())
        # await task1
        # await task2
        # self.loop.create_task(task1)
        # self.loop.create_task(task2)

    #####other method | does not work | sys:1: RuntimeWarning: coroutine 'Microdot.start_server' was never awaited
        # await self.start_web_server()
        # await self.run_thermostat()

    #Function for starting Microdot web-server API
    def start_web_server(self):
        app.run()

    # Function to read the current temperature from a temperature sensor
    async def read_temperature(self):
        # read temperature from sensor 
        self.current_temp = 15  # ex: set the current temperature to 15 //simulate

    # Function to set the target temperature
    def set_target_temp(self, temp):
        self.target_temp = temp

    # Function to start or stop the heating system based on the current and target temperatures
    def control_heating(self):
        if self.current_temp < self.target_temp:
            
            # Turn on the heating system 
            print("Starting heating system")
        else:
            # Stop the heating system
            
            print("Stopping heating system")

# Function to run the thermostat control loop
    async def run_thermostat(self):
        while True:
            # Read the current temperature from the sensor
            await thermostat.read_temperature()

            # Control the heating system based on the current and target temperatures
            thermostat.control_heating()

            # Wait for a short time before checking the temperature again
            await asyncio.sleep(1)

#Instantiating Microdot app
app = Microdot()
# Set up a route to allow users to set the target temperature
@app.route('/set_target_temp')
def set_target_temp(request):
    # Get the target temperature from the request
    target_temp = int(request.args['temp'])

    # Set the target temperature on the thermostat
    thermostat.set_target_temp(target_temp)

    # Return a response to confirm the target temperature has been set
    return "Target temperature set to {}".format(target_temp)

# Create an instance of the Thermostat class
thermostat = Thermostat()

#===###
# Start the system
thermostat.call__async_main()

#===###

# Start the Microdot app
# asyncio.create_task(app.run())
# app.run()

# Start the thermostat control loop
# asyncio.create_task(run_thermostat())
################################################

Answer 1

A short example which works out of the box.

First copy and paste microdot.py from the official repo.

Than copy this file (microdot_async_example.py), change ssid and password and go! IP is printed in terminal and port is 5000. So you can reach the server with: for example: http://192.168.178.100:5000/

import network
import time
import uasyncio
from microdot import Microdot
from machine import Pin

# WIFI CONNECTION
ssid = 'SOMEWIFINAME'
password = 'SOMEWIFIPASSWORD'
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
time.sleep(1.5)
wlan.connect(ssid, password)
print('Connecting')
while wlan.isconnected() == False and wlan.status() < 0:
    print(f".", end = "")
    time.sleep(1)
print("Connected! IP Address = " + wlan.ifconfig()[0])

# CONFIGURE SERVER
app = Microdot()
@app.route('/')
def index(request):
    return 'Hello, You!'

async def start_server():
    app.run(port=5000)

# ANOTHER STUFF YOU WANNA DO WHILE SERVER IS RUNNING
async def button_monitor():
    # JUST READING VALUE FROM A BUTTON
    button_pin = Pin(2, Pin.IN, Pin.PULL_DOWN)
    while True:
        if button_pin.value() == 1:         
            print("Button pressed!")
        await uasyncio.sleep(1)

# ASYNCIO MAGIC
async def main():
    uasyncio.create_task(start_server())
    uasyncio.create_task(button_monitor())
    while True:
        await uasyncio.sleep(100)
        
uasyncio.run(main())

Answer 2

Never mind, I have found the solution, here it is:

    async def __async_main(self):
    ##### does work OK
        task1 = self.loop.create_task(self.run_thermostat())
        task2 = self.loop.create_task(self.start_web_server())
        # Iterate over the tasks and wait for them to complete one by one
        for task in asyncio.as_completed([task1, task2]):
            await task

and the Microdot should be started with app.server_start() and not by app.run() like this:

    #Function for starting Microdot web-server API
    async def start_web_server(self):
      await app.start_server(port=5000, debug=True)

Now, the Microdot web api runs along with other async tasks

In conclusion the main problem was starting the Microdot with app.run() , and also I had to add the task in the loop event like this:

task2 = self.loop.create_task(self.start_web_server())

before was:

task2 = asyncio.create_task(self.start_web_server())

and in this way it does not append into the loop, this loop:

self.loop = asyncio.get_event_loop()

Here is the complete working code:

from microdot_asyncio import Microdot
import asyncio

class Thermostat:
    def __init__(self):
        self.current_temp = 0
        self.target_temp = 0        
        self.loop = asyncio.get_event_loop()
    
    def call__async_main(self):
        self.loop.run_until_complete(self.__async_main())
        # print(self.loop)
    
    async def __async_main(self):
    ##### does work OK
        task1 = self.loop.create_task(self.run_thermostat())
        task2 = self.loop.create_task(self.start_web_server())
        # Iterate over the tasks and wait for them to complete one by one
        for task in asyncio.as_completed([task1, task2]):
            await task
        
    ####### does work OK
        # task1 = self.loop.create_task(self.run_thermostat())
        # task2 = self.loop.create_task(self.start_web_server())
        # await task1
        # await task2

    #Function for starting Microdot web-server API
    async def start_web_server(self):
      await app.start_server(port=5000, debug=True)

    # Function to read the current temperature from a temperature sensor
    async def read_temperature(self):
        # read temperature from sensor 
        self.current_temp = 15  # ex: set the current temperature to 15 //simulate

    # Function to set the target temperature
    def set_target_temp(self, temp):
        self.target_temp = temp

    # Function to start or stop the heating system based on the current and target temperatures
    def control_heating(self):
        if self.current_temp < self.target_temp:
            
            # Turn on the heating system 
            print("Starting heating system")
        else:
            # Stop the heating system
            
            print("Stopping heating system")

# Function to run the thermostat control loop
    async def run_thermostat(self):
        while True:
            # Read the current temperature from the sensor
            await thermostat.read_temperature()

            # Control the heating system based on the current and target temperatures
            thermostat.control_heating()

            # Wait for a short time before checking the temperature again
            await asyncio.sleep(1)

#Instantiating Microdot app
app = Microdot()
# Set up a route to allow users to set the target temperature
@app.route('/set_target_temp')
def set_target_temp(request):
    # Get the target temperature from the request
    target_temp = int(request.args['temp'])

    # Set the target temperature on the thermostat
    thermostat.set_target_temp(target_temp)

    # Return a response to confirm the target temperature has been set
    return "Target temperature set to {}".format(target_temp)

# Create an instance of the Thermostat class
thermostat = Thermostat()

#===###
# Start the system
thermostat.call__async_main()

#===###

you can navigate to your ESP32 IP or localhost and enter this link to change the target_temp variabile:

http://localhost:5000/set_target_temp?temp=21

... you should then see in the VScode console : Starting heating system .... if you enter another int query parameter for the temp like this:

http://localhost:5000/set_target_temp?temp=12

...you shoud see in the VScode console : Stopping heating system

CONCLUSION:

• the code works
• the run_thermostat() function runs along (asyncio) with app = Microdot() and does what is supposed to do
• the Microdot decorator @app.route('/set_target_temp') works OK