event.py: "nicegui_globals.loop.is_running" raises Attribute Error: 'NoneType' object has no attribute 'is_running'

[AndrewSheetMetal]

Hi,
I receive an error, that makes no sense for me for several reasons.

My code is the following:

def handle_velocity(velocity: Velocity):
    print(f"Received new velocity: {velocity}")

mqtt_odometer_velocity_listener.UPDATED.register(handle_velocity)

Whenever the Event is emitted via self.UPDATED.emit(velocity), I get the following error:

Traceback (most recent call last):
File "/rosys/rosys/event.py", line 77, in emit
if nicegui_globals.loop.is_running():
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'is_running'
ERROR:rosys.event:could not emit listener=EventListener(callback=<function handle_velocity at 0x7f4df8fb6f20>, filepath='/app/main.py', line=297)

The handle_velocity function is invoked, so everything works for me, but the error log messages bombards the console.

I looked into the event.py:

    def emit(self, *args) -> None:
        """Fires event without waiting for the result."""
        for listener in self.listeners:
            try:
                result = invoke(listener.callback, *args)
                if isinstance(result, Awaitable):
                    if nicegui_globals.loop.is_running():
                        name = f'{listener.filepath}:{listener.line}'
                        tasks.append(background_tasks.create(result, name=name))
                    else:
                        startup_coroutines.append(result)
            except Exception:
                log.exception(f'could not emit {listener=}')

we can see, that the condition, that raises the Attribute Error is only executed, if the result is an instance of Awaitable.
But my handle_velocity function does not return an Awaitable.

So, I have two questions:

1. Why is nicegui_globals.loop a 'NoneType' object? - and how can I change this?
2. Why is if isinstance(result, Awaitable): true although I do not return anything in handle_velocity?

pip show return 0.9.0 for rosys and 1.3.2 for nicegui

It would be very nice, if you could help me, because I really don't understand the reason for the error.

Best regards
Andreas

[rodja]

It could be that your mqtt is not running on the same event loop. RoSys, NiceGUI and the underlying FastAPI are async frameworks. That means, they run an event loop and no io-bound or cpu-bound tasks should be directly executed on the main thread but rather be "awaited". See https://fastapi.tiangolo.com/async/ for a good in-depth explanation.

You might want to try asyncio-mqtt, aiomqtt or gmqtt.

[AndrewSheetMetal]

Thanks for the fast response!

On the README.md of aiomqtt (asyncio-mqtt), they recommend fastapi-mqtt for an integration into a FastAPI Application.

I tried it out and it works as expected.

Here is an example.
The interesting part, that differs from the Guide of FastMqtt is, that I took the FastAPI App of NiceGUI with
from nicegui.globals import app and mqtt.init_app(app).

Here the complete example:

import json
import uuid
import gmqtt.mqtt

import numpy as np
import rosys

from fastapi_mqtt import FastMQTT, MQTTConfig
from nicegui.globals import app

class MqttListener:
    def __init__(
            self,
            ip_address:str, 
            port: int,
            topic: str,
        ) -> None:

        self.UPDATED = rosys.event.Event()

        self._ip_address = ip_address
        self._port = port
        self._topic = topic

        mqtt_config = MQTTConfig(host=ip_address, port=port, keepalive=60)

        mqtt = FastMQTT(
            config=mqtt_config,
            client_id=self.__class__.__name__
        )

        self.mqtt = mqtt
        mqtt.init_app(app) #use the same fastapi app as the nicegui app
                
        @mqtt.on_connect()
        def connect(client: gmqtt.client.Client, flags, rc, properties):
            mqtt.client.subscribe(self._topic)
            print(f"{self.__class__.__name__} connected to mqtt ")

        @mqtt.on_message()
        async def message(client: gmqtt.client.Client, topic, payload, qos, properties):
            if topic == self._topic:
                self.UPDATED.emit(payload.decode("utf-8"))

        @mqtt.on_disconnect()
        def disconnect(client: gmqtt.client.Client, packet, exc=None):
            print(f"{self.__class__.__name__} disconnected from mqtt")
       

[rodja]

Looks great! I'll close the issue then.