异步 python 脚本刚刚在控制台中返回（在 IDE / 调试器中工作）

Question

该脚本运行在 rasperry 零上的 thonny 中运行良好，并完成其测量和与 mqtt 服务器交互的工作。这是一个在后台运行s 的脚本，我将它添加到 /etc/rc.local 行 fi 和 exit 0 之间： python3 /home/pi/pisensor/scripts/dht22_mqtt.py & 重新启动后，mqtt 服务器上的预期消息没有出现，所以我在控制台上尝试运行ning 脚本。我在目录中用 python3 dht22_mqtt.py 调用它，几秒钟后它看起来像是完成了（控制台中的新行），但什么也没做（没有消息到 mqtt 服务器，也没有打印消息).没有显示库丢失或出现问题的错误。运行从控制台使用这样的异步脚本会不会有问题？

当从控制台运行时是否可以调试脚本？我试了python3的-d参数，结果一样。我用 dht 和 paho-mqtt 库尝试了 pip3，但它已经安装了。

您可以在这里查看脚本：

from time import sleep
import asyncio
from paho.mqtt import subscribe, publish
from random import randint
from typing import Awaitable, Any


async def run_sequence(*functions: Awaitable[Any]) -> None:
    for function in functions:
        await function


async def run_parallel(*functions: Awaitable[Any]) -> None:
    await asyncio.gather(*functions)


class IOTSensorDHT22:

    def __init__(self,
                 mqtt_server="localhost",
                 mqtt_port=1883,
                 sensor_pin=4,
                 sensor_values_topic_prefix="/sensor/raw/bedroom/",
                 sensor_interval=60,
                 sensor_interval_topic="/setting/sensor/inside/power/interval"):
        self.messages = None
        self.values = None
        self.mqtt_server = mqtt_server
        self.mqtt_port = mqtt_port
        self.sensor_pin = sensor_pin
        self.sensor_values_topic_prefix = sensor_values_topic_prefix
        self.sensor_interval = sensor_interval
        self.sensor_interval_topic = sensor_interval_topic
        self.mqtt_client_id = f"mqtt_client_{randint(0, 1000)}"

    async def query_interval(self):
        new_interval = int(subscribe.simple(self.sensor_interval_topic, hostname=self.mqtt_server).payload)
        if new_interval != self.sensor_interval:
            print(
                f"new interval {new_interval} - old: {self.sensor_interval} - assigning new value: {new_interval != self.sensor_interval}")
            self.sensor_interval = new_interval

    async def publish_values(self):
        print(f"publishing messages: {self.messages}")
        publish.multiple(self.messages, hostname=self.mqtt_server)

    async def measure(self):
        self.values = Adafruit_DHT.read_retry(Adafruit_DHT.DHT22, self.sensor_pin)

    async def build_messages(self):
        messages = (self.build_message("hum", self.values[0]), self.build_message("temp", self.values[1]))
        print(messages)
        self.messages = messages

    def build_message(self, topic_appendix, value):
        return self.sensor_values_topic_prefix + topic_appendix, round(value, 1)

    def sleep(self):
        print(f"going to sleep for {self.sensor_interval}")
        sleep(self.sensor_interval)

    async def run(self):
        while True:
            await run_parallel(
                run_sequence(self.measure(), self.build_messages(), self.publish_values()),
                self.query_interval()
            )
            self.sleep()


if __name__ is "__main__":
    asyncio.run(
        IOTSensorDHT22(
            mqtt_server="192.168.188.37",  #
            sensor_pin=12,
            sensor_values_topic_prefix="/sensor/raw/bedroom/",
            sensor_interval=60,
            sensor_interval_topic="/setting/sensor/inside/power/interval/").run())

编辑：感谢编辑帮助:)

更新 - 已解决：

我写了if __name__ is "__main__":

本来应该是

if __name__ == "__main__":

这确实是我的错误，但它误导了我，它在 IDE (Thonny) 中运行良好，但在运行从控制台运行时什么也没做。

Answer 1

您可以使用 ipdb 在控制台中进行调试。

pip install ipdb

然后将其放入您的代码中您要调试的位置：

import ipdb
ipdb.set_trace()

然后您就可以单步执行您的代码，就像使用来自 IDE 的调试器一样。有关如何使用 ipdb 的更多说明，请参阅 this tutorial.

异步 python 脚本刚刚在控制台中返回（在 IDE / 调试器中工作）

Async python script just returning in console (works in IDE / debugger)

python

console

python-asyncio