Как я могу гарантировать, что телеметрия в реальном времени будет публиковаться брокеру MQTT каждую секунду во время зап ⇐ Python

[Anonymous]

Я создаю своего рода симулятор UTM, используя Python. С одной стороны, у меня есть flask-сервер, который должен отправлять план полета дрону (встроенный вdronekit иdronekit-sitl), который выполняет план полета.
Рабочий процесс следующий:

• Сервер flask отправляет план полета вdronekit иdronekit-sitl через MQTT-брокер в теме под названием «flight/plan».
• Dronekit-sitl получает местоположение дома из плана полета и настраивает транспортное средство. Затем он публикует строку подключения к теме «полет/дрон».
• Менеджер полета в дроновом наборе получает строку подключения и путевые точки из тем и создает транспортное средство, которое выполняет полет в соответствии с к плану полета.

До этого момента всё хорошо. Проблема возникает позже:
4. В качестве последней задачи дрон должен разработать телеметрическое сообщение и каждую секунду публиковать его в теме mqtt «полет/телеметрия».
Цикл указывает, что сообщение генерируется и предположительно публикуется, но реальность такова, что вместо того, чтобы публиковать информацию каждую секунду, сообщение не публикуется до тех пор, пока не завершится полет, то есть время, в течение которого сообщения публикуются массово.
Я пробовал загружать потоки процессы, использующие разные QoS в mqtt и все такое, но я, похоже, не нашел проблемы. Журналы Mosquitto не показывают никаких проблем, они просто показывают, что сообщения публикуются и получаются другими клиентами массово, а не каждую секунду.
Эта штука работает локально, хотя симулятор часть работает на виртуальной машине в Linux по соображениям совместимости.
Настройка части телеметрии выглядит следующим образом. Я опустил все, что не связано с проблемой.
mqtt_manager.py
class MQTTManager:
def __init__(self, broker_host, broker_port):
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.client.on_disconnect = self.on_disconnect
self.client.connect(broker_host, broker_port, 60)
self.client.loop_start()
self.connection_string = None
self.flight_plan = None

def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print(f"Conectado al broker MQTT con código: {rc}")
client.subscribe("flight/plans")
client.subscribe("flight/drone",qos=2)
else:
print(f"Error al conectar: {rc}")

def on_message(self, client, userdata, msg):
print(f"Mensaje recibido: {msg.topic} {msg.payload}")
if msg.topic == "flight/plans":
self.flight_plan = json.loads(msg.payload.decode())
if self.connection_string:
self.start_flight()
elif msg.topic == "flight/drone":
self.connection_string = msg.payload.decode()
if self.flight_plan:
self.start_flight()

def start_flight(self):
process_flight_plan(self.flight_plan, self.connection_string, self)

def on_disconnect(self, client, userdata, rc):
print(f"Desconectado con código: {rc}")
if rc != 0:
print("Intentando reconectar...")
try:
client.reconnect()
except Exception as e:
print(f"Error al reconectar: {e}")

def publish_telemetry(self, topic, telemetry_data):
try:
self.client.publish(topic, json.dumps(telemetry_data), qos=2)
print(f"Telemetría enviada: {telemetry_data}")
except Exception as e:
print(f"Error al publicar telemetría: {e}")

mqtt_manager = MQTTManager('localhost', 1883)

flight_manager.py
def send_telemetry_periodically(mqtt_manager):
while True:
send_telemetry(mqtt_manager)
time.sleep(1)

def process_flight_plan(flight_plan, connection_string, mqtt_manager):
vehicle_manager = VehicleManager.get_instance()

waypoints = flight_plan['waypoints']
if not waypoints[0]:
print("No hay waypoints en el plan de vuelo recibido.")
return

try:
vehicle_manager.connect_vehicle(connection_string)

print("Vehículo armado")

print("Despegando...")
vehicle_manager.get_vehicle().simple_takeoff(10)

# Hilo para enviar telemetría
telemetry_thread = threading.Thread(target=send_telemetry_periodically, args=(mqtt_manager,))
telemetry_thread.daemon = True
telemetry_thread.start()

for point in flight_plan['waypoints']:
location = LocationGlobalRelative(point['lat'], point['lon'], point['alt'])
print(f"Navegando a: {location}")
move_thread = threading.Thread(target=vehicle_manager.move_vehicle_to_location, args=(location,))
move_thread.start()
move_thread.join()

print("Aterrizando...")

except Exception as e:
print(f"Error en process_flight_plan:{str(e)}")


telemetry_manager.py
from modules.vehicle_manager import VehicleManager

def send_telemetry(mqtt_manager):
vehicle_manager = VehicleManager.get_instance()
vehicle = vehicle_manager.get_vehicle()
if vehicle and vehicle.mode.name == 'GUIDED':
telemetry = vehicle_manager.capture_vehicle_state()
print("Mensaje de telemetría: ", telemetry)
mqtt_manager.publish_telemetry("flight/telemetry",telemetry)

vehicle.py
class VehicleManager:
_instance = None
vehicle = None
connection_string = None

@staticmethod
def get_instance():
if VehicleManager._instance is None:
VehicleManager()
return VehicleManager._instance

def __init__(self):
if VehicleManager._instance is not None:
raise Exception("This class is a singleton!")
else:
VehicleManager._instance = self

def get_vehicle(self):
return self.vehicle

def capture_vehicle_state(self):
if self.vehicle:
return {
'altitude': self.vehicle.location.global_relative_frame.alt,
'latitude': self.vehicle.location.global_relative_frame.lat,
'longitude': self.vehicle.location.global_relative_frame.lon,
'armed': self.vehicle.armed,
'mode': self.vehicle.mode.name
}
else:
return None



Подробнее здесь: https://stackoverflow.com/questions/786 ... er-every-s